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, + ), )