Conversation Engine

The conversation_engine service is the LIS 3 text-chat WebSocket service. It exposes a single WebSocket endpoint that drives one conversational turn per request through guard, persistence, optional image analysis, memory, RAG, prompt assembly, LLM streaming, and TTS. External dependencies are reached through hexagonal ports and adapters, and conversation history is written to Postgres.

Purpose & port

The service binds to port 8050 by default (EngineConfig.port, env CHAT_ENGINE_PORT, legacy alias LIVE_PORT). The default host is localhost. A FastAPI app built by container.build_live_app accepts each socket and hands the connection to a ConnectionHandler.

There are no HTTP chat routes. The only chat surface is one WebSocket; the two HTTP routes are operational.

Method	Path	Purpose
WS	/message_pipeline	The single live-chat WebSocket. Registered with no prefix, so the mounted path is ws://<host>:8050/message_pipeline.
GET	/health	Aggregated readiness (tags=["ops"]).
GET	/status	Live-session snapshot (tags=["ops"]).

/health probes only memory (MEMORY_URL/health), rag (RAG_URL with /retrieve stripped + /health), and image_analysis (IMAGE_ANALYSIS_URL with /analyze stripped + /health), each with a 2-second timeout. It returns status: "ok" only if all three report status == "ok", otherwise status: "degraded". It does not check guard, the LLM, TTS, or the database.

/status returns {active_sessions, sessions} from LiveService.snapshot() — for each live session: session_id, avatar_id, user_id, status, voice_output, created_at.

note

