feat(server+tui): event bridge — live streaming and historical replay

- SessionEventBridge: replaySnapshot() sends all historical events on connect; streamLive(sessionId) subscribes to live event flow per session - DomainEventMapper: maps 14 domain event types to ServerMessage; reads InferenceCompleted response text from ArtifactStore with empty-string fallback - GlobalStreamHandler: wires bridge on connect for replay, launches per-session live-stream job on StartSession, cancels all jobs on disconnect; removes ProtocolError("ping") heartbeat hack - SessionStreamHandler: fixes replay to use DomainEventMapper instead of emitting SessionStarted for every event; injects real artifactStore; removes ProtocolError("ping") heartbeat; adds structured logging throughout - Main: startup log block showing port, model config, sandbox, stores, tools, providers - SessionsReducer: StageCompleted/StageFailed now clears currentStage; adds 5 missing tests (InferenceCompleted, StageStarted, ToolCompleted, SessionPaused) - RouterPanel: new TUI widget rendering active session's lastResponseText - TuiWsClient: remove println that was polluting the TUI on WS connect failure
2026-05-19 00:03:53 +04:00
parent 03615dc5b9
commit f32d400138
14 changed files with 802 additions and 103 deletions
@@ -35,8 +35,11 @@ import com.correx.infrastructure.tools.ToolConfig
 import com.correx.core.events.EventDispatcher
 import io.ktor.server.engine.embeddedServer
 import io.ktor.server.netty.Netty
+import org.slf4j.LoggerFactory
 import java.nio.file.Paths

