correx/docs/modules/apps/app-server.adoc

= app-server

== purpose

Ktor-based HTTP + WebSocket server that hosts the Correx orchestration runtime. Wires together core domain modules with infrastructure adapters to run, observe, and manage agentic workflow sessions.

== responsibilities

* Expose REST endpoints for session lifecycle (create, get, cancel, list events)
* Expose REST endpoints for workflow discovery and provider health
* Maintain long-lived WebSocket connections for per-session and global event streaming
* Bridge domain events (`StoredEvent`) to wire protocol messages (`ServerMessage`)
* Coordinate approval workflows: register WS clients, receive decisions, forward to orchestration
* Log all emitted events via `LoggingEventStore`
* Initialize and wire the full dependency graph: event store, repositories, engines, registries

== non-responsibilities

* Does not perform inference — delegates to `core:inference` via `InferenceRouter`
* Does not execute tools — delegates to `infrastructure:tools` via `ToolExecutor`
* Does not persist state beyond what the event store and artifact store provide
* Does not handle TLS termination or authentication (no security middleware)
* Does not provide horizontal scaling or clustering

== key types

=== ServerModule
* **kind**: class
* **purpose**: Top-level wiring container holding all server-scoped dependencies (orchestrator, event store, registries, repositories, coordinator). Creates an `ApprovalCoordinator` and subscribes to `ApprovalRequestedEvent` on `start()`.
* **fields**: `moduleScope: CoroutineScope` — SupervisorJob-backed scope owned by the module; hosts the event subscription and approval timeout jobs.

=== Application.configureServer(module)
* **kind**: extension function
* **purpose**: Installs Ktor plugins (CallLogging, WebSockets, ContentNegotiation, StatusPages) and defines routing tree.

=== GlobalStreamHandler
* **kind**: class
* **purpose**: Handles `/stream` WebSocket connections. Sends initial snapshot (provider health + workflow list), then streams all events from `eventStore.subscribeAll()` via `SessionEventBridge` + `DomainEventMapper`.

=== SessionStreamHandler
* **kind**: class
* **purpose**: Handles `/sessions/{id}/stream` WebSocket connections. Replays events from optionally specified `lastEventId` cursor, then forwards incoming client messages (approval, cancel, chat input, ping) to the appropriate module component.

=== DomainEventMapper
* **kind**: class
* **purpose**: Maps domain `StoredEvent.payload` variants to protocol `ServerMessage` variants for WebSocket transmission.

=== SessionEventBridge
* **kind**: class
* **purpose**: Reconstructs snapshot state (orchestration status, pending approvals, recent events) for all active sessions from the event store. Used during global stream initialization.

=== ApprovalCoordinator
* **kind**: class
* **purpose**: Manages approval request routing: registers session and global WS clients, broadcasts `ApprovalRequired` messages, schedules timeout-based rejection, forwards `ApprovalResponse` decisions to the orchestrator.

=== ApprovalResponseMapper (internal)
* **kind**: extension function
* **purpose**: Translates `ClientMessage.ApprovalResponse` to domain `ApprovalDecision`.

=== LoggingEventStore
* **kind**: class
* **purpose**: Decorator over `EventStore` that logs all `append`/`appendAll` calls with session ID, event type, and outcome.

=== ProviderRegistry
* **kind**: interface
* **purpose**: Thin server-side adapter over `DefaultProviderRegistry` — exposes `listAll()` and `healthCheckAll()`.

=== WorkflowRegistry
* **kind**: interface
* **purpose**: Abstraction for discovering and loading `WorkflowGraph` definitions by ID.

=== FileSystemWorkflowRegistry
* **kind**: class
* **purpose**: Implements `WorkflowRegistry` by scanning `~/.config/correx/workflows/*.toml` lazily. Caches loaded graphs.

=== ServerMessage / ClientMessage / Dtos
* **kind**: sealed interfaces / data classes
* **purpose**: Wire protocol types. `ServerMessage` has two marker subtypes: `SessionMessage` (event-derived, carries sequence cursors) and `NonEventMessage` (control messages, no cursors).

== event flow

*inbound:* REST requests (POST/DELETE sessions, GET status/workflows/providers) and WebSocket `ClientMessage` frames (approval responses, cancel, chat input, ping).

*outbound:* WebSocket `ServerMessage` frames from `DomainEventMapper` (session lifecycle, stage lifecycle, tool lifecycle, inference lifecycle, approval requests). Also HTTP JSON responses for REST endpoints.

== integration points

* `:core:kernel` — `DefaultSessionOrchestrator` for running sessions, `ApprovalGateway` for injecting decisions
* `:core:events` — `EventStore`, `StoredEvent`, all event payload types
* `:core:approvals` — `DefaultApprovalEngine`, `Tier`, domain approval types
* `:core:sessions` — `DefaultSessionRepository`, `DefaultSessionReducer`, `SessionProjector`
* `:core:inference` — `DefaultInferenceRouter`, `InferenceProjector`, `InferenceRepository`
* `:core:transitions` — `DefaultTransitionResolver`, `WorkflowGraph`
* `:core:context` — `DefaultContextPackBuilder`, `DefaultContextCompressor`
* `:core:validation` — `ValidationPipeline`, `ArtifactPayloadValidator`
* `:core:risk` — `DefaultRiskAssessor`
* `:core:router` — `RouterFacade`, `RouterConfig`
* `:core:tools` — `ToolRegistry`
* `:core:config` — `ConfigLoader`, `ApprovalConfig`
* `:infrastructure:*` — providers, tool executor, workflow loader, artifact repository, approval repository, persistence

== invariants

* `ServerModule.start()` is idempotent — calling it twice is a no-op for the second call (`subscriptionJob` guard)
* Event subscription is established BEFORE `replaySnapshot()` reads `lastGlobalSequence` to prevent missed events
* Orchestrator is launched in `module.moduleScope` after protocol messages are delivered (ensures no silent data loss on disconnected WS)
* Approval timeout is scheduled per-request; `resolved` map prevents double-resolution
* `DomainEventMapper` returns `null` for unmapped payload types, which are silently dropped from the stream

== PlantUML diagram

[plantuml, app-server, "png"]
----
include::../../diagrams/app-server.puml[]
----




== known issues

* `GET /sessions` returns `emptyList()` unconditionally — no server-side session listing is implemented
* `SessionStreamHandler` does not handle `ClientMessage.ResumeSession` — it returns a `ProtocolError`
* `DomainEventMapper` defaults to `NoopArtifactStore`, but all production call sites (`GlobalStreamHandler`, `SessionStreamHandler`) inject `module.artifactStore` — the `NoopArtifactStore` path is only reachable if a caller omits the argument
* `RiskSummaryDto` is hardcoded to `"unknown"` in `mapApprovalRequested` — the approval event does not carry risk information
* Session IDs are generated server-side as random UUIDs; client-provided `sessionId` in `StartSessionRequest` is accepted but not used

== open questions

* Should the server support session listing by filtering on status or workflow ID?
* Should `RiskSummary` data be attached to `ApprovalRequestedEvent` payload, or should the server query it separately?
* Is the global stream (`/stream`) scaling strategy adequate for N concurrent clients? The `Channel` buffer size is fixed at 1024.