fix(narration): drop stale pause narrations once approval resolves (G)

Pause narrations are tagged with a pauseKey; a per-session pending-pause set gates
the lane worker so a resolved/resumed pause is skipped instead of blending with the
current status. ApprovalDecisionResolved drops the specific request's pause;
OrchestrationResumed clears all session pauses. Replay-safe (no recorded events change).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt

@@ -1,8 +1,10 @@
 package com.correx.apps.server.narration
 
+import com.correx.core.events.events.ApprovalDecisionResolvedEvent
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.ExecutionPlanRejectedEvent
 import com.correx.core.events.events.OrchestrationPausedEvent
+import com.correx.core.events.events.OrchestrationResumedEvent
 import com.correx.core.events.events.StageCompletedEvent
 import com.correx.core.events.events.StageFailedEvent
 import com.correx.core.events.events.StoredEvent
@@ -20,6 +22,7 @@ import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
 import org.slf4j.LoggerFactory
+import java.util.Collections
 import java.util.concurrent.ConcurrentHashMap
 
 class NarrationSubscriber(
@@ -28,7 +31,27 @@ class NarrationSubscriber(
     private val scope: CoroutineScope,
     private val maxPerRun: Int = DEFAULT_MAX_PER_RUN,
 ) {
-    private data class SessionLane(val channel: Channel<NarrationTrigger>, var used: Int)
+    /**
+     * A queued narration plus the pause-trigger bookkeeping needed to drop it if the
+     * pause it described resolves before the lane worker delivers it.
+     *
+     * [pauseKey] is non-null only for pause-trigger narrations. It is the resolving signal's
+     * scope: the originating approval request id when known (per-request supersession), or a
+     * session-wide sentinel for non-approval pauses (per-session supersession).
+     */
+    private data class QueuedNarration(val trigger: NarrationTrigger, val pauseKey: String?)
+
+    /**
+     * Per-session lane. [pendingPauses] holds the pause keys still queued (not yet delivered
+     * by the worker). When an approval resolves or the orchestration resumes, the matching
+     * keys are removed so the worker skips the now-stale pause narration instead of blending
+     * it with the current status.
+     */
+    private data class SessionLane(
+        val channel: Channel<QueuedNarration>,
+        var used: Int,
+        val pendingPauses: MutableSet<String> = Collections.newSetFromMap(ConcurrentHashMap()),
+    )
 
     private val lanes = ConcurrentHashMap<String, SessionLane>()
 
@@ -50,6 +73,7 @@ class NarrationSubscriber(
                     instruction = approvalInstruction(p),
                     stageId = p.stageId?.value,
                 ),
+                pauseKey = pauseKeyForRequest(p.requestId.value),
             )
             is StageCompletedEvent -> enqueue(
                 sid,
@@ -100,9 +124,16 @@ class NarrationSubscriber(
                             instruction = "Orchestration paused in stage ${p.stageId.value}: ${p.reason}. Inform the user.",
                             stageId = p.stageId.value,
                         ),
+                        pauseKey = SESSION_PAUSE_KEY,
                     )
                 }
             }
+            // An approval resolving makes its queued pause narration stale: drop it (scoped to
+            // that request) so a resolved pause never narrates as if still pending.
+            is ApprovalDecisionResolvedEvent -> dropPendingPause(sid, pauseKeyForRequest(p.requestId.value))
+            // Resuming supersedes every pause still queued for the session (approval-pending or
+            // not): the lane is live again, so any pending "paused" narration is stale.
+            is OrchestrationResumedEvent -> dropAllPendingPauses(sid)
             else -> Unit
         }
     }
@@ -122,20 +153,46 @@ class NarrationSubscriber(
         }
     }
 