+private val log = LoggerFactory.getLogger("com.correx.apps.server.Main")
+
 fun main() {
    val artifactStore = InfrastructureModule.createArtifactStore()
    val eventStore = LoggingEventStore(InfrastructureModule.createEventStore(artifactStore))
@@ -47,8 +50,6 @@ fun main() {
    val toolRegistry = InfrastructureModule.createToolRegistry(buildToolConfig(artifactStore, sandboxRoot))
    val toolExecutor = InfrastructureModule.createToolExecutor(
        registry = toolRegistry,
-        approvalEngine = approvalEngine,
-        eventStore = eventStore,
        eventDispatcher = EventDispatcher(eventStore),
        workDir = sandboxRoot,
    )
@@ -72,10 +73,26 @@ fun main() {
    val module = ServerModule(
        orchestrator = orchestrator,
        eventStore = eventStore,
+        artifactStore = artifactStore,
        sessionRepository = repositories.sessionRepository,
        workflowRegistry = FileSystemWorkflowRegistry(InfrastructureModule.createWorkflowLoader()),
        providerRegistry = infraRegistry.asServerRegistry(),
    )
+    val modelId   = System.getenv("CORREX_MODEL_ID")   ?: "default"
+    val modelPath = System.getenv("CORREX_MODEL_PATH") ?: "(not set)"
+    val llamaUrl  = System.getenv("CORREX_LLAMA_URL")  ?: "http://127.0.0.1:10000"
+    log.info("=== correx server starting ===")
+    log.info("  port         : 8080")
+    log.info("  model id     : {}", modelId)
+    log.info("  model path   : {}", modelPath)
+    log.info("  llama url    : {}", llamaUrl)
+    log.info("  sandbox root : {}", sandboxRoot)
+    log.info("  event store  : {}", eventStore::class.simpleName)
+    log.info("  artifact store: {}", artifactStore::class.simpleName)
+    log.info("  tools        : {}", toolRegistry.all().map { it.name })
+    log.info("  providers    : {} registered", infraRegistry.listAll().size)
+    log.info("==============================")
+
    embeddedServer(Netty, port = 8080) { configureServer(module) }.start(wait = true)
 }

@@ -2,6 +2,7 @@ package com.correx.apps.server

 import com.correx.apps.server.registry.ProviderRegistry
 import com.correx.apps.server.registry.WorkflowRegistry
+import com.correx.core.artifactstore.ArtifactStore
 import com.correx.core.events.stores.EventStore
 import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
 import com.correx.core.sessions.DefaultSessionRepository
@@ -9,6 +10,7 @@ import com.correx.core.sessions.DefaultSessionRepository
 data class ServerModule(
    val orchestrator: DefaultSessionOrchestrator,
    val eventStore: EventStore,
+    val artifactStore: ArtifactStore,
    val sessionRepository: DefaultSessionRepository,
    val workflowRegistry: WorkflowRegistry,
    val providerRegistry: ProviderRegistry,
@@ -0,0 +1,94 @@
+package com.correx.apps.server.bridge
+
+import com.correx.apps.server.protocol.PauseReason
+import com.correx.apps.server.protocol.RiskSummaryDto
+import com.correx.apps.server.protocol.ServerMessage
+import com.correx.core.artifactstore.ArtifactStore
+import com.correx.core.events.events.ApprovalRequestedEvent
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.events.InferenceCompletedEvent
+import com.correx.core.events.events.InferenceStartedEvent
+import com.correx.core.events.events.InferenceTimeoutEvent
+import com.correx.core.events.events.OrchestrationPausedEvent
+import com.correx.core.events.events.StageCompletedEvent
+import com.correx.core.events.events.StageFailedEvent
+import com.correx.core.events.events.ToolExecutionCompletedEvent
+import com.correx.core.events.events.ToolExecutionFailedEvent
+import com.correx.core.events.events.ToolExecutionRejectedEvent
+import com.correx.core.events.events.ToolInvocationRequestedEvent
+import com.correx.core.events.events.TransitionExecutedEvent
+import com.correx.core.events.events.WorkflowCompletedEvent
+import com.correx.core.events.events.WorkflowFailedEvent
+import com.correx.core.events.events.WorkflowStartedEvent
+import com.correx.core.events.events.StoredEvent
+
+class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
+    suspend fun map(event: StoredEvent): ServerMessage? =
+        domainEventToServerMessage(event, artifactStore)
+}
+
+private object NoopArtifactStore : ArtifactStore {
+    override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("noop")
+    override suspend fun get(id: ArtifactId): ByteArray? = null
+    override suspend fun flushBefore(commit: suspend () -> Unit) = commit()
+}
+
+@Suppress("CyclomaticComplexMethod")
+suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: ArtifactStore): ServerMessage? =
+    when (val p = event.payload) {
+        is WorkflowStartedEvent -> mapWorkflowStarted(p)
+        is WorkflowCompletedEvent -> ServerMessage.SessionCompleted(sessionId = p.sessionId)
+        is WorkflowFailedEvent -> ServerMessage.SessionFailed(sessionId = p.sessionId, reason = p.reason)
+        is TransitionExecutedEvent -> ServerMessage.StageStarted(sessionId = p.sessionId, stageId = p.to)
+        is StageCompletedEvent -> ServerMessage.StageCompleted(sessionId = p.sessionId, stageId = p.stageId)
+        is StageFailedEvent -> ServerMessage.StageFailed(
+            sessionId = p.sessionId, stageId = p.stageId, reason = p.reason,
+        )
+        is OrchestrationPausedEvent -> mapOrchestrationPaused(p)
+        is InferenceStartedEvent -> ServerMessage.InferenceStarted(sessionId = p.sessionId, stageId = p.stageId)
+        is InferenceCompletedEvent -> mapInferenceCompleted(p, artifactStore)
+        is InferenceTimeoutEvent -> ServerMessage.InferenceTimedOut(
+            sessionId = p.sessionId, stageId = p.stageId, elapsedMs = p.timeoutMs,
+        )
+        is ToolInvocationRequestedEvent -> ServerMessage.ToolStarted(
+            sessionId = p.sessionId, toolName = p.toolName, tier = p.tier,
+        )
+        is ToolExecutionCompletedEvent -> ServerMessage.ToolCompleted(
+            sessionId = p.sessionId, toolName = p.toolName, outputSummary = p.receipt.outputSummary,
+        )
+        is ToolExecutionFailedEvent -> ServerMessage.ToolFailed(
+            sessionId = p.sessionId, toolName = p.toolName, reason = p.reason,
+        )
+        is ToolExecutionRejectedEvent -> ServerMessage.ToolRejected(
+            sessionId = p.sessionId, toolName = p.toolName, reason = p.reason,
+        )
+        is ApprovalRequestedEvent -> mapApprovalRequested(p)
+        else -> null
+    }
+
+private fun mapWorkflowStarted(p: WorkflowStartedEvent): ServerMessage =
+    ServerMessage.SessionStarted(sessionId = p.sessionId, workflowId = p.startStageId.value)
+
+private fun mapOrchestrationPaused(p: OrchestrationPausedEvent): ServerMessage {
+    val reason = if (p.reason == "APPROVAL_PENDING") PauseReason.APPROVAL_PENDING else PauseReason.USER_REQUESTED
+    return ServerMessage.SessionPaused(sessionId = p.sessionId, reason = reason)
+}
+
+private suspend fun mapInferenceCompleted(p: InferenceCompletedEvent, artifactStore: ArtifactStore): ServerMessage {
+    val text = runCatching {
+        artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: ""
+    }.getOrElse { "" }
+    return ServerMessage.InferenceCompleted(
+        sessionId = p.sessionId, stageId = p.stageId, outputSummary = text, responseText = text,
+    )
+}
+
+private fun mapApprovalRequested(p: ApprovalRequestedEvent): ServerMessage =
+    ServerMessage.ApprovalRequired(
+        sessionId = p.sessionId,
+        requestId = p.requestId,
+        tier = p.tier,
+        riskSummary = RiskSummaryDto("unknown", emptyList(), ""),
+        toolName = null,
+        preview = null,
+    )
@@ -0,0 +1,24 @@
+package com.correx.apps.server.bridge
+
+import com.correx.apps.server.protocol.ServerMessage
+import com.correx.core.artifactstore.ArtifactStore
+import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.SessionId
+
+class SessionEventBridge(
+    private val eventStore: EventStore,
+    private val artifactStore: ArtifactStore,
+    private val send: suspend (ServerMessage) -> Unit,
+) {
+    suspend fun replaySnapshot() {
+        eventStore.allEvents().toList()
+            .mapNotNull { domainEventToServerMessage(it, artifactStore) }
+            .forEach { send(it) }
+    }
+
+    suspend fun streamLive(sessionId: SessionId) {
+        eventStore.subscribe(sessionId).collect { event ->
+            domainEventToServerMessage(event, artifactStore)?.let { send(it) }
+        }
+    }
+}
@@ -1,6 +1,7 @@
 package com.correx.apps.server.ws

 import com.correx.apps.server.ServerModule
+import com.correx.apps.server.bridge.SessionEventBridge
 import com.correx.apps.server.protocol.ClientMessage
 import com.correx.apps.server.protocol.ProtocolSerializer
 import com.correx.apps.server.protocol.ProviderHealthDto
@@ -16,31 +17,34 @@ import com.correx.core.utils.TypeId
 import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.websocket.Frame
 import io.ktor.websocket.readText
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.datetime.Clock
 import org.slf4j.LoggerFactory
-import java.util.*
+import java.util.UUID

-private const val HEARTBEAT_INTERVAL_MS = 30_000L
 private val log = LoggerFactory.getLogger(GlobalStreamHandler::class.java)

 class GlobalStreamHandler(private val module: ServerModule) {

    suspend fun handle(session: DefaultWebSocketServerSession) {
        log.info("client connected")
-        sendInitialSnapshot(session)

-        val heartbeatJob = session.launch {
-            while (isActive) {
-                delay(HEARTBEAT_INTERVAL_MS)
-                val ping = ServerMessage.ProtocolError("ping")
-                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(ping)))
-            }
+        val sendFrame: suspend (ServerMessage) -> Unit = {
+            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(it)))
        }

