diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 7db734db..b6407959 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -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, + ) { + 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) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 999e8a51..cdfdd5f2 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -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> = 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>> = + 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 { + 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>() + 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). diff --git a/testing/integration/src/test/kotlin/ClarificationLoopTest.kt b/testing/integration/src/test/kotlin/ClarificationLoopTest.kt new file mode 100644 index 00000000..6bf3b0d6 --- /dev/null +++ b/testing/integration/src/test/kotlin/ClarificationLoopTest.kt @@ -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 { + 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): DefaultSessionOrchestrator { + var index = 0 + val router = object : InferenceRouter { + override suspend fun route( + stageId: StageId, + requiredCapabilities: Set, + ) = 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() + 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", + ) + } +}