diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt index e762c82e..09658eee 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt +++ b/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, 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, + var used: Int, + val pendingPauses: MutableSet = Collections.newSetFromMap(ConcurrentHashMap()), + ) private val lanes = ConcurrentHashMap() @@ -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(capacity = Channel.UNLIMITED) + val channel = Channel(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" } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt index 3bdf5515..c5f198ab 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt +++ b/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() + val release = CompletableDeferred() + // 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() + 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") + } }