-    private fun enqueue(sessionId: SessionId, trigger: NarrationTrigger) {
+    private fun enqueue(sessionId: SessionId, trigger: NarrationTrigger, pauseKey: String? = null) {
         val lane = lanes.computeIfAbsent(sessionId.value) { startLane(sessionId) }
         if (lane.used >= maxPerRun) {
             log.debug("narration budget reached for session {}; skipping {}", sessionId.value, trigger.kind)
             return
         }
         lane.used += 1
-        lane.channel.trySend(trigger)
+        if (pauseKey != null) {
+            lane.pendingPauses.add(pauseKey)
+        }
+        lane.channel.trySend(QueuedNarration(trigger, pauseKey))
+    }
+
+    /** Marks the pause for [pauseKey] resolved so the lane worker skips it if still queued. */
+    private fun dropPendingPause(sessionId: SessionId, pauseKey: String) {
+        lanes[sessionId.value]?.pendingPauses?.remove(pauseKey)
+    }
+
+    /** Marks every queued pause for the session resolved (used on resume). */
+    private fun dropAllPendingPauses(sessionId: SessionId) {
+        lanes[sessionId.value]?.pendingPauses?.clear()
     }
 
     private fun startLane(sessionId: SessionId): SessionLane {
-        val channel = Channel<NarrationTrigger>(capacity = Channel.UNLIMITED)
+        val channel = Channel<QueuedNarration>(capacity = Channel.UNLIMITED)
+        val lane = SessionLane(channel, used = 0)
         scope.launch {
-            for (trigger in channel) {
+            for (queued in channel) {
+                // A pause whose key was dropped (approval resolved / resumed) is stale: skip it
+                // rather than narrate a paused state that no longer holds. removeIf-style check:
+                // claim the key so a re-enqueued pause with the same id is not also dropped.
+                if (queued.pauseKey != null && !lane.pendingPauses.remove(queued.pauseKey)) {
+                    log.debug(
+                        "skipping stale pause narration for session {} (key {})",
+                        sessionId.value,
+                        queued.pauseKey,
+                    )
+                    continue
+                }
+                val trigger = queued.trigger
                 runCatching { routerFacade.narrate(sessionId, trigger) }
                     .onFailure { e ->
                         if (e is CancellationException) throw e
@@ -148,12 +205,17 @@ class NarrationSubscriber(
                     }
             }
         }
-        return SessionLane(channel, used = 0)
+        return lane
     }
 
     companion object {
         private const val DEFAULT_MAX_PER_RUN = 100
         private const val MAX_PREVIEW_CHARS = 800
+
+        /** Supersession scope for pauses with no approval request (e.g. USER_REQUESTED). */
+        private const val SESSION_PAUSE_KEY = "__session_pause__"
         private val log = LoggerFactory.getLogger(NarrationSubscriber::class.java)
+
+        private fun pauseKeyForRequest(requestId: String): String = "req:$requestId"
     }
 }

apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt

@@ -1,18 +1,23 @@
 package com.correx.apps.server.narration
 
+import com.correx.core.approvals.ApprovalOutcome
+import com.correx.core.approvals.ApprovalStatus
 import com.correx.core.approvals.Tier
+import com.correx.core.events.events.ApprovalDecisionResolvedEvent
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.EventMetadata
 import com.correx.core.events.events.EventPayload
 import com.correx.core.events.events.ExecutionPlanRejectedEvent
 import com.correx.core.events.events.NewEvent
 import com.correx.core.events.events.OrchestrationPausedEvent
+import com.correx.core.events.events.OrchestrationResumedEvent
 import com.correx.core.events.events.StageCompletedEvent
 import com.correx.core.events.events.StageFailedEvent
 import com.correx.core.events.events.StoredEvent
 import com.correx.core.events.events.WorkflowCompletedEvent
 import com.correx.core.events.events.WorkflowStartedEvent
 import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.ApprovalDecisionId
 import com.correx.core.events.types.ApprovalRequestId
 import com.correx.core.events.types.EventId
 import com.correx.core.events.types.SessionId
@@ -24,10 +29,12 @@ import com.correx.core.router.ChatMode
 import com.correx.core.router.RouterFacade
 import com.correx.core.router.model.NarrationTrigger
 import com.correx.core.router.model.RouterResponse
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.runBlocking
@@ -231,4 +238,133 @@ class NarrationSubscriberTest {
             "instruction must include the source",
         )
     }
+
+    /**
+     * Router facade that blocks the lane worker on the first narration until [release] completes,
+     * letting the test enqueue further events (and resolve the pause) while the pause narration is
+     * still pending in the channel.
+     */
+    private class GatingRouterFacade : RouterFacade {
+        val calls = CopyOnWriteArrayList<NarrationTrigger>()
+        val release = CompletableDeferred<Unit>()
+        // Signals that the worker has pulled the first (warmup) trigger and is now blocked,
+        // so any narrations enqueued afterwards are guaranteed to still be queued.
+        val blocked = CompletableDeferred<Unit>()
+        private var first = true
+
+        override suspend fun onUserInput(sessionId: SessionId, input: String, mode: ChatMode): RouterResponse =
+            error("unused in narration tests")
+
+        override suspend fun narrate(sessionId: SessionId, trigger: NarrationTrigger) {
+            if (first) {
+                first = false
+                blocked.complete(Unit)
+                release.await()
+            }
+            calls.add(trigger)
+        }
+    }
+
+    private fun approvalRequested(requestId: String): ApprovalRequestedEvent = ApprovalRequestedEvent(
+        requestId = ApprovalRequestId(requestId),
+        tier = Tier.T2,
+        validationReportId = ValidationReportId("vr-1"),
+        riskSummaryId = null,
+        sessionId = sessionId,
+        stageId = StageId("write_script"),
+        projectId = null,
+        toolName = "file_write",
+        preview = "--- a/x.sh\n+++ b/x.sh",
+    )
+
+    private fun approvalResolved(requestId: String): ApprovalDecisionResolvedEvent = ApprovalDecisionResolvedEvent(
+        decisionId = ApprovalDecisionId("dec-1"),
+        requestId = ApprovalRequestId(requestId),
+        outcome = ApprovalOutcome.APPROVED,
+        status = ApprovalStatus.COMPLETED,
+        tier = Tier.T2,
+        resolutionTimestamp = timestamp,
+        reason = null,
+    )
+
+    @Test
+    fun `resolved approval drops the still-queued pause narration`(): Unit = runBlocking {
+        val facade = GatingRouterFacade()
+        NarrationSubscriber(fakeStore, facade, scope).start()
+
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        // Warmup narration the worker will block on, keeping later narrations queued.
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        // Wait until the worker is parked on the warmup before enqueuing the pause, so the pause
+        // is provably still queued (not yet delivered) when its approval resolves.
+        withTimeout(2_000L) { facade.blocked.await() }
+        // Pause is enqueued, then its approval resolves before the worker can drain it.
+        liveFlow.emit(storedEvent(approvalRequested("req-1"), seq = 3L))
+        liveFlow.emit(storedEvent(approvalResolved("req-1"), seq = 4L))
+        // A current, unrelated narration arrives after resolution and must still surface.
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("b"), TransitionId("t2")), seq = 5L))
+        // Let the (single, FIFO) collector drain its buffer so the resolve is handled pre-release.
+        delay(100L)
+
+        // Release the worker; it drains the warmup, skips the stale pause, narrates the second stage.
+        facade.release.complete(Unit)
+
+        withTimeout(2_000L) { while (facade.calls.size < 2) yield() }
+        delay(100L) // allow a stray (stale) pause narration to arrive if the drop failed
+
+        assertEquals(listOf("stage_completed", "stage_completed"), facade.calls.map { it.kind })
+        assertTrue(facade.calls.none { it.kind == "paused" }, "stale pause narration must be dropped")
+    }
+
+    @Test
+    fun `unrelated approval resolution leaves the pending pause intact`(): Unit = runBlocking {
+        val facade = GatingRouterFacade()
+        NarrationSubscriber(fakeStore, facade, scope).start()
+
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        withTimeout(2_000L) { facade.blocked.await() }
+        // Pause for req-1, but a *different* request (req-2) resolves: req-1's pause must survive.
+        liveFlow.emit(storedEvent(approvalRequested("req-1"), seq = 3L))
+        liveFlow.emit(storedEvent(approvalResolved("req-2"), seq = 4L))
+        delay(100L)
+
+        facade.release.complete(Unit)
+
+        withTimeout(2_000L) { while (facade.calls.size < 2) yield() }
+
+        assertEquals(listOf("stage_completed", "paused"), facade.calls.map { it.kind })
+    }
+
+    @Test
+    fun `resume drops every queued pause narration for the session`(): Unit = runBlocking {
+        val facade = GatingRouterFacade()
+        NarrationSubscriber(fakeStore, facade, scope).start()
+
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        withTimeout(2_000L) { facade.blocked.await() }
+        // A non-approval pause (session-scoped) plus an approval pause, both superseded by resume.
+        liveFlow.emit(
+            storedEvent(OrchestrationPausedEvent(sessionId, StageId("a"), "USER_REQUESTED"), seq = 3L),
+        )
+        liveFlow.emit(storedEvent(approvalRequested("req-1"), seq = 4L))
+        liveFlow.emit(storedEvent(OrchestrationResumedEvent(sessionId, StageId("a")), seq = 5L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("b"), TransitionId("t2")), seq = 6L))
+        delay(100L)
+
+        facade.release.complete(Unit)
+
+        withTimeout(2_000L) { while (facade.calls.size < 2) yield() }
+        delay(100L)
+
+        assertEquals(listOf("stage_completed", "stage_completed"), facade.calls.map { it.kind })
+        assertTrue(facade.calls.none { it.kind == "paused" }, "resume must drop all queued pauses")
+    }
 }