From e2f44387e10309b80fdf4136419854146103a0bf Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 12 Jul 2026 12:11:49 +0400 Subject: [PATCH] fix(approval,stream): resolve-after-submit + real tier (S1), persisted sessionSequence on live frames (S2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit S1: ApprovalCoordinator marked a request resolved before submitApprovalDecision, so a throwing submit left it permanently unanswerable; now resolve only on success and clear the flag on failure so the client can retry. Recorded decision also used a hardcoded Tier.T2 regardless of the real tier — now tracked per request. S2: streamGlobal numbered live frames from a per-connection counter starting at 1, colliding with the snapshot's lastSessionSequence (MAX(session_sequence)) after reconnect and causing the TUI dedup filter to drop/misorder frames; now uses each event's own persisted sessionSequence. Approval broadcast no longer sends 0. --- .../com/correx/apps/server/ServerModule.kt | 6 ++- .../server/approval/ApprovalCoordinator.kt | 23 +++++++--- .../apps/server/bridge/SessionEventBridge.kt | 5 +-- .../apps/server/ws/GlobalStreamHandler.kt | 13 +++--- .../approval/ApprovalCoordinatorWiringTest.kt | 45 ++++++++++++++++++- 5 files changed, 72 insertions(+), 20 deletions(-) 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 c0bacf8b..5b6c6fd8 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 @@ -186,7 +186,9 @@ class ServerModule( resumeAbandonedSessions() subscriptionJob = eventStore.subscribeAll() .filter { it.payload is ApprovalRequestedEvent } - .onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) } + .onEach { + approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent, it.sessionSequence) + } .launchIn(moduleScope) // When a PAUSED session's approval is resolved without a live orchestrator coroutine @@ -595,7 +597,7 @@ class ServerModule( } approvalState.requests.values .filter { req -> approvalState.decisions.values.none { it.requestId == req.id } } - .forEach { req -> approvalCoordinator.registerPendingRequest(req.id, sessionId) } + .forEach { req -> approvalCoordinator.registerPendingRequest(req.id, sessionId, req.tier) } } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt index e249b678..4c2c98cf 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -23,6 +23,7 @@ open class ApprovalCoordinator( private val globalClients: MutableSet = ConcurrentHashMap.newKeySet() private val resolved: ConcurrentHashMap = ConcurrentHashMap() private val requestSessions: ConcurrentHashMap = ConcurrentHashMap() + private val requestTiers: ConcurrentHashMap = ConcurrentHashMap() fun registerClient(sessionId: SessionId, session: DefaultWebSocketServerSession) { sessionClients.getOrPut(sessionId) { ConcurrentHashMap.newKeySet() }.add(session) @@ -40,8 +41,9 @@ open class ApprovalCoordinator( globalClients.remove(session) } - suspend fun onApprovalRequested(event: ApprovalRequestedEvent) { + suspend fun onApprovalRequested(event: ApprovalRequestedEvent, sessionSequence: Long = 0L) { requestSessions[event.requestId] = event.sessionId + requestTiers[event.requestId] = event.tier val msg = ServerMessage.ApprovalRequired( sessionId = event.sessionId, requestId = event.requestId, @@ -55,7 +57,7 @@ open class ApprovalCoordinator( toolName = event.toolName, preview = event.preview, sequence = 0L, - sessionSequence = 0L, + sessionSequence = sessionSequence, ) broadcast(event.sessionId, msg) } @@ -75,8 +77,9 @@ open class ApprovalCoordinator( * during snapshot replay. This makes [lookupSession] work for approvals that were created * before the current connection's live stream started. */ - fun registerPendingRequest(requestId: ApprovalRequestId, sessionId: SessionId) { + fun registerPendingRequest(requestId: ApprovalRequestId, sessionId: SessionId, tier: Tier) { requestSessions[requestId] = sessionId + requestTiers[requestId] = tier } open suspend fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { @@ -87,12 +90,20 @@ open class ApprovalCoordinator( sessionSequence = null, ) } - requestSessions.remove(msg.requestId) - val domain = msg.toDomain(sessionId, null, Tier.T2) + val tier = requestTiers[msg.requestId] ?: Tier.T2 + val domain = msg.toDomain(sessionId, null, tier) return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) } .fold( - onSuccess = { null }, + onSuccess = { + // Only retire the request once the decision was actually recorded. If submit + // throws, we clear the resolved flag so the client can retry instead of the + // request being permanently unanswerable. + requestSessions.remove(msg.requestId) + requestTiers.remove(msg.requestId) + null + }, onFailure = { + resolved.remove(msg.requestId) ServerMessage.ProtocolError( message = it.message ?: "Unknown error", sequence = 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 index 0eee9e72..c0abb96b 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 @@ -43,7 +43,6 @@ import com.correx.core.events.orchestration.OrchestrationStatus import com.correx.core.events.types.EventId import kotlinx.datetime.Clock import com.correx.core.events.stores.EventStore -import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId import com.correx.core.kernel.orchestration.OrchestrationRepository import com.correx.core.tools.registry.ToolRegistry @@ -164,8 +163,8 @@ class SessionEventBridge( // Re-register pending approvals so the ApprovalCoordinator can route responses // from clients that connected after the ApprovalRequestedEvent was emitted. - pendingApprovals.forEach { dto -> - approvalCoordinator?.registerPendingRequest(ApprovalRequestId(dto.requestId), sessionId) + pendingApprovalRequests.forEach { req -> + approvalCoordinator?.registerPendingRequest(req.id, sessionId, req.tier) } val toolRecords = rebuildTools(events) 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 ca08b3b8..44aa787d 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 @@ -830,18 +830,15 @@ internal suspend fun streamGlobal( signaled.collect { buffer.send(it) } } - // Per-session sequence counters so live events carry the correct sessionSequence - // and are not silently dropped by the TUI's SnapshotPhaseReducer dedup filter. - val sessionSequences = mutableMapOf() - try { subscribed.await() bridge.replaySnapshot() for (event in buffer) { - val sid = event.metadata.sessionId.value - val seq = sessionSequences.getOrDefault(sid, 0L) + 1 - sessionSequences[sid] = seq - val msg = mapper.map(event, sessionSequence = seq) ?: continue + // Use the event's own persisted sessionSequence, not a per-connection counter. The + // snapshot advertises lastSessionSequence = MAX(session_sequence); a counter restarting + // at 1 per connection would collide with that after reconnect and the TUI's dedup/ + // ordering filter (keyed on sessionSequence) would silently drop or misorder frames. + val msg = mapper.map(event, sessionSequence = event.sessionSequence) ?: continue sendFrame(msg) } } finally { diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt index 5e83c312..2f4e0656 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt @@ -60,13 +60,56 @@ class ApprovalCoordinatorWiringTest { scope.cancel() } - private class RecordingGateway : ApprovalGateway { + private class RecordingGateway(@Volatile var failNext: Boolean = false) : ApprovalGateway { val submissions = CopyOnWriteArrayList>() override suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) { + if (failNext) { + failNext = false + error("submit boom") + } submissions.add(requestId to decision) } } + private fun approvalEvent(tier: Tier) = ApprovalRequestedEvent( + requestId = requestId, + tier = tier, + validationReportId = ValidationReportId("vr-1"), + riskSummaryId = null, + sessionId = sessionId, + stageId = null, + projectId = null, + ) + + @Test + fun `submit failure keeps request answerable and retry succeeds`(): Unit = runBlocking { + val gateway = RecordingGateway(failNext = true) + val coord = ApprovalCoordinator(gateway) + coord.onApprovalRequested(approvalEvent(Tier.T2)) + val msg = ClientMessage.ApprovalResponse(requestId, ApprovalDecision.APPROVE, steeringNote = null) + + val first = coord.handleResponse(msg, sessionId) + assertInstanceOf(ServerMessage.ProtocolError::class.java, first) + assertTrue(gateway.submissions.isEmpty()) + + // Retry must not be blocked by a stale resolved flag. + val second = coord.handleResponse(msg, sessionId) + assertNull(second) + assertEquals(1, gateway.submissions.size) + } + + @Test + fun `recorded decision carries the request's real tier`(): Unit = runBlocking { + val gateway = RecordingGateway() + val coord = ApprovalCoordinator(gateway) + coord.onApprovalRequested(approvalEvent(Tier.T3)) + val msg = ClientMessage.ApprovalResponse(requestId, ApprovalDecision.APPROVE, steeringNote = null) + + coord.handleResponse(msg, sessionId) + + assertEquals(Tier.T3, gateway.submissions[0].second.tier) + } + private fun storedEvent(payload: EventPayload, seq: Long): StoredEvent = StoredEvent( metadata = EventMetadata( eventId = EventId("evt-$seq"),