From 14141f2f72265a16d2c8016e601a39792643273d Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 24 May 2026 23:01:21 +0400 Subject: [PATCH] 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. --- .../apps/server/bridge/SessionEventBridge.kt | 23 +++++--- .../server/bridge/SessionEventBridgeTest.kt | 55 ++++++++++++++++++- 2 files changed, 68 insertions(+), 10 deletions(-) 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 b29ab297..44333661 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 @@ -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) 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 c25ca849..e8cb5258 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 @@ -74,7 +74,8 @@ class SessionEventBridgeTest { override fun allEvents(): Sequence = allEventsList.asSequence() override fun allSessionIds(): Set = allEventsList.map { it.metadata.sessionId }.toSet() override fun subscribeAll(): Flow = 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() + 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() + 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() + 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() + 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(Channel.UNLIMITED)