= 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.