+        val bridge = SessionEventBridge(
+            eventStore = module.eventStore,
+            artifactStore = module.artifactStore,
+            send = sendFrame,
+        )
+        sendInitialSnapshot(session)
+        bridge.replaySnapshot()
+
+        val liveStreamJobs = mutableMapOf<SessionId, Job>()
+
        try {
            for (frame in session.incoming) {
                if (frame is Frame.Text) {
@@ -49,7 +53,7 @@ class GlobalStreamHandler(private val module: ServerModule) {
                        .onSuccess { msg ->
                            log.debug("recv: {}", msg::class.simpleName)
                            runCatching {
-                                handleClientMessage(session, msg)
+                                handleClientMessage(session, msg, liveStreamJobs, sendFrame)
                            }.onFailure {
                                log.error("handle failed", it)
                            }
@@ -64,7 +68,7 @@ class GlobalStreamHandler(private val module: ServerModule) {
        } catch (_: ClosedReceiveChannelException) {
            log.info("client disconnected")
        } finally {
-            heartbeatJob.cancel()
+            liveStreamJobs.values.forEach { it.cancel() }
        }
    }

@@ -88,64 +92,76 @@ class GlobalStreamHandler(private val module: ServerModule) {
        }
    }

-    private suspend fun handleClientMessage(session: DefaultWebSocketServerSession, msg: ClientMessage) {
+    private suspend fun handleClientMessage(
+        session: DefaultWebSocketServerSession,
+        msg: ClientMessage,
+        liveStreamJobs: MutableMap<SessionId, Job>,
+        sendFrame: suspend (ServerMessage) -> Unit,
+    ) {
        when (msg) {
            is ClientMessage.Ping -> Unit
-
-            is ClientMessage.StartSession -> {
-                val graph = module.workflowRegistry.find(msg.workflowId)
-                log.info("find returned: {}", graph)
-                if (graph == null) {
-                    val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}")
-                    session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
-                    return
-                }
-                val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
-                log.info("starting session={} workflow={}", sessionId.value, msg.workflowId)
-                session.launch {
-                    val config = OrchestrationConfig(
-                        defaultSystemPromptPath = "~/.config/correx/prompts/default_system.md",
-                    )
-                    runCatching { module.orchestrator.run(sessionId, graph, config) }
-                        .onFailure { ex ->
-                            log.error("run failed for session={}: {}", sessionId.value, ex.message, ex)
-                            module.eventStore.append(
-                                NewEvent(
-                                    metadata = EventMetadata(
-                                        eventId = EventId(UUID.randomUUID().toString()),
-                                        sessionId = sessionId,
-                                        timestamp = Clock.System.now(),
-                                        schemaVersion = 1,
-                                        causationId = null,
-                                        correlationId = null,
-                                    ),
-                                    payload = WorkflowFailedEvent(
-                                        sessionId = sessionId,
-                                        stageId = graph.start,
-                                        reason = ex.message ?: "unexpected orchestrator failure",
-                                        retryExhausted = false,
-                                    ),
-                                ),
-                            )
-                        }
-                }
-                val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId)
-                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started)))
-            }
-
-            is ClientMessage.CancelSession -> {
-                module.orchestrator.cancel(msg.sessionId)
-            }
-
-            is ClientMessage.ResumeSession -> {
-                val error = ServerMessage.ProtocolError("ResumeSession not supported")
-                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
-            }
-
-            is ClientMessage.ApprovalResponse -> {
-                val error = ServerMessage.ProtocolError("ApprovalResponse not supported on global stream")
-                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
-            }
+            is ClientMessage.StartSession -> handleStartSession(session, msg, liveStreamJobs, sendFrame)
+            is ClientMessage.CancelSession -> module.orchestrator.cancel(msg.sessionId)
+            is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported"))
+            is ClientMessage.ApprovalResponse ->
+                session.send(encodeError("ApprovalResponse not supported on global stream"))
        }
    }
