feat(narration): ground pause narration in the pending approval + roomier budget

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.
This commit is contained in:
2026-06-03 22:55:13 +04:00
parent e6084dcbda
commit e95d2633f8
4 changed files with 81 additions and 2 deletions
@@ -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<NarrationTrigger>, 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<String, SessionLane>()
private val pendingApprovals = ConcurrentHashMap<String, PendingApproval>()
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)
}
}
@@ -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()