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 index 7678400b..93e359af 100644 --- 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 @@ -8,7 +8,9 @@ import com.correx.apps.server.protocol.SessionStateDto import com.correx.apps.server.protocol.StageToolDecl import com.correx.apps.server.protocol.ToolDecl import com.correx.apps.server.registry.WorkflowRegistry -import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.model.ApprovalState import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.InferenceStartedEvent @@ -35,12 +37,13 @@ class SessionEventBridge( private val eventStore: EventStore, private val artifactStore: ArtifactStore, private val orchestrationRepository: OrchestrationRepository, - private val approvalRepository: DefaultApprovalRepository, private val workflowRegistry: WorkflowRegistry, private val toolRegistry: ToolRegistry, private val approvalCoordinator: ApprovalCoordinator? = null, private val send: suspend (ServerMessage) -> Unit, ) { + private val approvalProjector = ApprovalProjector(DefaultApprovalReducer()) + suspend fun replaySnapshot() { val lastGlobal = eventStore.lastGlobalSequence() eventStore.allSessionIds().forEach { sessionId -> @@ -49,8 +52,14 @@ class SessionEventBridge( val lastSession = eventStore.lastSequence(sessionId) ?: 0L - val approvalState = if (orchState.pendingApproval) { - approvalRepository.getApprovalState(sessionId) + // Single read from the event store — avoids inconsistency between + // approval state and recentEvents with a cursor-based store. + val events = eventStore.read(sessionId) + + val approvalState: ApprovalState? = if (orchState.pendingApproval) { + events.fold(approvalProjector.initial()) { state: ApprovalState, event: StoredEvent -> + approvalProjector.apply(state, event) + } } else null val pendingApprovals = approvalState?.requests?.values @@ -66,7 +75,7 @@ class SessionEventBridge( } ?: emptyList() - val recentEvents = eventStore.read(sessionId) + val recentEvents = events .mapNotNull { eventToEntry(it) } .takeLast(7) 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 fc7d2c50..6fc2a38d 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 @@ -51,7 +51,6 @@ class GlobalStreamHandler(private val module: ServerModule) { eventStore = module.eventStore, artifactStore = module.artifactStore, orchestrationRepository = module.orchestrationRepository, - approvalRepository = module.approvalRepository, workflowRegistry = module.workflowRegistry, toolRegistry = module.toolRegistry, send = sendFrame, @@ -217,30 +216,35 @@ class GlobalStreamHandler(private val module: ServerModule) { ) } } - val started = ServerMessage.SessionStarted( - sessionId = sessionId, - workflowId = msg.workflowId, - sequence = 0L, - sessionSequence = 0L, - ) - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) - // Emit StageToolManifest for the newly created session (ARCH-TOOL-1) - val stages = graph.stages.map { (stageId, stageConfig) -> - val tools = stageConfig.allowedTools.mapNotNull { toolName -> - module.toolRegistry.resolve(toolName)?.let { tool -> - ToolDecl(name = tool.name, tier = tool.tier.level) - } - } - StageToolDecl(stageId = stageId.value, tools = tools) - } - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage( - ServerMessage.StageToolManifest( + runCatching { + val started = ServerMessage.SessionStarted( sessionId = sessionId, workflowId = msg.workflowId, - stages = stages, - ), - ))) + sequence = 0L, + sessionSequence = 0L, + ) + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) + + // Emit StageToolManifest for the newly created session (ARCH-TOOL-1) + val stages = graph.stages.map { (stageId, stageConfig) -> + val tools = stageConfig.allowedTools.mapNotNull { toolName -> + module.toolRegistry.resolve(toolName)?.let { tool -> + ToolDecl(name = tool.name, tier = tool.tier.level) + } + } + StageToolDecl(stageId = stageId.value, tools = tools) + } + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage( + ServerMessage.StageToolManifest( + sessionId = sessionId, + workflowId = msg.workflowId, + stages = stages, + ), + ))) + }.onFailure { ex -> + log.warn("failed to send session start messages for session={}: {}", sessionId.value, ex.message) + } } } 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 index e9f059f4..8361e737 100644 --- 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 @@ -3,9 +3,6 @@ package com.correx.apps.server.bridge import com.correx.apps.server.protocol.EventEntryDto import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.server.registry.WorkflowRegistry -import com.correx.core.approvals.ApprovalProjector -import com.correx.core.approvals.DefaultApprovalReducer -import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.EventPayload @@ -25,7 +22,6 @@ import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.events.types.TransitionId import com.correx.core.kernel.orchestration.OrchestrationRepository -import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.tools.contract.Tool import com.correx.core.tools.registry.ToolRegistry import com.correx.core.sessions.projections.replay.EventReplayer @@ -98,13 +94,6 @@ class SessionEventBridgeTest { }, ) - private val approvalRepository = DefaultApprovalRepository( - DefaultEventReplayer( - fakeEventStore(), - ApprovalProjector(DefaultApprovalReducer()), - ), - ) - private val noopWorkflowRegistry = object : WorkflowRegistry { override fun listAll(): List = emptyList() override fun find(workflowId: String): com.correx.core.transitions.graph.WorkflowGraph? = null @@ -128,7 +117,6 @@ class SessionEventBridgeTest { store, noopArtifactStore, orchRepo, - approvalRepository, noopWorkflowRegistry, noopToolRegistry, ) { sent.add(it) } @@ -158,7 +146,6 @@ class SessionEventBridgeTest { store, noopArtifactStore, orchRepo, - approvalRepository, noopWorkflowRegistry, noopToolRegistry, ) { sent.add(it) } @@ -187,7 +174,6 @@ class SessionEventBridgeTest { store, noopArtifactStore, orchRepo, - approvalRepository, noopWorkflowRegistry, noopToolRegistry, ) { sent.add(it) } @@ -213,7 +199,6 @@ class SessionEventBridgeTest { store, noopArtifactStore, activeOrchestrationRepository(), - approvalRepository, noopWorkflowRegistry, noopToolRegistry, ) { sent.add(it) } @@ -235,7 +220,7 @@ class SessionEventBridgeTest { fun `replaySnapshot with no sessions emits only SnapshotComplete`() = runTest { val store = fakeEventStore(allEventsList = emptyList()) val sent = mutableListOf() - val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository, noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } + val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } bridge.replaySnapshot() assertEquals(1, sent.size) assertEquals(ServerMessage.SnapshotComplete, sent[0]) @@ -246,7 +231,7 @@ class SessionEventBridgeTest { val events = listOf(storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L)) val store = fakeEventStore(allEventsList = events) val sent = mutableListOf() - val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository, noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } + val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } bridge.replaySnapshot() assertEquals(ServerMessage.SnapshotComplete, sent.last()) } @@ -261,7 +246,7 @@ class SessionEventBridgeTest { override suspend fun lastGlobalSequence(): Long = 5L } val sent = mutableListOf() - val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository, noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } + val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } bridge.replaySnapshot() val snapshot = sent[0] as ServerMessage.SessionSnapshot assertEquals(5L, snapshot.lastSequence) @@ -275,7 +260,7 @@ class SessionEventBridgeTest { ) val store = fakeEventStore(allEventsList = events) val sent = mutableListOf() - val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository, noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } + val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { sent.add(it) } bridge.replaySnapshot() val snapshot = sent[0] as ServerMessage.SessionSnapshot assertEquals(3L, snapshot.lastSessionSequence) @@ -293,7 +278,6 @@ class SessionEventBridgeTest { store, noopArtifactStore, activeOrchestrationRepository(), - approvalRepository, noopWorkflowRegistry, noopToolRegistry, ) { sent.add(it) } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt index 75d2b6ea..898b6728 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt @@ -4,9 +4,6 @@ import com.correx.apps.server.bridge.DomainEventMapper import com.correx.apps.server.bridge.SessionEventBridge import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.server.registry.WorkflowRegistry -import com.correx.core.approvals.ApprovalProjector -import com.correx.core.approvals.DefaultApprovalReducer -import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.EventPayload @@ -21,7 +18,6 @@ import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.kernel.orchestration.OrchestrationRepository -import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.tools.contract.Tool import com.correx.core.tools.registry.ToolRegistry import com.correx.core.sessions.projections.replay.EventReplayer @@ -116,14 +112,6 @@ class GlobalStreamHandlerTest { override fun all(): List = emptyList() } - private fun fakeApprovalRepository(store: EventStore): DefaultApprovalRepository = - DefaultApprovalRepository( - DefaultEventReplayer( - store, - ApprovalProjector(DefaultApprovalReducer()), - ), - ) - /** * Creates a bridge and a unified send channel so that snapshot messages and live messages * both flow to the same [received] channel. @@ -133,8 +121,7 @@ class GlobalStreamHandlerTest { orchRepo: OrchestrationRepository, ): Pair> { val received = Channel(Channel.UNLIMITED) - val approvals = fakeApprovalRepository(store) - val bridge = SessionEventBridge(store, noopArtifactStore, orchRepo, approvals, noopWorkflowRegistry, noopToolRegistry) { received.send(it) } + val bridge = SessionEventBridge(store, noopArtifactStore, orchRepo, noopWorkflowRegistry, noopToolRegistry) { received.send(it) } return bridge to received } @@ -179,9 +166,8 @@ class GlobalStreamHandlerTest { repeat(100) { val liveFlow = MutableSharedFlow(extraBufferCapacity = 64) val store = fakeEventStore(liveFlow, emptySet(), lastGlobal = 5L) - val approvals = fakeApprovalRepository(store) val received = Channel(Channel.UNLIMITED) - val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), approvals, noopWorkflowRegistry, noopToolRegistry) { + val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { received.send(it) } val mapper = DomainEventMapper(noopArtifactStore) @@ -222,9 +208,8 @@ class GlobalStreamHandlerTest { fun `empty - zero sessions - receives SnapshotComplete then live events`() = runTest { val liveFlow = MutableSharedFlow(extraBufferCapacity = 16) val store = fakeEventStore(liveFlow, emptySet(), lastGlobal = 0L) - val approvals = fakeApprovalRepository(store) val received = Channel(Channel.UNLIMITED) - val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), approvals, noopWorkflowRegistry, noopToolRegistry) { + val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { received.send(it) } val mapper = DomainEventMapper(noopArtifactStore) @@ -272,8 +257,7 @@ class GlobalStreamHandlerTest { return 0L } } - val approvals = fakeApprovalRepository(store) - val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), approvals, noopWorkflowRegistry, noopToolRegistry) { } + val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { } val mapper = DomainEventMapper(noopArtifactStore) val job = launch { @@ -290,8 +274,7 @@ class GlobalStreamHandlerTest { fun `shutdown - cancel outer scope ends subscription with no leaks`() = runTest { val liveFlow = MutableSharedFlow() val store = fakeEventStore(liveFlow, emptySet(), lastGlobal = 0L) - val approvals = fakeApprovalRepository(store) - val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), approvals, noopWorkflowRegistry, noopToolRegistry) { } + val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), noopWorkflowRegistry, noopToolRegistry) { } val mapper = DomainEventMapper(noopArtifactStore) val job = launch { 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 3cf574b0..e846f004 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 @@ -105,12 +105,12 @@ object SessionsReducer { return if (wfi > 0) { sessions.copy(selectedWorkflowIndex = wfi - 1) } else { - // Move back to last session, or stay at first workflow if no sessions + // Move back to last session, or wrap to last workflow if no sessions val list = filteredSessions(sessions) if (list.isNotEmpty()) { sessions.copy(selectedWorkflowIndex = -1, selectedId = list.last().id) } else { - sessions.copy(selectedWorkflowIndex = -1) + sessions.copy(selectedWorkflowIndex = sessions.workflows.lastIndex) } } } @@ -487,8 +487,7 @@ object SessionsReducer { clock: () -> Long, ): Pair> { val result = touchSession(sessions, msg.sessionId.value, "COMPLETED", clock) - val cleared = if (msg.sessionId.value == result.selectedId) result.copy(selectedId = null) else result - return cleared to emptyList() + return result to emptyList() } private fun processSessionPausedMessage(