+
+    private fun encodeError(message: String): Frame.Text =
+        Frame.Text(ProtocolSerializer.encodeServerMessage(ServerMessage.ProtocolError(message)))
+
+    private suspend fun handleStartSession(
+        session: DefaultWebSocketServerSession,
+        msg: ClientMessage.StartSession,
+        liveStreamJobs: MutableMap<SessionId, Job>,
+        sendFrame: suspend (ServerMessage) -> Unit,
+    ) {
+        val graph = module.workflowRegistry.find(msg.workflowId)
+        log.info("find returned: {}", graph)
+        if (graph == null) {
+            val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}")
+            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
+            return
+        }
+        val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
+        log.info("starting session={} workflow={}", sessionId.value, msg.workflowId)
+        session.launch {
+            val config = OrchestrationConfig(
+                defaultSystemPromptPath = "~/.config/correx/prompts/default_system.md",
+            )
+            runCatching { module.orchestrator.run(sessionId, graph, config) }
+                .onFailure { ex ->
+                    log.error("run failed for session={}: {}", sessionId.value, ex.message, ex)
+                    module.eventStore.append(
+                        NewEvent(
+                            metadata = EventMetadata(
+                                eventId = EventId(UUID.randomUUID().toString()),
+                                sessionId = sessionId,
+                                timestamp = Clock.System.now(),
+                                schemaVersion = 1,
+                                causationId = null,
+                                correlationId = null,
+                            ),
+                            payload = WorkflowFailedEvent(
+                                sessionId = sessionId,
+                                stageId = graph.start,
+                                reason = ex.message ?: "unexpected orchestrator failure",
+                                retryExhausted = false,
+                            ),
+                        ),
+                    )
+                }
+        }
+        liveStreamJobs.put(sessionId, session.launch {
+            val bridge = SessionEventBridge(
+                eventStore = module.eventStore,
+                artifactStore = module.artifactStore,
+                send = sendFrame,
+            )
+            bridge.streamLive(sessionId)
+        })?.cancel()
+        val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId)
+        session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started)))
+    }
 }
