From e95d2633f80a9bdd78733bcaead158ca9154f66c Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 3 Jun 2026 22:55:13 +0400 Subject: [PATCH] feat(narration): ground pause narration in the pending approval + roomier budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live QA showed two problems with paused-approval narration: - The narrator was told to "name the stage, outcome, and reason" but the pause trigger only carried "APPROVAL_PENDING" with no detail — on a first-stage pause there's no L2 history either, so it had nothing to ground on. NarrationSubscriber now captures the latest ApprovalRequestedEvent per session (tool name, tier, preview) and folds it into the pause instruction (preview truncated to 800 chars). - Narration reused the 512-token chat generation config; a reasoning model spent the whole budget "thinking" and returned empty content (finishReason=length). Add a separate narrationGenerationConfig (maxTokens=1024) so thinking can complete and still leave room for the one or two sentences. Adds a subscriber test asserting the pause narration names the tool and includes the preview. --- .../server/narration/NarrationSubscriber.kt | 30 ++++++++++++- .../narration/NarrationSubscriberTest.kt | 42 +++++++++++++++++++ .../com/correx/core/router/RouterFacade.kt | 2 +- .../correx/core/router/model/RouterConfig.kt | 9 ++++ 4 files changed, 81 insertions(+), 2 deletions(-) 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 42b64435..36e5b3e1 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,5 +1,6 @@ package com.correx.apps.server.narration +import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageFailedEvent @@ -28,7 +29,12 @@ class NarrationSubscriber( ) { private data class SessionLane(val channel: Channel, var used: Int) + // Latest approval awaiting a decision per session, so a pause narration can name the + // tool and show the proposed change instead of the bare "APPROVAL_PENDING" reason. + private data class PendingApproval(val toolName: String, val tier: String, val preview: String) + private val lanes = ConcurrentHashMap() + private val pendingApprovals = ConcurrentHashMap() fun start() { eventStore.subscribeAll().onEach { handle(it) }.launchIn(scope) @@ -38,6 +44,11 @@ class NarrationSubscriber( val sid = event.metadata.sessionId when (val p = event.payload) { is WorkflowStartedEvent -> lanes[sid.value]?.let { it.used = 0 } + is ApprovalRequestedEvent -> pendingApprovals[sid.value] = PendingApproval( + toolName = p.toolName ?: "a tool", + tier = p.tier.toString(), + preview = p.preview.orEmpty(), + ) is StageCompletedEvent -> enqueue( sid, NarrationTrigger(kind = "stage_completed", instruction = "Stage ${p.stageId.value} completed. Summarise what was accomplished."), @@ -56,12 +67,28 @@ class NarrationSubscriber( ) is OrchestrationPausedEvent -> enqueue( sid, - NarrationTrigger(kind = "paused", instruction = "Orchestration paused: ${p.reason}. Inform the user."), + NarrationTrigger(kind = "paused", instruction = pauseInstruction(sid, p)), ) else -> Unit } } + private fun pauseInstruction(sessionId: SessionId, event: OrchestrationPausedEvent): String { + val pending = pendingApprovals[sessionId.value] + if (!event.reason.contains("APPROVAL", ignoreCase = true) || pending == null) { + return "Orchestration paused in stage ${event.stageId.value}: ${event.reason}. Inform the user." + } + return buildString { + append("The workflow paused in stage ${event.stageId.value} awaiting your approval of the ") + append("'${pending.toolName}' tool (tier ${pending.tier}). ") + append("Tell the operator what it is about to do and why it needs sign-off.") + if (pending.preview.isNotBlank()) { + append("\n\nProposed change:\n") + append(pending.preview.take(MAX_PREVIEW_CHARS)) + } + } + } + private fun enqueue(sessionId: SessionId, trigger: NarrationTrigger) { val lane = lanes.computeIfAbsent(sessionId.value) { startLane(sessionId) } if (lane.used >= maxPerRun) { @@ -93,6 +120,7 @@ class NarrationSubscriber( companion object { private const val DEFAULT_MAX_PER_RUN = 100 + private const val MAX_PREVIEW_CHARS = 800 private val log = LoggerFactory.getLogger(NarrationSubscriber::class.java) } } 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 db607dab..aab3b8f7 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,24 @@ package com.correx.apps.server.narration +import com.correx.core.approvals.Tier +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.NewEvent +import com.correx.core.events.events.OrchestrationPausedEvent 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.ApprovalRequestId import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.events.types.TransitionId +import com.correx.core.events.types.ValidationReportId +import org.junit.jupiter.api.Assertions.assertTrue import com.correx.core.router.ChatMode import com.correx.core.router.RouterFacade import com.correx.core.router.model.NarrationTrigger @@ -106,6 +112,42 @@ class NarrationSubscriberTest { assertEquals("workflow_completed", facade.calls[2].second.kind) } + @Test + fun `pause narration is grounded with the pending approval tool and preview`(): Unit = runBlocking { + val facade = RecordingRouterFacade() + NarrationSubscriber(fakeStore, facade, scope).start() + + while (liveFlow.subscriptionCount.value == 0) yield() + + liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L)) + liveFlow.emit( + storedEvent( + ApprovalRequestedEvent( + requestId = ApprovalRequestId("req-1"), + tier = Tier.T2, + validationReportId = ValidationReportId("vr-1"), + riskSummaryId = null, + sessionId = sessionId, + stageId = StageId("write_script"), + projectId = null, + toolName = "file_write", + preview = "--- a/healthcheck.sh\n+++ b/healthcheck.sh", + ), + seq = 2L, + ), + ) + liveFlow.emit( + storedEvent(OrchestrationPausedEvent(sessionId, StageId("write_script"), "APPROVAL_PENDING"), seq = 3L), + ) + + withTimeout(2_000L) { while (facade.calls.isEmpty()) yield() } + + val instruction = facade.calls.first().second.instruction + assertEquals("paused", facade.calls.first().second.kind) + assertTrue(instruction.contains("file_write"), "pause narration must name the tool: $instruction") + assertTrue(instruction.contains("healthcheck.sh"), "pause narration must include the preview: $instruction") + } + @Test fun `maxPerRun=1 skips second trigger after first narration`(): Unit = runBlocking { val facade = RecordingRouterFacade() diff --git a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt index 480a6efe..b76f9ab7 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt @@ -181,7 +181,7 @@ class DefaultRouterFacade( sessionId = sessionId, stageId = effectiveStageId, contextPack = contextPack, - generationConfig = config.generationConfig, + generationConfig = config.narrationGenerationConfig, responseFormat = ResponseFormat.Text, ) val inferenceResponse = provider.infer(inferenceRequest) diff --git a/core/router/src/main/kotlin/com/correx/core/router/model/RouterConfig.kt b/core/router/src/main/kotlin/com/correx/core/router/model/RouterConfig.kt index 6a802758..817f49b2 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/model/RouterConfig.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/model/RouterConfig.kt @@ -14,4 +14,13 @@ data class RouterConfig( topP = 0.9, maxTokens = 512, ), + // Narration uses a larger token ceiling: reasoning models spend tokens "thinking" + // before emitting content, and a 512 cap left no room for the actual line (the + // response came back empty with finishReason=length). 1024 lets thinking complete + // and still produce the one or two sentences. + val narrationGenerationConfig: GenerationConfig = GenerationConfig( + temperature = 0.7, + topP = 0.9, + maxTokens = 1024, + ), )