feat(kernel): producer-exit clarification loop

When a stage produces an LLM artifact carrying a non-empty questions array, the orchestrator parks the run, records ClarificationRequested, awaits the operator's answers (submitClarification seam), records ClarificationAnswered, injects the answered Q&A as an L2 USER entry, and re-runs the same stage so the questions are resolved by the role that asked them. Runs on a dedicated pendingClarifications map and the enterStage success-branch loop, so a clarification round never consumes the failure-retry budget; capped at three rounds. Integration test covers the happy path and the cap.
2026-06-14 03:31:30 +04:00
parent df0144582a
commit 939331d895
3 changed files with 334 additions and 1 deletions
@@ -12,6 +12,8 @@ import com.correx.core.artifactstore.ArtifactStore
 import com.correx.core.events.events.ApprovalDecisionResolvedEvent
 import com.correx.core.events.events.ApprovalRequestedEvent
 import com.correx.core.events.events.ArtifactContentStoredEvent
+import com.correx.core.events.events.ClarificationAnswer
+import com.correx.core.events.events.ClarificationAnsweredEvent
 import com.correx.core.events.events.ArtifactValidatedEvent
 import com.correx.core.events.events.OrchestrationResumedEvent
 import com.correx.core.events.events.RefinementIterationEvent
@@ -21,6 +23,7 @@ import com.correx.core.events.orchestration.OrchestrationState
 import com.correx.core.events.types.ApprovalDecisionId
 import com.correx.core.events.types.ApprovalRequestId
 import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.ClarificationRequestId
 import com.correx.core.events.types.ArtifactLifecyclePhase
 import com.correx.core.events.types.SessionId
 import com.correx.core.events.types.StageId
@@ -270,6 +273,31 @@ class DefaultSessionOrchestrator(
        emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
    }

