feat(server-04): fix snapshot cursor order, deterministic approvals, unconditional SnapshotComplete

Capture lastGlobalSequence() before any projection read to anchor the buffer-drain window; capture lastSequence(sessionId) before each session's projection read. SessionSnapshot.lastSequence now carries the global cursor shared across all sessions in the batch. pendingApprovals populated from filtered approval state and sorted by (timestamp, requestId) for deterministic ordering. SnapshotComplete emitted unconditionally including the zero-sessions case. Four new tests: zero-sessions, SnapshotComplete-last ordering, global cursor isolation, and per-session cursor value.
2026-05-24 23:01:21 +04:00
parent ac05ad8733
commit 14141f2f72
2 changed files with 68 additions and 10 deletions
@@ -1,5 +1,6 @@
 package com.correx.apps.server.bridge

+import com.correx.apps.server.protocol.ApprovalDto
 import com.correx.apps.server.protocol.ServerMessage
 import com.correx.apps.server.protocol.SessionStateDto
 import com.correx.core.approvals.DefaultApprovalRepository
@@ -17,20 +18,26 @@ class SessionEventBridge(
    private val send: suspend (ServerMessage) -> Unit,
 ) {
    suspend fun replaySnapshot() {
+        val lastGlobal = eventStore.lastGlobalSequence()
        eventStore.allSessionIds().forEach { sessionId ->
            val orchState = orchestrationRepository.getState(sessionId)
            if (orchState.status == OrchestrationStatus.IDLE) return@forEach

+            val lastSession = eventStore.lastSequence(sessionId) ?: 0L
+
            val approvalState = if (orchState.pendingApproval) {
                approvalRepository.getApprovalState(sessionId)
            } else null

-            val pendingRequest = approvalState?.requests?.values
-                ?.firstOrNull { req ->
-                    approvalState.decisions.values.none { it.requestId == req.id }
-                }
+            val pendingApprovals = approvalState?.requests?.values
+                ?.filter { req -> approvalState.decisions.values.none { it.requestId == req.id } }
+                ?.sortedWith(compareBy({ it.timestamp }, { it.id.value }))
+                ?.map { ApprovalDto(requestId = it.id.value, tier = it.tier.name) }
+                ?: emptyList()
+
+            val pendingRequest = approvalState?.requests?.values
+                ?.firstOrNull { req -> approvalState.decisions.values.none { it.requestId == req.id } }

-            val lastSeq = eventStore.lastSequence(sessionId) ?: 0L
            send(ServerMessage.SessionSnapshot(
                sessionId = sessionId,
                workflowId = orchState.workflowId,
@@ -47,9 +54,9 @@ class SessionEventBridge(
                    currentStageId = orchState.currentStageId?.value,
                    pauseReason = orchState.pauseReason,
                ),
-                pendingApprovals = emptyList(),
-                lastSequence = lastSeq,
-                lastSessionSequence = 0L,
+                pendingApprovals = pendingApprovals,
+                lastSequence = lastGlobal,
+                lastSessionSequence = lastSession,
            ))
        }
        send(ServerMessage.SnapshotComplete)
@@ -74,7 +74,8 @@ class SessionEventBridgeTest {
        override fun allEvents(): Sequence<StoredEvent> = allEventsList.asSequence()
        override fun allSessionIds(): Set<SessionId> = allEventsList.map { it.metadata.sessionId }.toSet()
        override fun subscribeAll(): Flow<StoredEvent> = TODO("Not needed in this test context")
-        override suspend fun lastGlobalSequence(): Long = TODO("Not needed in this test context")
+        override suspend fun lastGlobalSequence(): Long =
+            allEventsList.maxOfOrNull { it.sequence } ?: 0L
    }

    private fun activeOrchestrationRepository(
@@ -119,7 +120,7 @@ class SessionEventBridgeTest {
        val snapshot = sent[0] as ServerMessage.SessionSnapshot
        assertEquals(sessionId, snapshot.sessionId)
        assertEquals(2L, snapshot.lastSequence)
-        assertEquals(0L, snapshot.lastSessionSequence)
+        assertEquals(2L, snapshot.lastSessionSequence)
        assertEquals(ServerMessage.SnapshotComplete, sent[1])
    }

@@ -176,6 +177,56 @@ class SessionEventBridgeTest {
        )
    }

+    @Test
+    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) { sent.add(it) }
+        bridge.replaySnapshot()
+        assertEquals(1, sent.size)
+        assertEquals(ServerMessage.SnapshotComplete, sent[0])
+    }
+
+    @Test
+    fun `replaySnapshot emits SnapshotComplete as final message`() = runTest {
+        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) { sent.add(it) }
+        bridge.replaySnapshot()
+        assertEquals(ServerMessage.SnapshotComplete, sent.last())
+    }
+
+    @Test
+    fun `replaySnapshot lastSequence reflects global cursor before projection read`() = runTest {
+        val events = listOf(
+            storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
+            storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
+        )
+        val store = object : EventStore by fakeEventStore(allEventsList = events) {
+            override suspend fun lastGlobalSequence(): Long = 5L
+        }
+        val sent = mutableListOf<ServerMessage>()
+        val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository) { sent.add(it) }
+        bridge.replaySnapshot()
+        val snapshot = sent[0] as ServerMessage.SessionSnapshot
+        assertEquals(5L, snapshot.lastSequence)
+    }
+
+    @Test
+    fun `replaySnapshot lastSessionSequence equals per-session cursor`() = runTest {
+        val events = listOf(
+            storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
+            storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 3L),
+        )
+        val store = fakeEventStore(allEventsList = events)
+        val sent = mutableListOf<ServerMessage>()
+        val bridge = SessionEventBridge(store, noopArtifactStore, activeOrchestrationRepository(), approvalRepository) { sent.add(it) }
+        bridge.replaySnapshot()
+        val snapshot = sent[0] as ServerMessage.SessionSnapshot
+        assertEquals(3L, snapshot.lastSessionSequence)
+    }
+
    @Test
    fun `streamLive cancellation stops collection`() = runTest {
        val channel = Channel<StoredEvent>(Channel.UNLIMITED)