fix(approval,stream): resolve-after-submit + real tier (S1), persisted sessionSequence on live frames (S2)

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.

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

apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt

@@ -23,6 +23,7 @@ open class ApprovalCoordinator(
     private val globalClients: MutableSet<DefaultWebSocketServerSession> = ConcurrentHashMap.newKeySet()
     private val resolved: ConcurrentHashMap<ApprovalRequestId, Boolean> = ConcurrentHashMap()
     private val requestSessions: ConcurrentHashMap<ApprovalRequestId, SessionId> = ConcurrentHashMap()
+    private val requestTiers: ConcurrentHashMap<ApprovalRequestId, Tier> = 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,

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)

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<String, Long>()
-
     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 {

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<Pair<ApprovalRequestId, DomainApprovalDecision>>()
         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"),