fix: resolve four findings from code review

- Stop silently booting user from session view on completion (SessionsReducer.processSessionCompletedMessage no longer deselects) - Wrap StageToolManifest send in runCatching to survive WS disconnect after orchestrator launch in module scope - Fix navigateUp dead state: wrap to last workflow instead of -1 - Consolidate event store reads in SessionEventBridge (single read for both recentEvents and approval state), removing DefaultApprovalRepository dependency
2026-05-26 16:01:57 +04:00
parent 3d95a946a8
commit e3812330d2
5 changed files with 52 additions and 73 deletions
@@ -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)

@@ -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)
+        }
    }
 }

@@ -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<com.correx.apps.server.registry.WorkflowSummary> = 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<ServerMessage>()
-        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<ServerMessage>()
-        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<ServerMessage>()
-        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<ServerMessage>()
-        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) }
@@ -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<Tool> = 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<SessionEventBridge, Channel<ServerMessage>> {
        val received = Channel<ServerMessage>(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<StoredEvent>(extraBufferCapacity = 64)
            val store = fakeEventStore(liveFlow, emptySet(), lastGlobal = 5L)
-            val approvals = fakeApprovalRepository(store)
            val received = Channel<ServerMessage>(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<StoredEvent>(extraBufferCapacity = 16)
        val store = fakeEventStore(liveFlow, emptySet(), lastGlobal = 0L)
-        val approvals = fakeApprovalRepository(store)
        val received = Channel<ServerMessage>(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<StoredEvent>()
        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 {
@@ -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<SessionsState, List<Effect>> {
        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(