+    /**
+     * Delivers the operator's answers to a pending clarification raised by a stage. The waiting
+     * stage (parked in [requestClarificationIfNeeded]) completes and re-runs with the answers in
+     * context. If the server restarted while the clarification was pending there is no live
+     * coroutine to complete, so the answers are recorded directly and the session resumed.
+     */
+    suspend fun submitClarification(
+        sessionId: SessionId,
+        stageId: StageId,
+        requestId: ClarificationRequestId,
+        answers: List<ClarificationAnswer>,
+    ) {
+        val deferred = pendingClarifications[requestId]
+        if (deferred != null) {
+            deferred.complete(answers)
+            return
+        }
+        log.warn(
+            "submitClarification: no live deferred for requestId={} (server restart?), recording directly",
+            requestId.value,
+        )
+        emit(sessionId, ClarificationAnsweredEvent(requestId, sessionId, stageId, answers))
+        emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
+    }
+
    private suspend fun executeMove(
        ctx: EnrichedExecutionContext,
        decision: TransitionDecision.Move,
@@ -356,6 +384,11 @@ class DefaultSessionOrchestrator(
            ).outcome
            when (result) {
                is StageExecutionResult.Success -> {
+                    if (requestClarificationIfNeeded(ctx.sessionId, stageId, ctx.graph)) {
+                        // The stage raised open questions and the operator answered them; loop to
+                        // re-run the stage with the answers injected (no failure-retry budget spent).
+                        continue
+                    }
                    compactionService?.let { svc ->
                        val journalState = decisionJournalRepository.getJournal(ctx.sessionId)
                        val journalText = DecisionJournalRenderer().render(journalState)
@@ -30,6 +30,10 @@ import com.correx.core.events.events.ArtifactContentStoredEvent
 import com.correx.core.events.events.ArtifactCreatedEvent
 import com.correx.core.events.events.BriefGroundingCheckedEvent
 import com.correx.core.events.events.ContextTruncatedEvent
+import com.correx.core.events.events.ClarificationAnswer
+import com.correx.core.events.events.ClarificationAnsweredEvent
+import com.correx.core.events.events.ClarificationQuestions
+import com.correx.core.events.events.ClarificationRequestedEvent
 import com.correx.core.events.events.GroundingReference
 import com.correx.core.events.events.ToolCallAssessedEvent
 import com.correx.core.events.events.ArtifactValidatedEvent
@@ -71,6 +75,7 @@ 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.ArtifactId
+import com.correx.core.events.types.ClarificationRequestId
 import com.correx.core.events.types.ContextEntryId
 import com.correx.core.events.types.ContextPackId
 import com.correx.core.events.types.EventId
@@ -142,6 +147,9 @@ private const val REPO_MAP_INJECT_TOP_K = 30
 private const val HTTP_TIMEOUT = 408
 private const val HTTP_TOO_MANY_REQUESTS = 429

+// Cap on clarification rounds per stage, so a stage that keeps re-asking eventually proceeds.
+private const val MAX_CLARIFICATION_ROUNDS = 3
+
@SuppressWarnings(
    "ForbiddenComment",
    "UnusedParameter",
@@ -185,6 +193,13 @@ abstract class SessionOrchestrator(
    internal val pendingApprovals: ConcurrentHashMap<ApprovalRequestId, CompletableDeferred<ApprovalDecision>> =
        ConcurrentHashMap()

+    // A stage whose produced artifact carries open questions parks its run on one of these until the
+    // operator answers (submitClarification completes it). Separate from pendingApprovals so a
+    // clarification round never consumes the failure-retry budget.
+    internal val pendingClarifications:
+        ConcurrentHashMap<ClarificationRequestId, CompletableDeferred<List<ClarificationAnswer>>> =
+        ConcurrentHashMap()
+
    /** Cache mapping "sessionId:artifactId" to JSON-serialized artifact payload content.
     *  Populated when ProcessResult artifacts are stored on tool failure.
     *  Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */
@@ -328,6 +343,7 @@ abstract class SessionOrchestrator(

        val schemaEntries = buildSchemaEntries(responseFormat, stageId)
        val steeringEntries = buildSteeringNoteEntries(sessionId)
+        val clarificationEntries = buildClarificationAnswerEntries(sessionId)

        // Maintain the running transcript in true chronological order. Reading it back
        // from currentContext.layers (a Map grouped by layer) would scramble turn order
@@ -385,7 +401,8 @@ abstract class SessionOrchestrator(
            ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList()
        var accumulatedEntries =
            systemPrompt + profileEntries + projectProfileEntries + journalEntries + repoMapEntries +
-                needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + retryFeedbackEntries
+                needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries +
+                clarificationEntries + retryFeedbackEntries
        val contextPack = contextPackBuilder.build(
            id = ContextPackId(UUID.randomUUID().toString()),
            sessionId = sessionId,
@@ -1003,6 +1020,73 @@ abstract class SessionOrchestrator(
        }
    }

+    /**
+     * Injects the operator's answers to a stage's open questions as an L2 USER entry, so the stage
+     * sees its own questions resolved on the clarification re-run. Correlates each answer's
+     * questionId back to the prompt recorded on the [ClarificationRequestedEvent].
+     */
+    private suspend fun buildClarificationAnswerEntries(sessionId: SessionId): List<ContextEntry> {
+        val events = eventStore.read(sessionId)
+        val answered = events.mapNotNull { it.payload as? ClarificationAnsweredEvent }
+        if (answered.isEmpty()) return emptyList()
+        val prompts = events
+            .mapNotNull { it.payload as? ClarificationRequestedEvent }
+            .flatMap { it.questions }
+            .associate { it.id to it.prompt }
+        val rendered = answered.flatMap { it.answers }.joinToString("\n") { a ->
+            "Q: ${prompts[a.questionId] ?: a.questionId}\nA: ${a.value}"
+        }
+        val content = "The operator answered your open questions. Treat these answers as authoritative " +
+            "and do not ask them again:\n$rendered"
+        return listOf(
+            ContextEntry(
+                id = ContextEntryId(UUID.randomUUID().toString()),
+                layer = ContextLayer.L2,
+                content = content,
+                sourceType = "clarificationAnswer",
+                sourceId = sessionId.value,
+                tokenEstimate = estimateTokens(content),
+                role = EntryRole.USER,
+            ),
+        )
+    }
+
+    /**
+     * If [stageId] just produced an LLM artifact carrying a non-empty `questions` array, park the
+     * run: record the questions (invariant #9 — observed at parse time), await the operator's
+     * answers, record them, and return true so the caller re-runs the stage with the answers in
+     * context. Bounded by [MAX_CLARIFICATION_ROUNDS] so a stage that keeps re-asking eventually
+     * proceeds. Returns false when there are no questions or the round budget is spent.
+     */
+    internal suspend fun requestClarificationIfNeeded(
+        sessionId: SessionId,
+        stageId: StageId,
+        graph: WorkflowGraph,
+    ): Boolean {
+        val slot = graph.stages[stageId]?.produces?.firstOrNull { it.kind.llmEmitted } ?: return false
+        val content = artifactContentCache["${sessionId.value}:${slot.name.value}"] ?: return false
+        val questions = ClarificationQuestions.parse(content)
+        if (questions.isEmpty()) return false
+
+        val priorRounds = eventStore.read(sessionId)
+            .count { (it.payload as? ClarificationRequestedEvent)?.stageId == stageId }
+        if (priorRounds >= MAX_CLARIFICATION_ROUNDS) return false
+
+        val requestId = ClarificationRequestId(UUID.randomUUID().toString())
+        val deferred = CompletableDeferred<List<ClarificationAnswer>>()
+        pendingClarifications[requestId] = deferred
+        emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "CLARIFICATION_PENDING"))
+        emit(sessionId, ClarificationRequestedEvent(requestId, sessionId, stageId, questions))
+        return try {
+            val answers = deferred.await()
+            emit(sessionId, ClarificationAnsweredEvent(requestId, sessionId, stageId, answers))
+            emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
+            true
+        } finally {
+            pendingClarifications.remove(requestId)
+        }
+    }
+
    /**
     * Renders a top-K slice of the recorded repo map as a droppable L3 context entry. Reads
     * the latest [RepoMapComputedEvent] from the log (replay-safe — never re-scans the FS).
@@ -0,0 +1,216 @@
+import com.correx.core.approvals.ApprovalProjector
+import com.correx.core.approvals.DefaultApprovalReducer
+import com.correx.core.approvals.DefaultApprovalRepository
+import com.correx.core.approvals.domain.DefaultApprovalEngine
+import com.correx.core.artifacts.DefaultArtifactReducer
+import com.correx.core.artifacts.kind.ConfigArtifactKind
+import com.correx.core.artifacts.kind.JsonSchema
+import com.correx.core.artifacts.kind.TypedArtifactSlot
+import com.correx.core.events.events.ClarificationAnswer
+import com.correx.core.events.events.ClarificationAnsweredEvent
+import com.correx.core.events.events.ClarificationRequestedEvent
+import com.correx.core.events.events.WorkflowCompletedEvent
+import com.correx.core.events.execution.RetryPolicy
+import com.correx.core.events.types.ArtifactId
+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.inference.InferenceRepository
+import com.correx.core.inference.InferenceRouter
+import com.correx.core.inference.InferenceState
+import com.correx.core.inference.ModelCapability
+import com.correx.core.journal.DecisionJournalProjector
+import com.correx.core.journal.DefaultDecisionJournalReducer
+import com.correx.core.journal.DefaultDecisionJournalRepository
+import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
+import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
+import com.correx.core.kernel.orchestration.OrchestrationConfig
+import com.correx.core.kernel.orchestration.OrchestrationProjector
+import com.correx.core.kernel.orchestration.OrchestrationRepository
+import com.correx.core.kernel.orchestration.OrchestratorEngines
+import com.correx.core.kernel.orchestration.OrchestratorRepositories
+import com.correx.core.kernel.retry.DefaultRetryCoordinator
+import com.correx.core.risk.DefaultRiskAssessor
+import com.correx.core.sessions.DefaultSessionRepository
+import com.correx.core.sessions.projections.replay.DefaultEventReplayer
+import com.correx.core.sessions.projections.replay.EventReplayer
+import com.correx.core.transitions.graph.StageConfig
+import com.correx.core.transitions.graph.TransitionEdge
+import com.correx.core.transitions.graph.WorkflowGraph
+import com.correx.core.validation.pipeline.ValidationPipeline
+import com.correx.infrastructure.persistence.InMemoryEventStore
+import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository
+import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore
+import com.correx.testing.fixtures.cyclePolicyMissingValidator
+import com.correx.testing.fixtures.context.ContextFixtures
+import com.correx.testing.fixtures.inference.MockInferenceProvider
+import com.correx.testing.fixtures.transitions.TransitionFixtures
+import com.correx.testing.kernel.MockSessionEventReplayer
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.yield
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+/**
+ * Integration test for the producer-exit clarification loop: a stage whose produced artifact
+ * carries an unanswered `questions` array parks its run, surfaces a [ClarificationRequestedEvent],
+ * and on operator answer re-runs the same stage with the answers in context — without consuming the
+ * failure-retry budget (retry maxAttempts is 1 here). Bounded at three rounds.
+ */
+class ClarificationLoopTest {
+
+    private val eventStore = InMemoryEventStore()
+    private val sessionRepository = DefaultSessionRepository(MockSessionEventReplayer())
+    private val orchestrationRepository = OrchestrationRepository(
+        DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())),
+    )
+    private val inferenceRepository = InferenceRepository(
+        object : EventReplayer<InferenceState> {
+            override fun rebuild(sessionId: SessionId) = InferenceState()
+        },
+    )
+    private val approvalRepository = DefaultApprovalRepository(
+        DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())),
+    )
+    private val decisionJournalRepository = DefaultDecisionJournalRepository(
+        DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())),
+    )
+    private val repositories = OrchestratorRepositories(
+        eventStore = eventStore,
+        inferenceRepository = inferenceRepository,
+        orchestrationRepository = orchestrationRepository,
+        sessionRepository = sessionRepository,
+        artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()),
+        approvalRepository = approvalRepository,
+    )
+
+    private val analysisKind = ConfigArtifactKind(
+        id = "analysis",
+        schema = JsonSchema(type = "object", properties = emptyMap()),
+        llmEmitted = true,
+    )
+    private val analysisId = ArtifactId("analysis")
+    private val analystStage = StageId("analyst")
+
+    /** Single stage that produces `analysis`, then transitions to the terminal `done`. */
+    private fun graph() = WorkflowGraph(
+        id = "clarify",
+        stages = mapOf(
+            analystStage to StageConfig(
+                produces = listOf(TypedArtifactSlot(name = analysisId, kind = analysisKind)),
+            ),
+        ),
+        transitions = setOf(
+            TransitionEdge(
+                id = TransitionId("analyst-to-done"),
+                from = analystStage,
+                to = StageId("done"),
+                condition = { true },
+            ),
+        ),
+        start = analystStage,
+    )
+
+    private val withQuestions =
+        """{"summary":"build a UI","questions":[{"id":"stack","prompt":"Which stack?","options":["React","Vue"]}]}"""
+    private val clean = """{"summary":"React chosen — no open questions"}"""
+
+    /** Provider returns [responses] in order, clamping to the last once exhausted. */
+    private fun orchestrator(responses: List<String>): DefaultSessionOrchestrator {
+        var index = 0
+        val router = object : InferenceRouter {
+            override suspend fun route(
+                stageId: StageId,
+                requiredCapabilities: Set<ModelCapability>,
+            ) = MockInferenceProvider(fixedResponse = responses[minOf(index++, responses.lastIndex)])
+        }
+        return DefaultSessionOrchestrator(
+            repositories = repositories,
+            engines = OrchestratorEngines(
+                transitionResolver = TransitionFixtures.simpleResolver(),
+                contextPackBuilder = ContextFixtures.simpleBuilder(),
+                inferenceRouter = router,
+                validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())),
+                approvalEngine = DefaultApprovalEngine(),
+                riskAssessor = DefaultRiskAssessor(),
+            ),
+            retryCoordinator = DefaultRetryCoordinator(eventStore),
+            artifactStore = NoopArtifactStore(),
+            decisionJournalRepository = decisionJournalRepository,
+        )
+    }
+
+    private val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
+
+    @Test
+    fun `open questions park the stage, the answer re-runs it, then the workflow completes`():
+        Unit = runBlocking {
+        val sessionId = SessionId("clarify-happy")
+        val orchestrator = orchestrator(listOf(withQuestions, clean))
+        val runJob = launch { orchestrator.run(sessionId, graph(), config) }
+
+        val request = withTimeout(5_000) {
+            var r: ClarificationRequestedEvent? = null
+            while (r == null) {
+                r = eventStore.read(sessionId).firstNotNullOfOrNull { it.payload as? ClarificationRequestedEvent }
+                if (r == null) yield()
+            }
+            r
+        }
+        assertEquals(analystStage, request.stageId)
+        assertEquals("Which stack?", request.questions.single().prompt)
+
+        orchestrator.submitClarification(
+            sessionId, request.stageId, request.requestId,
+            listOf(ClarificationAnswer(questionId = "stack", value = "React")),
+        )
+        runJob.join()
+
+        val events = eventStore.read(sessionId)
+        assertNotNull(
+            events.firstNotNullOfOrNull { it.payload as? ClarificationAnsweredEvent },
+            "Expected a ClarificationAnsweredEvent",
+        )
+        assertNotNull(
+            events.find { it.payload is WorkflowCompletedEvent },
+            "Workflow should complete after the clarification re-run produced a clean analysis",
+        )
+    }
+
+    @Test
+    fun `clarification rounds are capped so a stuck stage still proceeds`(): Unit = runBlocking {
+        val sessionId = SessionId("clarify-cap")
+        // Always returns questions: without a cap this would loop forever.
+        val orchestrator = orchestrator(listOf(withQuestions))
+        val runJob = launch { orchestrator.run(sessionId, graph(), config) }
+
+        val handled = mutableSetOf<String>()
+        withTimeout(10_000) {
+            while (eventStore.read(sessionId).none { it.payload is WorkflowCompletedEvent }) {
+                val pending = eventStore.read(sessionId)
+                    .mapNotNull { it.payload as? ClarificationRequestedEvent }
+                    .firstOrNull { it.requestId.value !in handled }
+                if (pending != null) {
+                    handled += pending.requestId.value
+                    orchestrator.submitClarification(
+                        sessionId, pending.stageId, pending.requestId,
+                        listOf(ClarificationAnswer(questionId = "stack", value = "React")),
+                    )
+                }
+                yield()
+            }
+        }
+        runJob.join()
+
+        val rounds = eventStore.read(sessionId).count { it.payload is ClarificationRequestedEvent }
+        assertEquals(3, rounds, "Clarification must be capped at three rounds")
+        assertNotNull(
+            eventStore.read(sessionId).find { it.payload is WorkflowCompletedEvent },
+            "Workflow should proceed once the clarification cap is reached",
+        )
+    }
+}