Clients never connect to port 8050 directly. They reach the service through the gateway WebSocket WS /api_lis/sessions/{session_id}/message, which relays to GatewayConfig.engine_ws_url (default ws://localhost:8050/message_pipeline) and injects user_id, avatar_uuid, system_prompt, voice-clone, and LoRA config into the upstream session.start.

Layer map

The service follows a ports-and-adapters layout: transport (api/), transport-agnostic orchestration (application/), outbound adapters (services/), and persistence (db/).

conversation_engine/
├── main.py                     # module `app` = build_live_app(enable_db=...)
├── container.py                # build_live_service / build_live_app wiring
├── config.py                   # re-exports ENGINE_CONFIG / APP_CONFIG / DB_CONFIG
├── payloads.py                 # wire-format Pydantic models + envelope() / error_envelope()
├── api/                        # transport layer
│   ├── app.py                  # create_live_app: CORS, /health, /status, DB-pool lifespan
│   ├── handler.py              # ConnectionHandler: event dispatch + turn-task lifecycle
│   └── routers/
│       └── message_pipeline.py # WS /message_pipeline router
├── application/                # transport-agnostic orchestration
│   ├── pipeline.py             # Pipeline.run_turn + session lifecycle
│   ├── live_service.py         # LiveService in-memory session store
│   ├── session.py              # LiveSession state
│   ├── prompt.py               # assemble_prompt + grounding modes
│   ├── image_turn.py           # resolve_question / validate_transport_image
│   ├── normalizing.py          # normalize_lora_name
│   ├── safety.py               # safety refusal text
│   ├── events.py               # DomainEvent types
│   └── ports.py                # Protocol ports
├── services/                   # outbound adapters (implement the ports)
│   ├── guard.py                # GuardClient
│   ├── llm.py                  # LLMClient
│   ├── lora.py                 # LoRAClient
│   ├── memory.py               # MemoryClient
│   ├── rag.py                  # RAGClient
│   ├── image_analysis.py       # ImageAnalysisClient
│   ├── attachment_storage.py   # AttachmentStorage (filesystem)
│   └── tts.py                  # TTSClient + FallbackTTSClient
└── db/                         # persistence
    ├── engine.py               # async psycopg pool
    ├── models.py               # ConversationMessage, ConversationAttachment
    └── repositories/
        ├── conversation.py     # ConversationRepository
        └── attachment.py       # AttachmentRepository

note

handler.py lives under conversation_engine/api/, not application/. The FastAPI app description string still reads "single /text_chat WebSocket"; the only WebSocket route is /message_pipeline.

Session model

Sessions are kept in an in-memory store, so resume works only within the same process while the session is still live.

LiveService (application/live_service.py) holds a dict[str, LiveSession] and exposes:

• open_session(config) — returns (session, resumed). If config.session_id matches a live session it is reused (resumed=True); otherwise a new LiveSession is created (with a fresh UUID when no session_id is supplied).
• get_session(session_id) / close_session(session_id) — lookup and removal.
• snapshot() — the list rendered by /status.

LiveSession (application/session.py) is a slotted dataclass holding one connection's config (avatar, user, voice-clone, memory flags, system_prompt, collection_name, lora_name) plus runtime state: status, active_task, lora_started, effective_lora_name, history, and history_base. It enforces a single active turn:

Method	Behavior
can_accept_turn()	True when there is no active task or the task is done.
start_turn(task)	Attaches the turn task; raises if a turn is already active; sets status="busy".
cancel_turn()	Cancels and awaits the active task (swallows CancelledError); resets status="ready".
finish_turn(task=None)	Clears the active task and returns status to ready.

Session lifecycle work runs outside the turn, in Pipeline.on_session_start / on_session_end:

• on_session_start loads recent history once into session.history (limited by context_window) and, on any failure, silently sets it to []. It then normalizes lora_name and pins the LoRA adapter.
• on_session_end releases the LoRA adapter (best-effort) if it was pinned.

warning

On resume (resumed=True) the handler skips on_session_start: history is not reloaded and the LoRA adapter is not re-pinned.

Connection handling

ConnectionHandler (api/handler.py) owns one connection's dispatch and turn-task lifecycle. create_message_pipeline_router accepts the socket, wraps send to only emit while the socket is WebSocketState.CONNECTED, runs the handler, and calls handler.cleanup() in a finally.

The handler validates each frame into an InboundEnvelope and dispatches one of five inbound events:

Event	Action
session.start	Opens/resumes the session; runs on_session_start only for new sessions; replies session.started (resumed flag included).
session.ping	Replies session.pong.
session.cancel	Replies session.cancelled immediately, then schedules cancel_turn() as a task named cancel_{sid}.
session.close	Closes the session (session.closed, reason client_close) and ends the loop.
message.text	Starts one turn task named turn_{sid}.

Unknown events return INVALID_EVENT. Any non-session.start event that arrives before a session exists returns SESSION_NOT_STARTED.

Turn task management. A turn runs as an asyncio task tracked by session.active_task. Exactly one turn is allowed per session — a second message.text while a turn is active returns REQUEST_ALREADY_IN_PROGRESS (recoverable). The task iterates Pipeline.run_turn, translating each DomainEvent to a wire envelope via _to_envelope.

Fatal errors. If a TurnError with recoverable=False is emitted, or an unhandled exception occurs (sent as INTERNAL_ERROR, recoverable=False), the turn is marked fatal. In the task's finally the handler closes the generator, calls finish_turn(), and on a fatal error runs _close_session("fatal_error") — which sends session.closed{reason:"fatal_error"} and closes the WebSocket. The client must reconnect and send session.start again.

Idle timeout. _recv_or_idle_close wraps receive() in asyncio.wait_for(timeout=session_idle_timeout_sec) (default 300s). On timeout, if a turn is still running it keeps waiting; otherwise it closes the session with reason idle_timeout.

See Conversation Pipeline for the full ordered turn flow and WebSocket Protocol for the exact inbound and outbound event payloads.

Service adapters

External dependencies are reached through ports (application/ports.py); the adapters below (services/) implement them. Several adapters fail open — they swallow their own errors so the turn can continue.

Adapter	Responsibility
GuardClient (guard.py)	OpenAI chat-completions safety check on the last 2048 chars; flagged unsafe iff the response contains unsafe or controversial. Catches all exceptions internally and returns safe=True (fail-open), so it never raises.
LLMClient (llm.py)	Streaming relay: POSTs {prompt, system_prompt, temperature, max_tokens, stream: true, optional lora_name} to LLM_URL and yields raw aiter_text() chunks unchanged. It does not send a model field.
LoRAClient (lora.py)	notify_starting POSTs status=starting and returns the backend ready bool; notify_ending POSTs status=ending. Both catch all exceptions (notify_starting returns False, never raises), so a failed check surfaces as LORA_NOT_READY, not LORA_ERROR.
MemoryClient (memory.py)	Injects via GET {MEMORY_URL}/get_injection_context?user_id&avatar_id; extracts via POST {MEMORY_URL}/extract. Both no-op when user_id is falsy.
RAGClient (rag.py)	POSTs {query, collection_name, user_collection_name} to RAG_URL and validates the response into RAGContext. Retrieve-only, no generation.
ImageAnalysisClient (image_analysis.py)	POSTs {image:{mime_type, data_base64, filename}} to IMAGE_ANALYSIS_URL and returns ImageAnalysisResult (visual_text, latency_ms, model_id, analysis_type). Image only; no question is sent.
AttachmentStorage (attachment_storage.py)	Decodes base64 and writes bytes to {base}/users/{user_id}/sessions/{session_id}/{message_id}.{ext} off-thread; returns (relative_storage_key, size_bytes).
TTSClient + FallbackTTSClient (tts.py)	TTSClient speaks the synthesis.start / audio.chunk / synthesis.completed WebSocket protocol to one backend. FallbackTTSClient tries the primary (ElevenLabs, TTS_WS_URL, default 8014) and, if it yields no chunk before failing, transparently replays the request to the fallback (OmniVoice, TTS_OMNIVOICE_WS_URL, default 8015).

note

Guard is fail-open at two layers: GuardClient.check already returns safe=True on any error, and the pipeline additionally wraps the call. With the default adapters, GUARD_ERROR and LORA_ERROR are effectively unreachable. TTS fallback is pre-first-chunk only — once any AudioChunk is yielded, mid-stream failures are not retried.

Persistence

The service uses an async psycopg pool (db/engine.py) whose lifespan is opened and closed by create_live_app only when enable_db=True (AppConfig.enable_db_on_startup, default true; tests use False). Row models live in db/models.py (ConversationMessage, ConversationAttachment). Two repositories target the lis schema.

ConversationRepository (db/repositories/conversation.py) owns lis.conversation_history as sole writer:

• append_turn(session_id, role, content) — INSERTs and RETURNING *.
• get_recent_turns(session_id, limit) — selects the newest N rows (created_at DESC, id DESC), LEFT JOINs lis.conversation_attachments, groups fan-out rows into one HistoryEntry per message with its attachments, and returns them in chronological ascending order.

AttachmentRepository (db/repositories/attachment.py) owns lis.conversation_attachments:

• insert(...) — writes the attachment row with description=NULL, RETURNING *.
• update_description(attachment_id, description) — back-fills the description after image analysis.
• get_by_message_id(message_id) — selects attachments ordered by id.

Only attachment metadata is stored in lis.conversation_attachments; the image bytes themselves are written to the filesystem under ATTACHMENT_STORAGE_BASE_PATH by AttachmentStorage.

note

Blocked (unsafe) messages are not persisted — the safety gate returns before the user-message write. Image-only turns persist the default question text ("Describe the relevant visible content in this image concisely.") as the user message content. A failed user-message write raises a non-recoverable DB_ERROR that closes the session.

Configuration

Key settings from EngineConfig (see config.py):

Setting	Env var	Default	Purpose
port	CHAT_ENGINE_PORT (or LIVE_PORT)	8050	Port the app binds to.
context_window	CHAT_ENGINE_CONTEXT_WINDOW (or LIVE_CONTEXT_WINDOW)	20	History turns loaded at session start and the tail slice used in the prompt.
session_idle_timeout_sec	CHAT_ENGINE_SESSION_IDLE_TIMEOUT_SEC	300.0	Idle receive timeout before an idle session is closed.
guard_url	GUARD_URL	http://localhost:4915/v1/chat/completions	OpenAI-compatible safety guard endpoint.
guard_model	GUARD_MODEL	Qwen/Qwen3Guard-Gen-0.6B	Model name sent in the guard request.
llm_url	LLM_URL	https://core-llm.nezlamna-online.education/api/generate	Streaming LLM endpoint the pipeline relays to.
llm_temperature	LLM_TEMPERATURE	0.3	Sampling temperature passed to llm.stream.
llm_max_tokens	LLM_MAX_TOKENS	1024	Max tokens passed to llm.stream.
lora_check_url	LORA_CHECK_URL	https://core-llm.nezlamna-online.education/api/lora/check	Endpoint notified with status=starting/ending.
rag_url	RAG_URL	http://localhost:8051/retrieve	RAG retrieve endpoint (also basis for the /health probe).
image_analysis_url	IMAGE_ANALYSIS_URL (aliases VLM_URL, ORCH_VLM_URL)	http://localhost:8012/analyze	VLM analyze endpoint (also basis for the /health probe).
image_analysis_max_image_bytes	IMAGE_ANALYSIS_MAX_IMAGE_BYTES (alias VLM_MAX_IMAGE_BYTES)	4194304	Max decoded image size enforced before analysis.
memory_url	MEMORY_URL	http://localhost:8052	Memory service base URL.
tts_ws_url	TTS_WS_URL	ws://localhost:8014/ws/tts	Primary TTS backend (ElevenLabs).
tts_omnivoice_ws_url	TTS_OMNIVOICE_WS_URL	ws://localhost:8015/ws/tts	Fallback TTS backend (OmniVoice).
attachment_storage_base_path	ATTACHMENT_STORAGE_BASE_PATH	./attachment_storage	Filesystem root for saved attachment files.
enable_db_on_startup	ENABLE_DB_ON_STARTUP	true	Whether the app opens/closes the Postgres pool in its lifespan.
pool_max_size	DB_POOL_MAX	10	Async psycopg pool max size (min_size=0).