From f32d4001382a38afee32d88c5571aa433b560706 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 19 May 2026 00:03:53 +0400 Subject: [PATCH] =?UTF-8?q?feat(server+tui):=20event=20bridge=20=E2=80=94?= =?UTF-8?q?=20live=20streaming=20and=20historical=20replay?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- apps/server/build.gradle | 2 + .../kotlin/com/correx/apps/server/Main.kt | 21 +- .../com/correx/apps/server/ServerModule.kt | 2 + .../apps/server/bridge/DomainEventMapper.kt | 94 ++++++ .../apps/server/bridge/SessionEventBridge.kt | 24 ++ .../apps/server/ws/GlobalStreamHandler.kt | 154 +++++---- .../apps/server/ws/SessionStreamHandler.kt | 43 ++- .../server/bridge/DomainEventMapperTest.kt | 300 ++++++++++++++++++ .../server/bridge/SessionEventBridgeTest.kt | 135 ++++++++ .../main/kotlin/com/correx/apps/tui/TuiApp.kt | 5 + .../correx/apps/tui/components/RouterPanel.kt | 24 ++ .../apps/tui/reducer/SessionsReducer.kt | 20 +- .../com/correx/apps/tui/ws/TuiWsClient.kt | 2 +- .../apps/tui/reducer/SessionsReducerTest.kt | 79 +++++ 14 files changed, 802 insertions(+), 103 deletions(-) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt create mode 100644 apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt diff --git a/apps/server/build.gradle b/apps/server/build.gradle index f4901c0b..f074bb31 100644 --- a/apps/server/build.gradle +++ b/apps/server/build.gradle @@ -43,6 +43,8 @@ dependencies { implementation "io.ktor:ktor-server-status-pages:$ktor_version" implementation "io.ktor:ktor-server-call-logging:$ktor_version" + testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:$kotlinx_coroutines_version" + implementation "org.apache.logging.log4j:log4j-core:2.24.1" implementation "org.apache.logging.log4j:log4j-slf4j2-impl:2.24.1" implementation "org.slf4j:slf4j-api:2.0.16" diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 7304b597..d746377e 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -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) } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 9df81668..eb089017 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -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, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt new file mode 100644 index 00000000..85fb3d5b --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt @@ -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, + ) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt new file mode 100644 index 00000000..246c9295 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt @@ -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) } + } + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index 072bf351..5f5f338f 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -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() + 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, + 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, + 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))) + } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt index e225803d..aad54495 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt @@ -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))) } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt new file mode 100644 index 00000000..c9ee0228 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt @@ -0,0 +1,300 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.PauseReason +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.approvals.Tier +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.EventMetadata +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.SessionStartedEvent +import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.events.StageFailedEvent +import com.correx.core.events.events.StoredEvent +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.ToolReceipt +import com.correx.core.events.events.ToolRequest +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.types.ArtifactId +import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.InferenceRequestId +import com.correx.core.events.types.ProviderId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.ToolInvocationId +import com.correx.core.events.types.TransitionId +import com.correx.core.events.types.ValidationReportId +import com.correx.core.inference.TokenUsage +import kotlinx.datetime.Clock +import kotlinx.datetime.Instant +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test + +class DomainEventMapperTest { + + private val sessionId = SessionId("session-1") + private val stageId = StageId("stage-1") + private val timestamp = Instant.parse("2026-01-01T00:00:00Z") + private val noopStore: ArtifactStore = object : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("art-1") + override suspend fun get(id: ArtifactId): ByteArray? = null + override suspend fun flushBefore(commit: suspend () -> Unit) = commit() + } + + private fun storedEvent(payload: com.correx.core.events.events.EventPayload): StoredEvent = + StoredEvent( + metadata = EventMetadata( + eventId = EventId("evt-1"), + sessionId = sessionId, + timestamp = timestamp, + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = 1L, + payload = payload, + ) + + @Test + fun `WorkflowStartedEvent maps to SessionStarted`(): Unit = runTest { + val event = storedEvent(WorkflowStartedEvent(sessionId = sessionId, startStageId = stageId)) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), result) + } + + @Test + fun `WorkflowCompletedEvent maps to SessionCompleted`(): Unit = runTest { + val event = storedEvent( + WorkflowCompletedEvent(sessionId = sessionId, terminalStageId = stageId, totalStages = 1), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.SessionCompleted(sessionId), result) + } + + @Test + fun `WorkflowFailedEvent maps to SessionFailed`(): Unit = runTest { + val event = storedEvent( + WorkflowFailedEvent(sessionId = sessionId, stageId = stageId, reason = "boom", retryExhausted = false), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.SessionFailed(sessionId, "boom"), result) + } + + @Test + fun `TransitionExecutedEvent maps to StageStarted with to stageId`(): Unit = runTest { + val from = StageId("from-stage") + val to = StageId("to-stage") + val event = storedEvent( + TransitionExecutedEvent(sessionId = sessionId, from = from, to = to, transitionId = TransitionId("t1")), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.StageStarted(sessionId, to), result) + } + + @Test + fun `StageCompletedEvent maps to StageCompleted`(): Unit = runTest { + val event = storedEvent( + StageCompletedEvent(sessionId = sessionId, stageId = stageId, transitionId = TransitionId("t1")), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.StageCompleted(sessionId, stageId), result) + } + + @Test + fun `StageFailedEvent maps to StageFailed`(): Unit = runTest { + val event = storedEvent( + StageFailedEvent( + sessionId = sessionId, stageId = stageId, transitionId = TransitionId("t1"), reason = "err", + ), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.StageFailed(sessionId, stageId, "err"), result) + } + + @Test + fun `OrchestrationPausedEvent with APPROVAL_PENDING maps to APPROVAL_PENDING`(): Unit = runTest { + val event = storedEvent( + OrchestrationPausedEvent(sessionId = sessionId, stageId = stageId, reason = "APPROVAL_PENDING"), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.SessionPaused(sessionId, PauseReason.APPROVAL_PENDING), result) + } + + @Test + fun `OrchestrationPausedEvent with other reason maps to USER_REQUESTED`(): Unit = runTest { + val event = storedEvent( + OrchestrationPausedEvent(sessionId = sessionId, stageId = stageId, reason = "USER_REQUESTED"), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.SessionPaused(sessionId, PauseReason.USER_REQUESTED), result) + } + + @Test + fun `InferenceStartedEvent maps to InferenceStarted`(): Unit = runTest { + val event = storedEvent(InferenceStartedEvent( + requestId = InferenceRequestId("req-1"), + sessionId = sessionId, + stageId = stageId, + providerId = ProviderId("p1"), + promptArtifactId = ArtifactId("art-prompt"), + )) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.InferenceStarted(sessionId, stageId), result) + } + + @Test + fun `InferenceCompletedEvent reads artifact and maps to InferenceCompleted`(): Unit = runTest { + val artifactId = ArtifactId("art-resp") + val responseText = "hello world" + val store = object : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("x") + override suspend fun get(id: ArtifactId): ByteArray? = + if (id == artifactId) responseText.toByteArray(Charsets.UTF_8) else null + override suspend fun flushBefore(commit: suspend () -> Unit) = commit() + } + val event = storedEvent(InferenceCompletedEvent( + requestId = InferenceRequestId("req-1"), + sessionId = sessionId, + stageId = stageId, + providerId = ProviderId("p1"), + tokensUsed = TokenUsage(10, 5), + latencyMs = 100L, + responseArtifactId = artifactId, + )) + val result = domainEventToServerMessage(event, store) + assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, responseText, responseText), result) + } + + @Test + fun `InferenceCompletedEvent falls back to empty string when artifact missing`(): Unit = runTest { + val event = storedEvent(InferenceCompletedEvent( + requestId = InferenceRequestId("req-1"), + sessionId = sessionId, + stageId = stageId, + providerId = ProviderId("p1"), + tokensUsed = TokenUsage(10, 5), + latencyMs = 100L, + responseArtifactId = ArtifactId("missing"), + )) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, "", ""), result) + } + + @Test + fun `InferenceTimeoutEvent maps to InferenceTimedOut`(): Unit = runTest { + val event = storedEvent(InferenceTimeoutEvent( + requestId = InferenceRequestId("req-1"), + sessionId = sessionId, + stageId = stageId, + providerId = ProviderId("p1"), + timeoutMs = 5000L, + )) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.InferenceTimedOut(sessionId, stageId, 5000L), result) + } + + @Test + fun `ToolInvocationRequestedEvent maps to ToolStarted`(): Unit = runTest { + val invId = ToolInvocationId("inv-1") + val event = storedEvent(ToolInvocationRequestedEvent( + invocationId = invId, + sessionId = sessionId, + stageId = stageId, + toolName = "file_write", + tier = Tier.T2, + request = ToolRequest( + invocationId = invId, sessionId = sessionId, stageId = stageId, toolName = "file_write", + ), + )) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.ToolStarted(sessionId, "file_write", Tier.T2), result) + } + + @Test + fun `ToolExecutionCompletedEvent maps to ToolCompleted`(): Unit = runTest { + val invId = ToolInvocationId("inv-1") + val receipt = ToolReceipt( + invocationId = invId, + toolName = "file_write", + exitCode = 0, + outputSummary = "wrote 3 lines", + durationMs = 10L, + tier = Tier.T2, + timestamp = timestamp, + ) + val event = storedEvent( + ToolExecutionCompletedEvent( + invocationId = invId, sessionId = sessionId, toolName = "file_write", receipt = receipt, + ), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.ToolCompleted(sessionId, "file_write", "wrote 3 lines"), result) + } + + @Test + fun `ToolExecutionFailedEvent maps to ToolFailed`(): Unit = runTest { + val event = storedEvent( + ToolExecutionFailedEvent( + invocationId = ToolInvocationId("inv-1"), + sessionId = sessionId, + toolName = "file_write", + reason = "disk full", + ), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.ToolFailed(sessionId, "file_write", "disk full"), result) + } + + @Test + fun `ToolExecutionRejectedEvent maps to ToolRejected`(): Unit = runTest { + val event = storedEvent( + ToolExecutionRejectedEvent( + invocationId = ToolInvocationId("inv-1"), + sessionId = sessionId, + toolName = "file_write", + tier = Tier.T3, + reason = "policy denied", + ), + ) + val result = domainEventToServerMessage(event, noopStore) + assertEquals(ServerMessage.ToolRejected(sessionId, "file_write", "policy denied"), result) + } + + @Test + fun `ApprovalRequestedEvent maps to ApprovalRequired`(): Unit = runTest { + val requestId = ApprovalRequestId("req-1") + val event = storedEvent(ApprovalRequestedEvent( + requestId = requestId, + tier = Tier.T3, + validationReportId = ValidationReportId("vr-1"), + riskSummaryId = null, + sessionId = sessionId, + stageId = stageId, + projectId = null, + )) + val result = domainEventToServerMessage(event, noopStore) as ServerMessage.ApprovalRequired + assertEquals(requestId, result.requestId) + assertEquals(Tier.T3, result.tier) + assertNull(result.toolName) + assertNull(result.preview) + } + + @Test + fun `unmapped event returns null`(): Unit = runTest { + val event = storedEvent(SessionStartedEvent(sessionId = sessionId)) + val result = domainEventToServerMessage(event, noopStore) + assertNull(result) + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt new file mode 100644 index 00000000..7b29d5a2 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt @@ -0,0 +1,135 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import kotlinx.datetime.Instant +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class SessionEventBridgeTest { + + private val sessionId = SessionId("session-1") + private val stageId = StageId("stage-1") + private val timestamp = Instant.parse("2026-01-01T00:00:00Z") + + private val noopArtifactStore: ArtifactStore = object : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("art-1") + override suspend fun get(id: ArtifactId): ByteArray? = null + override suspend fun flushBefore(commit: suspend () -> Unit) = commit() + } + + private fun storedEvent(payload: EventPayload, seq: Long = 1L): StoredEvent = + StoredEvent( + metadata = EventMetadata( + eventId = EventId("evt-$seq"), + sessionId = sessionId, + timestamp = timestamp, + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = seq, + payload = payload, + ) + + private fun fakeEventStore( + allEventsList: List = emptyList(), + liveFlow: Flow = flow {}, + ): EventStore = object : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = allEventsList + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = allEventsList + override fun lastSequence(sessionId: SessionId): Long? = null + override fun subscribe(sessionId: SessionId): Flow = liveFlow + override fun allEvents(): Sequence = allEventsList.asSequence() + } + + @Test + fun `replaySnapshot sends mapped messages for all events`() = runTest { + val events = listOf( + storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L), + storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L), + ) + val store = fakeEventStore(allEventsList = events) + val sent = mutableListOf() + val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) } + + bridge.replaySnapshot() + + assertEquals(2, sent.size) + assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), sent[0]) + assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1]) + } + + @Test + fun `replaySnapshot skips unmapped events`() = runTest { + val events = listOf( + storedEvent(OrchestrationResumedEvent(sessionId, stageId), seq = 1L), + storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L), + ) + val store = fakeEventStore(allEventsList = events) + val sent = mutableListOf() + val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) } + + bridge.replaySnapshot() + + assertEquals(1, sent.size) + assertEquals(ServerMessage.SessionCompleted(sessionId), sent[0]) + } + + @Test + fun `streamLive sends mapped messages from live flow`() = runTest { + val events = listOf( + storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L), + storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L), + ) + val liveFlow: Flow = flow { events.forEach { emit(it) } } + val store = fakeEventStore(liveFlow = liveFlow) + val sent = mutableListOf() + val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) } + + bridge.streamLive(sessionId) + + assertEquals(2, sent.size) + assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), sent[0]) + assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1]) + } + + @Test + fun `streamLive cancellation stops collection`() = runTest { + val channel = Channel(Channel.UNLIMITED) + val liveFlow: Flow = flow { + for (event in channel) emit(event) + } + val store = fakeEventStore(liveFlow = liveFlow) + val sent = mutableListOf() + val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) } + + channel.send(storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L)) + + val job = launch { bridge.streamLive(sessionId) } + channel.send(storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L)) + channel.close() + job.join() + + assertEquals(2, sent.size) + } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt index d9f5b070..b165fd15 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt @@ -4,6 +4,7 @@ import com.correx.apps.tui.components.activeSessionWidget import com.correx.apps.tui.components.approvalPanelWidget import com.correx.apps.tui.components.filteredSessions import com.correx.apps.tui.components.inputBarWidget +import com.correx.apps.tui.components.routerPanelWidget import com.correx.apps.tui.components.sessionListWidget import com.correx.apps.tui.components.statusBarWidget import com.correx.apps.tui.input.Action @@ -39,6 +40,7 @@ private const val STATUS_HEIGHT = 3 private const val ACTIVE_SESSION_HEIGHT = 5 private const val APPROVAL_HEIGHT = 6 private const val INPUT_HEIGHT = 3 +private const val ROUTER_HEIGHT = 5 private const val KEYBINDS_HEIGHT = 3 private const val KEYBINDS_TEXT = "n new c cancel a approve r reject s steer / filter q quit" @@ -105,6 +107,7 @@ private fun render(frame: Frame, state: TuiState, listState: ListState) { add(Constraint.length(STATUS_HEIGHT)) add(Constraint.fill()) add(Constraint.length(ACTIVE_SESSION_HEIGHT)) + add(Constraint.length(ROUTER_HEIGHT)) if (hasApproval) add(Constraint.length(APPROVAL_HEIGHT)) if (hasInput) add(Constraint.length(INPUT_HEIGHT)) add(Constraint.length(KEYBINDS_HEIGHT)) @@ -126,6 +129,8 @@ private fun render(frame: Frame, state: TuiState, listState: ListState) { frame.renderWidget(activeSessionWidget(selectedSession), rects[idx++]) + frame.renderWidget(routerPanelWidget(selectedSession), rects[idx++]) + if (hasApproval) { frame.renderWidget(approvalPanelWidget(state.approval.active!!), rects[idx++]) } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt new file mode 100644 index 00000000..7c4b6c8e --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt @@ -0,0 +1,24 @@ +package com.correx.apps.tui.components + +import com.correx.apps.tui.state.SessionSummary +import dev.tamboui.style.Style +import dev.tamboui.text.Line +import dev.tamboui.text.Span +import dev.tamboui.text.Text +import dev.tamboui.widgets.block.Block +import dev.tamboui.widgets.block.BorderType +import dev.tamboui.widgets.block.Borders +import dev.tamboui.widgets.paragraph.Paragraph + +fun routerPanelWidget(session: SessionSummary?): Paragraph { + val dimStyle = Style.create().dim().gray() + val text = session?.lastResponseText?.takeIf { it.isNotEmpty() }?.let { responseText -> + Text.from(responseText.lineSequence().map { Line.from(Span.styled(it, dimStyle)) }.toList()) + } ?: Text.from(Line.from(Span.styled("(no response)", dimStyle))) + val block = Block.builder() + .title("router") + .borders(Borders.ALL) + .borderType(BorderType.ROUNDED) + .build() + return Paragraph.builder().text(text).block(block).build() +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt index b7de4c8d..7ecd999a 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt @@ -98,8 +98,16 @@ object SessionsReducer { } }, ) - is ServerMessage.StageCompleted -> touchSession(sessions, msg.sessionId.value, null, clock) - is ServerMessage.StageFailed -> touchSession(sessions, msg.sessionId.value, null, clock) + is ServerMessage.StageCompleted -> sessions.copy( + sessions = sessions.sessions.map { s -> + if (s.id == msg.sessionId.value) s.copy(currentStage = null, lastEventAt = clock()) else s + }, + ) + is ServerMessage.StageFailed -> sessions.copy( + sessions = sessions.sessions.map { s -> + if (s.id == msg.sessionId.value) s.copy(currentStage = null, lastEventAt = clock()) else s + }, + ) is ServerMessage.InferenceCompleted -> applyInferenceCompleted(sessions, msg, clock) is ServerMessage.ToolCompleted -> sessions.copy( sessions = sessions.sessions.map { s -> @@ -134,15 +142,11 @@ object SessionsReducer { private fun touchSession( sessions: SessionsState, sessionId: String, - status: String?, + status: String, clock: () -> Long, ): SessionsState = sessions.copy( sessions = sessions.sessions.map { s -> - if (s.id == sessionId) { - if (status != null) s.copy(status = status, lastEventAt = clock()) else s.copy(lastEventAt = clock()) - } else { - s - } + if (s.id == sessionId) s.copy(status = status, lastEventAt = clock()) else s }, ) } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt index c1acd6ee..64f35954 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt @@ -73,7 +73,7 @@ class TuiWsClient( } } } - }.onFailure { println("WS connect failed: ${it.message}") } + }.onFailure { } session = null _connection.tryEmit(ConnectionEvent.Disconnected) attempt++ diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt index 13e103f8..22f1c976 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt @@ -1,5 +1,6 @@ package com.correx.apps.tui.reducer +import com.correx.apps.server.protocol.PauseReason import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.tui.input.Action import com.correx.apps.tui.state.InputMode @@ -124,6 +125,34 @@ class SessionsReducerTest { assertEquals("FAILED", state.sessions[0].status) } + @Test + fun `ServerEventReceived StageCompleted clears currentStage`() { + val s = SessionsState( + sessions = listOf(session("s1").copy(currentStage = "stage-1")), + selectedId = "s1", + ) + val msg = ServerMessage.StageCompleted(sessionId = SessionId("s1"), stageId = StageId("stage-1")) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals(null, state.sessions[0].currentStage) + assertEquals(1000L, state.sessions[0].lastEventAt) + } + + @Test + fun `ServerEventReceived StageFailed clears currentStage`() { + val s = SessionsState( + sessions = listOf(session("s1").copy(currentStage = "stage-1")), + selectedId = "s1", + ) + val msg = ServerMessage.StageFailed( + sessionId = SessionId("s1"), + stageId = StageId("stage-1"), + reason = "error", + ) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals(null, state.sessions[0].currentStage) + assertEquals(1000L, state.sessions[0].lastEventAt) + } + @Test fun `CancelSelectedSession emits CancelSession effect`() { val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") @@ -140,4 +169,54 @@ class SessionsReducerTest { val (_, effects) = reduce(sessions = s, action = Action.CancelSelectedSession) assertTrue(effects.isEmpty()) } + + @Test + fun `ServerEventReceived InferenceCompleted updates lastOutput and lastResponseText`() { + val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") + val msg = ServerMessage.InferenceCompleted( + sessionId = SessionId("s1"), stageId = StageId("stage-1"), + outputSummary = "summary", responseText = "full response", + ) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals("summary", state.sessions[0].lastOutput) + assertEquals("full response", state.sessions[0].lastResponseText) + assertEquals(1000L, state.sessions[0].lastEventAt) + } + + @Test + fun `ServerEventReceived InferenceCompleted with empty responseText stores null`() { + val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") + val msg = ServerMessage.InferenceCompleted( + sessionId = SessionId("s1"), stageId = StageId("stage-1"), + outputSummary = "", responseText = "", + ) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals(null, state.sessions[0].lastResponseText) + } + + @Test + fun `ServerEventReceived StageStarted sets currentStage`() { + val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") + val msg = ServerMessage.StageStarted(sessionId = SessionId("s1"), stageId = StageId("stage-1")) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals("stage-1", state.sessions[0].currentStage) + assertEquals(1000L, state.sessions[0].lastEventAt) + } + + @Test + fun `ServerEventReceived ToolCompleted updates lastOutput`() { + val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") + val msg = ServerMessage.ToolCompleted(sessionId = SessionId("s1"), toolName = "file_write", outputSummary = "wrote 3 lines") + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals("file_write: wrote 3 lines", state.sessions[0].lastOutput) + assertEquals(1000L, state.sessions[0].lastEventAt) + } + + @Test + fun `ServerEventReceived SessionPaused with APPROVAL_PENDING sets status label`() { + val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") + val msg = ServerMessage.SessionPaused(sessionId = SessionId("s1"), reason = PauseReason.APPROVAL_PENDING) + val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) + assertEquals("PAUSED awaiting approval", state.sessions[0].status) + } }