@@ -1,6 +1,7 @@
 package com.correx.apps.server.ws

 import com.correx.apps.server.ServerModule
+import com.correx.apps.server.bridge.DomainEventMapper
 import com.correx.apps.server.protocol.ApprovalDecision
 import com.correx.apps.server.protocol.ClientMessage
 import com.correx.apps.server.protocol.ProtocolSerializer
@@ -19,37 +20,28 @@ import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.websocket.Frame
 import io.ktor.websocket.readText
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
-import kotlinx.coroutines.delay
-import kotlinx.coroutines.isActive
-import kotlinx.coroutines.launch
 import kotlinx.datetime.Clock
+import org.slf4j.LoggerFactory

-private const val HEARTBEAT_INTERVAL_MS = 30_000L
+private val log = LoggerFactory.getLogger(SessionStreamHandler::class.java)

 class SessionStreamHandler(private val module: ServerModule) {

+    private val mapper = DomainEventMapper(module.artifactStore)
+
    suspend fun handle(session: DefaultWebSocketServerSession, sessionId: SessionId, lastEventId: Long?) {
+        log.info("client connected session={} lastEventId={}", sessionId.value, lastEventId)
+
        val replayEvents = if (lastEventId != null) {
            module.eventStore.readFrom(sessionId, lastEventId)
        } else {
            emptyList()
        }
-
-        runCatching { module.sessionRepository.getSession(sessionId) }.onSuccess {
-            val snapshot = ServerMessage.SessionStarted(sessionId, sessionId.value)
-            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(snapshot)))
-        }
+        log.debug("replaying {} event(s) for session={}", replayEvents.size, sessionId.value)

        for (event in replayEvents) {
-            val msg = ServerMessage.SessionStarted(sessionId, event.metadata.eventId.value)
-            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg)))
-        }
-
-        val heartbeatJob = session.launch {
-            while (isActive) {
-                delay(HEARTBEAT_INTERVAL_MS)
-                val ping = ServerMessage.ProtocolError("ping")
-                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(ping)))
+            mapper.map(event)?.let { msg ->
+                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg)))
            }
        }

@@ -58,17 +50,19 @@ class SessionStreamHandler(private val module: ServerModule) {
                if (frame is Frame.Text) {
                    val text = frame.readText()
                    runCatching { ProtocolSerializer.decodeClientMessage(text) }
-                        .onSuccess { msg -> handleClientMessage(session, msg) }
+                        .onSuccess { msg ->
+                            log.debug("recv: {}", msg::class.simpleName)
+                            handleClientMessage(session, msg)
+                        }
                        .onFailure {
+                            log.warn("decode error: {}", it.message)
                            val error = ServerMessage.ProtocolError("Unknown message: ${it.message}")
                            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
                        }
                }
            }
        } catch (_: ClosedReceiveChannelException) {
-            // client disconnected
-        } finally {
-            heartbeatJob.cancel()
+            log.info("client disconnected session={}", sessionId.value)
        }
    }

@@ -76,13 +70,16 @@ class SessionStreamHandler(private val module: ServerModule) {
        when (msg) {
            is ClientMessage.Ping -> Unit
            is ClientMessage.CancelSession -> {
-                module.orchestrator.cancel(msg.sessionId)
+                runCatching { module.orchestrator.cancel(msg.sessionId) }
+                    .onFailure { log.error("cancel failed for session={}: {}", msg.sessionId.value, it.message, it) }
            }
            is ClientMessage.ApprovalResponse -> {
                val domainDecision = msg.toDomainDecision(msg.requestId.value)
                runCatching { module.orchestrator.submitApprovalDecision(msg.requestId, domainDecision) }
+                    .onFailure { log.error("submitApprovalDecision failed for request={}: {}", msg.requestId.value, it.message, it) }
            }
            else -> {
+                log.warn("unexpected message type={} in session stream", msg::class.simpleName)
                val error = ServerMessage.ProtocolError("Unexpected message type in session stream")
                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
            }