Extending the System

LIS 3 follows a ports-and-adapters (hexagonal) layout. Each subsystem keeps transport, business rules, and outbound integrations in separate directories, wired together in one place (container.py). This guide shows where each kind of change belongs and gives step-by-step recipes for the two most common extensions: adding a pipeline stage to conversation_engine and adding a REST endpoint to gateway_service. See the Architecture Overview for the full picture.

Where things live

Match the concern you are changing to its layer before writing any code. Paths are relative to the repository root.

Concern	Location
Transport layer (FastAPI app factory, HTTP/WS routers, connection handling)	conversation_engine/api/, gateway_service/api/
User-facing rules (auth, session CRUD, memory, relay orchestration)	gateway_service/application/
Live-chat turn orchestration (session store, prompt assembly, events, ports)	conversation_engine/application/
A single pipeline stage (guard, persist, image, memory/RAG, LLM, TTS ordering)	conversation_engine/application/pipeline.py
An external adapter (HTTP/WS client to an upstream service)	conversation_engine/services/, gateway_service/services/
A new standalone component service (memory, RAG, image analysis, STT, TTS)	component_services/
Dependency wiring (construct adapters, build the app)	conversation_engine/container.py, gateway_service/container.py

note

Business logic never imports an adapter directly. It depends on a Protocol interface declared in application/ports.py, and the concrete adapter from services/ is injected in container.py. This is what lets tests substitute fakes (both containers accept optional keyword overrides for every port).

Adding a pipeline stage to conversation_engine

The turn is an ordered async generator: Pipeline.run_turn yields DomainEvents in sequence (validation → guard → safety gate → persist user message → image analysis → parallel memory + RAG → prompt assembly → LLM stream → persist assistant message → TTS). To insert a new stage that calls an external dependency:

1. Add a Protocol to conversation_engine/application/ports.py. Declare the interface your stage depends on, for example:

   class MyStagePort(Protocol):
       async def run(self, *, text: str) -> MyStageResult: ...

2. Implement the adapter in conversation_engine/services/. Add a client (for example my_stage.py) that satisfies the Protocol. Follow the existing adapters: httpx for HTTP (see services/rag.py, services/image_analysis.py), websockets for WS (see services/tts.py). Reach the endpoint through a value from ENGINE_CONFIG rather than a hardcoded URL.

3. Inject it in Pipeline.__init__. conversation_engine/application/pipeline.py takes every port as a keyword-only argument and stores it on self. Add your port there:

   def __init__(self, *, guard: GuardPort, ..., my_stage: MyStagePort) -> None:
       ...
       self._my_stage = my_stage

4. Wire it in conversation_engine/container.py. build_live_service constructs the Pipeline with default adapters (guard or GuardClient(), and so on). Add your adapter the same way so it is optional and overridable:

   my_stage = my_stage or MyStageClient(),

5. Add the yield statements in Pipeline.run_turn at the right position. The stage's location in the generator determines its ordering guarantees — insert it relative to the existing stages (for example after the safety gate but before persistence). Yield a DomainEvent for anything the client should see, and decide the failure mode: yield a recoverable TurnError and return for a soft failure, or a non-recoverable TurnError for a fatal one.

warning

A non-recoverable TurnError (recoverable=False) is fatal. In ConnectionHandler, a fatal turn sends session.closed{reason:'fatal_error'} and closes the WebSocket — the client must reconnect and send session.start again. Today only DB_ERROR (user-message persist), LLM_ERROR, and unhandled INTERNAL_ERROR are non-recoverable. Mark a new stage recoverable unless its failure genuinely corrupts the turn.

tip

If your stage emits a new event type, define it in conversation_engine/application/events.py and map it to its outbound wire envelope in _to_envelope (conversation_engine/api/handler.py). Events with no envelope mapping never reach the client.

Adding a REST endpoint to gateway_service

REST routers mount under the /api_lis prefix. Endpoints are authenticated through the get_current_user dependency and delegate to an application service, which reaches the database or an upstream through a port.

1. Add the route in gateway_service/api/<domain>/. Extend an existing router factory (for example create_sessions_router in api/session/sessions.py) or add a new one that returns an APIRouter. Depend on get_current_user to enforce auth and obtain current_user["user_id"]:

   @router.get("/things", response_model=ThingsResponse)
   async def list_things(current_user: dict = Depends(get_current_user)) -> Any:
       return await service.list_things(current_user["user_id"])

2. Register the router in create_gateway_app(). In gateway_service/api/app.py, include it with the shared prefix so the path resolves under /api_lis:

   app.include_router(create_things_router(things_service), prefix="/api_lis")

3. Add the service method in gateway_service/application/<domain>_service.py. Keep business rules here, not in the router — for example ownership checks (session.user_id == caller else NotFoundError) as SessionService does. Raise the typed errors from gateway_service/errors.py; the registered exception handlers translate them to the right status and body.

4. Add a Port if the endpoint needs a new external dependency. Declare the interface in gateway_service/application/ports.py (repositories and upstream WS clients are all Protocols there) and implement it under gateway_service/db/repositories/ or gateway_service/services/.

5. Wire it in gateway_service/container.py. build_gateway_service instantiates repositories, clients, and the application services. Add your new port and pass it into the service that consumes it, keeping the optional-override keyword pattern used by the existing wiring.

note

API-key auth is fail-open: when GATEWAY_API_TOKEN is empty the key check is skipped, but get_current_user still requires a valid X-User-ID UUID. New endpoints inherit this behavior automatically by depending on get_current_user.