From 7f820bafe60c91f6fd83d3c779e0ecc077fd8713 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 13 Jul 2026 11:44:04 +0400 Subject: [PATCH] refactor(kernel): extract executeStage to extension, promote members internal Pilot for the SessionOrchestrator god-class decomposition. Moves executeStage (426-line stage-execution driver) out to SessionOrchestratorExecution.kt as an internal extension fun, hoists three nested holders (RunEffectives, ArtifactLadderOutcome, RenderedToolResult) to top-level in SessionOrchestratorTypes.kt, and promotes class/protected members to internal so extensions can reach them. Behavior-preserving relocation; kernel+test+server compile green. --- .../DefaultSessionOrchestrator.kt | 2 +- .../orchestration/SessionOrchestrator.kt | 831 +++--------------- .../SessionOrchestratorExecution.kt | 741 ++++++++++++++++ .../orchestration/SessionOrchestratorTypes.kt | 22 + 4 files changed, 891 insertions(+), 705 deletions(-) create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorTypes.kt 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 fb80df81..8224b548 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 @@ -99,7 +99,7 @@ class DefaultSessionOrchestrator( private val retryCoordinator: RetryCoordinator, artifactStore: ArtifactStore, tokenizer: Tokenizer? = null, - private val decisionJournalRepository: DefaultDecisionJournalRepository, + decisionJournalRepository: DefaultDecisionJournalRepository, private val compactionService: JournalCompactionService? = null, artifactKindRegistry: ArtifactKindRegistry? = null, repoKnowledgeRetriever: RepoKnowledgeRetriever? = null, 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 0f881618..df275ac0 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 @@ -196,23 +196,23 @@ import kotlin.coroutines.cancellation.CancellationException // with no progress. After this many all-rejected rounds we force the model to stop retrying and // produce its output; a single successful tool call resets the counter. // Value now lives in OrchestrationTuning.rejectionLoopNudgeThreshold. -private val WRITE_TOOL_NAMES = setOf("file_write", "file_edit") -private const val STAGE_COMPLETE_TOOL = "stage_complete" -private const val EMIT_ARTIFACT_TOOL = "emit_artifact" -private const val SCOPE_PROPOSAL_TOOL = "propose_scope" -private const val READ_BEFORE_WRITE_CODE = "READ_BEFORE_WRITE" -private const val REFERENCE_EXISTS_CODE = "REFERENCE_EXISTS" -private const val OUTPUT_SUMMARY_LIMIT = 500 +internal val WRITE_TOOL_NAMES = setOf("file_write", "file_edit") +internal const val STAGE_COMPLETE_TOOL = "stage_complete" +internal const val EMIT_ARTIFACT_TOOL = "emit_artifact" +internal const val SCOPE_PROPOSAL_TOOL = "propose_scope" +internal const val READ_BEFORE_WRITE_CODE = "READ_BEFORE_WRITE" +internal const val REFERENCE_EXISTS_CODE = "REFERENCE_EXISTS" +internal const val OUTPUT_SUMMARY_LIMIT = 500 // Global floor on tool-result text entering model context, applied beneath each tool's own // outputCompressor. A tool without a compressor (or one whose output survives compression) can still // flood a small model's window, so any Success body over this many chars is shown head+tail with a // marker and the FULL raw output spilled to the artifact store for retrieval via tool_output. // ponytail: fixed constants; lift to OrchestrationTuning if operators need to tune per-deployment. -private const val TOOL_RESULT_MAX_CHARS = 8_000 -private const val TOOL_RESULT_HEAD_LINES = 60 -private const val TOOL_RESULT_TAIL_LINES = 60 -private const val TOOL_OUTPUT_TOOL = "tool_output" +internal const val TOOL_RESULT_MAX_CHARS = 8_000 +internal const val TOOL_RESULT_HEAD_LINES = 60 +internal const val TOOL_RESULT_TAIL_LINES = 60 +internal const val TOOL_OUTPUT_TOOL = "tool_output" /** * Frame an over-cap tool output as `header` + head lines + a truncation marker (naming the @@ -236,15 +236,15 @@ internal fun frameTruncatedToolResult(header: String, compressed: String, ref: S // A stage prompt must explicitly ask about documentation for .md/docs paths to enter the repo // layout listing; otherwise docs only reach context through a real retrieval hit. -private val DOCS_REQUEST_REGEX = +internal val DOCS_REQUEST_REGEX = Regex("""\b(docs?|readme|adr|changelog|documentation|markdown)\b|\.md\b""", RegexOption.IGNORE_CASE) -private fun isDocPath(path: String): Boolean = +internal fun isDocPath(path: String): Boolean = path.endsWith(".md", ignoreCase = true) || path.split('/').any { it.equals("docs", ignoreCase = true) || it.equals("doc", ignoreCase = true) } // Source types whose entries land in the REQUIRED context bucket (see stampBuckets). -private val REQUIRED_SOURCE_TYPES = setOf( +internal val REQUIRED_SOURCE_TYPES = setOf( "systemPrompt", "agentPrompt", "agentInstructions", @@ -259,14 +259,14 @@ private val REQUIRED_SOURCE_TYPES = setOf( ) // HTTP statuses that are transient despite being 4xx (F-002 retry classification). -private const val HTTP_TIMEOUT = 408 -private const val HTTP_TOO_MANY_REQUESTS = 429 +internal const val HTTP_TIMEOUT = 408 +internal const val HTTP_TOO_MANY_REQUESTS = 429 // Cap on clarification rounds per stage — OrchestrationTuning.maxClarificationRounds. // Static-analysis output caps: the tail retained in the recorded event vs. the (larger) slice fed // back to the model so it can fix the failure. Tails, because the error summary sits at the end. -private const val CONTRACT_EVIDENCE_CAP = 400 +internal const val CONTRACT_EVIDENCE_CAP = 400 // Gate 3 semantic review: a FAIL blocks only on a correctness finding at/above this confidence. // Per-gate retry budgeting + hybrid salvage (design 2026-07-06-per-gate-retry-budgets.md) is now the // authority on when review retries stop — the gate simply reports the block and the RetryCoordinator @@ -276,16 +276,16 @@ private const val CONTRACT_EVIDENCE_CAP = 400 // this many blocks, findings surface without blocking so a stuck reviewer can never trap a stage // forever. Set well above any legitimate budget+salvage sequence. Objective text is capped too. // REVIEW_BLOCK_MIN_CONFIDENCE / REVIEW_BLOCK_RETRY_CAP now in OrchestrationTuning. -private const val REVIEW_OBJECTIVE_CAP = 4_000 -private const val STATIC_ANALYSIS_SUMMARY_CAP = 2_000 -private const val STATIC_ANALYSIS_FEEDBACK_CAP = 6_000 +internal const val REVIEW_OBJECTIVE_CAP = 4_000 +internal const val STATIC_ANALYSIS_SUMMARY_CAP = 2_000 +internal const val STATIC_ANALYSIS_FEEDBACK_CAP = 6_000 /** Profile command alias run before a build gate to make it runnable (e.g. `npm ci`). */ -private const val SETUP_COMMAND_ALIAS = "setup" +internal const val SETUP_COMMAND_ALIAS = "setup" // KindInference kinds for build manifests — a written file of one of these declares a buildable // project toolchain even before any source lands, so the auto build gate fires (see // SessionOrchestrator.sessionProducedBuildTarget). Kept in sync with KindInference's manifest kinds. -private val BUILD_MANIFEST_KINDS = setOf("package_json", "gradle_module") +internal val BUILD_MANIFEST_KINDS = setOf("package_json", "gradle_module") @SuppressWarnings( "ForbiddenComment", @@ -299,41 +299,41 @@ private val BUILD_MANIFEST_KINDS = setOf("package_json", "gradle_module") abstract class SessionOrchestrator( repositories: OrchestratorRepositories, engines: OrchestratorEngines, - protected val artifactStore: ArtifactStore, - private val decisionJournalRepository: DefaultDecisionJournalRepository, - private val decisionJournalRenderer: DecisionJournalRenderer = DecisionJournalRenderer(), - private val artifactKindRegistry: ArtifactKindRegistry? = null, - private val repoKnowledgeRetriever: RepoKnowledgeRetriever? = null, - private val readyTaskCounter: ReadyTaskCounter? = null, - private val taskClaimCoordinator: TaskClaimCoordinator? = null, - protected val tuning: OrchestrationTuning = OrchestrationTuning(), + internal val artifactStore: ArtifactStore, + internal val decisionJournalRepository: DefaultDecisionJournalRepository, + internal val decisionJournalRenderer: DecisionJournalRenderer = DecisionJournalRenderer(), + internal val artifactKindRegistry: ArtifactKindRegistry? = null, + internal val repoKnowledgeRetriever: RepoKnowledgeRetriever? = null, + internal val readyTaskCounter: ReadyTaskCounter? = null, + internal val taskClaimCoordinator: TaskClaimCoordinator? = null, + internal val tuning: OrchestrationTuning = OrchestrationTuning(), ) { - private val log = LoggerFactory.getLogger(this::class.java) - private val eventStore: EventStore = repositories.eventStore - private val transitionResolver: TransitionResolver = engines.transitionResolver - private val contextPackBuilder: ContextPackBuilder = engines.contextPackBuilder - private val inferenceRouter: InferenceRouter = engines.inferenceRouter - private val validationPipeline: ValidationPipeline = engines.validationPipeline - private val riskAssessor: RiskAssessor = engines.riskAssessor - private val promptResolver: PromptResolver = engines.promptResolver - private val toolExecutor: ToolExecutor? = engines.toolExecutor - private val toolRegistry: ToolRegistry? = engines.toolRegistry - private val toolCallAssessor: ToolCallAssessor? = engines.toolCallAssessor - private val workspacePolicy: WorkspacePolicy? = engines.workspacePolicy - private val worldProbe: WorldProbe = engines.worldProbe - private val staticAnalysisRunner: StaticAnalysisRunner? = engines.staticAnalysisRunner - private val contractAssertionEvaluator: ContractAssertionEvaluator? = engines.contractAssertionEvaluator - private val planCompilationCheck: PlanCompilationCheck? = engines.planCompilationCheck - private val semanticReviewer: SemanticReviewer? = engines.semanticReviewer - private val workspaceToolRegistryProvider: WorkspaceToolRegistryProvider? = engines.workspaceToolRegistryProvider - private val inferenceRepository: InferenceRepository = repositories.inferenceRepository + internal val log = LoggerFactory.getLogger(this::class.java) + internal val eventStore: EventStore = repositories.eventStore + internal val transitionResolver: TransitionResolver = engines.transitionResolver + internal val contextPackBuilder: ContextPackBuilder = engines.contextPackBuilder + internal val inferenceRouter: InferenceRouter = engines.inferenceRouter + internal val validationPipeline: ValidationPipeline = engines.validationPipeline + internal val riskAssessor: RiskAssessor = engines.riskAssessor + internal val promptResolver: PromptResolver = engines.promptResolver + internal val toolExecutor: ToolExecutor? = engines.toolExecutor + internal val toolRegistry: ToolRegistry? = engines.toolRegistry + internal val toolCallAssessor: ToolCallAssessor? = engines.toolCallAssessor + internal val workspacePolicy: WorkspacePolicy? = engines.workspacePolicy + internal val worldProbe: WorldProbe = engines.worldProbe + internal val staticAnalysisRunner: StaticAnalysisRunner? = engines.staticAnalysisRunner + internal val contractAssertionEvaluator: ContractAssertionEvaluator? = engines.contractAssertionEvaluator + internal val planCompilationCheck: PlanCompilationCheck? = engines.planCompilationCheck + internal val semanticReviewer: SemanticReviewer? = engines.semanticReviewer + internal val workspaceToolRegistryProvider: WorkspaceToolRegistryProvider? = engines.workspaceToolRegistryProvider + internal val inferenceRepository: InferenceRepository = repositories.inferenceRepository internal val orchestrationRepository: OrchestrationRepository = repositories.orchestrationRepository - protected open val tokenizer: Tokenizer? = null - private val approvalEngine: ApprovalEngine = engines.approvalEngine - private val approvalRepository: DefaultApprovalRepository = repositories.approvalRepository + internal open val tokenizer: Tokenizer? = null + internal val approvalEngine: ApprovalEngine = engines.approvalEngine + internal val approvalRepository: DefaultApprovalRepository = repositories.approvalRepository internal abstract val cancellations: ConcurrentHashMap - protected abstract val subagentRunner: SubagentRunner + internal abstract val subagentRunner: SubagentRunner internal val pendingApprovals: ConcurrentHashMap> = ConcurrentHashMap() @@ -356,19 +356,19 @@ abstract class SessionOrchestrator( /** 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. */ - protected val artifactContentCache: ConcurrentHashMap = ConcurrentHashMap() + internal val artifactContentCache: ConcurrentHashMap = ConcurrentHashMap() /** Drops a terminated session's cached artifact contents (the heaviest per-session state — full * file/JSON payloads). Safe: rehydrateArtifactContentCache rebuilds it from durable events if the * session is ever resumed. Called on WorkflowCompleted/WorkflowFailed. */ - private fun evictArtifactContentCache(sessionId: SessionId) { + internal fun evictArtifactContentCache(sessionId: SessionId) { val prefix = "${sessionId.value}:" artifactContentCache.keys.removeAll { it.startsWith(prefix) } } /** Deterministic extraction/repair ladder for near-miss LLM artifact text (prose-wrapped JSON, * code fences, trailing commas). Pure — recomputes on replay, records no events. */ - private val artifactExtractionPipeline = ArtifactExtractionPipeline() + internal val artifactExtractionPipeline = ArtifactExtractionPipeline() abstract suspend fun run( sessionId: SessionId, @@ -380,12 +380,6 @@ abstract class SessionOrchestrator( // --- per-run effective registry + policy --- - internal data class RunEffectives( - val registry: ToolRegistry?, - val executor: ToolExecutor?, - val policy: WorkspacePolicy?, - ) - internal fun effectivesFor(config: OrchestrationConfig): RunEffectives { val wsTools = config.workspace?.let { workspaceToolRegistryProvider?.forWorkspace(it) } val registry = wsTools?.registry ?: toolRegistry @@ -396,584 +390,14 @@ abstract class SessionOrchestrator( return RunEffectives(registry, executor, policy) } - // --- stage execution --- - @Suppress("CyclomaticComplexMethod") - internal suspend fun executeStage( - sessionId: SessionId, - stageId: StageId, - graph: WorkflowGraph, - session: Session, - config: OrchestrationConfig, - effectives: RunEffectives, - ): StageExecutionResult { - val stageConfig = requireNotNull(graph.stages[stageId]) { - "Stage '${stageId.value}' not declared in workflow graph" - } - val systemPrompt = stageConfig.metadata["systemPrompt"] - ?.let { path -> - runCatching { promptResolver.resolve(path) } - .onFailure { - log.error(error(stageId, path, it)) - } - .getOrNull() - } - ?.takeIf { it.isNotBlank() } - ?.let { text -> - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L0, - content = text, - sourceType = "systemPrompt", - sourceId = stageId.value, - tokenEstimate = estimateTokens(text), - role = EntryRole.SYSTEM, - ), - ) - } ?: config.defaultSystemPromptPath - ?.let { path -> - runCatching { promptResolver.resolve(path) } - .onFailure { - log.error(error(stageId, path, it)) - } - .getOrNull() - } - ?.takeIf { it.isNotBlank() } - ?.let { text -> - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L0, - content = text, - sourceType = "systemPrompt", - sourceId = stageId.value, - tokenEstimate = estimateTokens(text), - role = EntryRole.SYSTEM, - ), - ) - } ?: emptyList() - // A stage re-entered to repair a gate failure has its own (often generative/"scaffold") prompt - // SUPPRESSED: that mandate is what drove it to overwrite real files with stubs on re-entry. The - // recovery-ticket entry (buildRecoveryTicketEntry) is the sole mandate here — it already carries - // "make the smallest change, don't rewrite from scratch" plus the gate evidence. See - // mandateSuppressedByTicket. - val promptEntries = if (mandateSuppressedByTicket(sessionId, stageId, stageConfig)) { - emptyList() - } else { - stageConfig.metadata["promptInline"] - ?.takeIf { it.isNotBlank() } - ?.let { text -> - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L1, - content = text, - sourceType = "agentPrompt", - sourceId = stageId.value, - tokenEstimate = estimateTokens(text), - role = EntryRole.USER, - ), - ) - } - ?: stageConfig.metadata["prompt"] - ?.let { path -> - val resolvedText = runCatching { promptResolver.resolve(path) } - .onFailure { - log.error( - "[SessionOrchestrator] stage=${stageId.value}: " + - "failed to load prompt '$path': ${it.message}", - ) - } - .getOrNull() - val text = resolvedText?.takeIf { it.isNotBlank() } - ?: error( - "[SessionOrchestrator] stage=${stageId.value}: " + - "declared prompt '$path' could not be resolved", - ) - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L1, - content = text, - sourceType = "agentPrompt", - sourceId = stageId.value, - tokenEstimate = estimateTokens(text), - role = EntryRole.USER, - ), - ) - } - ?: emptyList() - } - - val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted } - require(llmEmittedSlots.size <= 1) { - "Stage '${stageId.value}' declares ${llmEmittedSlots.size} LLM-emitted artifact kinds; " + - "only 0 or 1 is supported" - } - val responseFormat = llmEmittedSlots.firstOrNull() - ?.let { ResponseFormat.Json(it.kind.deriveJsonSchema()) } - ?: ResponseFormat.Text - - 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 - // across rounds; instead we grow our own ordered list and let the builder restamp - // ordinals from it each round. - val journalState = decisionJournalRepository.getJournal(sessionId) - val summaryText = journalState.summaryArtifactId?.let { id -> - artifactStore.get(id)?.toString(Charsets.UTF_8) - ?: run { - log.warn("Journal summary artifact {} not found in store — rendering without summary", id.value) - null - } - } - val journalText = decisionJournalRenderer.render(journalState, summaryText) - val journalEntries = if (journalText.isBlank()) emptyList() else listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L0, - content = journalText, - sourceType = "decisionJournal", - sourceId = "decision-journal", - tokenEstimate = estimateTokens(journalText), - role = EntryRole.SYSTEM, - ), - ) - val repoMapEntries = buildContextualRepoEntries(sessionId, stageId, stageConfig) - val sessionEvents = eventStore.read(sessionId) - val criticIds = criticArtifactIds(sessionEvents, graph, stageId) - val criticFrom = sessionEvents - .mapNotNull { it.payload as? RefinementIterationEvent } - .lastOrNull { it.cycleKey.endsWith("->${stageId.value}") } - ?.cycleKey?.substringBefore("->") - val needsEntries = buildNeedsArtifactEntries(sessionId, stageConfig, criticIds, criticFrom) - val profileEntries = session.state.boundProfile?.about - ?.takeIf { it.isNotBlank() } - ?.let { about -> - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L0, - content = "## Operator profile\n$about", - sourceType = "operatorProfile", - sourceId = "operator-profile", - tokenEstimate = about.length / 4, - role = EntryRole.SYSTEM, - ), - ) - } ?: emptyList() - val projectProfileEntries = session.state.boundProjectProfile - ?.let { listOf(buildProjectProfileEntry(it)) } ?: emptyList() - // AGENTS.md injection is disabled for now. The workspace-root AGENTS.md (the DOX framework) - // carries a mandatory "Read Before Editing" walk — "read the whole AGENTS.md chain, do not - // rely on memory, re-read before editing" — which drives a small model into an endless - // read-loop that never writes (2026-07-05 root cause). Reincorporating DOX properly (scoped - // to the stage's subproject, not the repo root; not read-mandating) is deferred. - val agentInstructionsEntries = emptyList() - val retryFeedbackEntries = buildRetryFeedbackEntry(sessionEvents, stageId) - ?.let { listOf(it) } ?: emptyList() - val recoveryTicketEntries = buildRecoveryTicketEntry(sessionEvents, stageId) - ?.let { listOf(it) } ?: emptyList() - val vocabularyEntries = artifactKindRegistry - ?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" } - ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList() - // Deterministic claim: a stage flagged claimTask gets the kernel to claim the next ready task - // (or re-surface the one already claimed) and inject its context bundle as L0 — the loop - // advances by recorded fact, not by the agent remembering to claim. - val claimedTaskEntries = if (stageConfig.metadata["claimTask"] == "true") { - taskClaimCoordinator?.claimNext(sessionId)?.let { bundle -> - listOf( - ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L0, - content = bundle, - sourceType = "claimedTask", - sourceId = stageId.value, - tokenEstimate = estimateTokens(bundle), - role = EntryRole.SYSTEM, - ), - ) - } ?: emptyList() - } else { - emptyList() - } - // Remaining-delta checklist (stage-termination design 2026-07-11 §1a): the stage's currently - // unmet file-contract assertions as a forward-looking, shrinking TODO. Injected pinned - // (neverDrop) so it survives the budget/dedup passes that otherwise wipe the model's memory of - // progress; refreshed inside the tool loop each time a write lands (§1b, cache-until-write). - var remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives) - val remainingDeltaEntries = remainingDeltaResults - ?.let { buildRemainingDeltaEntry(contractFailureItems(it)) } - ?.let { listOf(it) } ?: emptyList() - var accumulatedEntries = stampBuckets( - systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries + - journalEntries + repoMapEntries + claimedTaskEntries + - needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + - clarificationEntries + retryFeedbackEntries + recoveryTicketEntries + remainingDeltaEntries, - ) - val contextPack = runCatching { - contextPackBuilder.build( - id = ContextPackId(UUID.randomUUID().toString()), - sessionId = sessionId, - stageId = stageId, - entries = accumulatedEntries, - budget = TokenBudget(limit = stageConfig.tokenBudget), - ) - }.getOrElse { e -> - if (e is CancellationException) throw e - if (e !is RequiredContextOverflowException) throw e - // Required context doesn't fit — retrying with the same inputs can't succeed. - log.error("[Orchestrator] stage={}: {}", stageId.value, e.message) - return StageExecutionResult.Failure(e.message ?: "required context overflow", retryable = false) - } - emitContextTruncationIfNeeded(sessionId, stageId, contextPack) - - var currentContext = contextPack - var inferenceResult = runInference( - sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, - ) - var toolRounds = 0 - var consecutiveReadOnlyRounds = 0 - var consecutiveRejectedRounds = 0 - // Fingerprints ("tool:arguments") of every read-only call made during the current - // no-write streak. A round only counts against the read-loop breaker if it repeats calls - // already seen — a stage reading N distinct files before writing is legitimate context - // gathering, not a loop, and shouldn't trip the same counter as re-reading the same file. - val seenReadFingerprints = mutableSetOf() - // Set when the model produces its artifact via the emit_artifact tool instead of a final - // JSON message; overrides the post-loop capture of the (then-empty) assistant text. - var llmArtifactOverride: String? = null - - val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" } - fun owesFileWrite(): Boolean = fileWrittenSlots.isNotEmpty() && - fileWrittenSlots.any { artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() } - - // Gate: if the stage declares require_task_decompose, block stage_complete until at least one - // task_decompose (or task_create) call has completed successfully in this stage. Checked via - // ToolInvocationRequestedEvent + ToolExecutionCompletedEvent in the session event log. - // ponytail: scans events on every stage_complete attempt; negligible cost (in-memory list, rare call). - fun owesTaskDecompose(): Boolean { - if (stageConfig.metadata["requireTaskDecompose"] != "true") return false - val events = eventStore.read(sessionId) - val requestedIds = events - .mapNotNull { it.payload as? ToolInvocationRequestedEvent } - .filter { it.stageId == stageId && it.toolName in setOf("task_decompose", "task_create") } - .map { it.invocationId } - .toSet() - if (requestedIds.isEmpty()) return true - val completedIds = events - .mapNotNull { it.payload as? ToolExecutionCompletedEvent } - .filter { it.toolName in setOf("task_decompose", "task_create") && it.invocationId in requestedIds } - .map { it.invocationId } - .toSet() - return completedIds.isEmpty() - } - - // Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS). - // Returns the new result rather than mutating inferenceResult, to preserve smart casts. - suspend fun pushBack(nudge: String, forceWriteOnly: Boolean = false): InferenceResult { - accumulatedEntries = accumulatedEntries + ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L2, - sourceType = "toolResult", - sourceId = UUID.randomUUID().toString(), - content = nudge, - tokenEstimate = estimateTokens(nudge), - role = EntryRole.TOOL, - ) - currentContext = contextPackBuilder.build( - id = ContextPackId(UUID.randomUUID().toString()), - sessionId = sessionId, - stageId = stageId, - entries = accumulatedEntries, - budget = TokenBudget(limit = stageConfig.tokenBudget), - ) - emitContextTruncationIfNeeded(sessionId, stageId, currentContext) - toolRounds++ - return runInference( - sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, - forceWriteOnly = forceWriteOnly, - ) - } - - val writeNudge = "ERROR: you described the change but did not apply it. You MUST call the " + - "file_edit or file_write tool to write the change to the file before finishing — do " + - "not output the file contents as a message." - - val readLoopNudge = "STOP reading. You have read enough files and this stage has produced " + - "nothing yet. You MUST now call the file_write tool to create the required file(s) for " + - "this stage. Do not call file_read or list_dir again before you have written a file." - - while ( - inferenceResult is InferenceResult.Success && - toolRounds < tuning.maxToolRounds && - (inferenceResult.response.finishReason is FinishReason.ToolCall || owesFileWrite()) - ) { - // Content turn (no tool call) but the stage still owes a file_written artifact: the - // model emitted the change as prose instead of invoking the write tool. Nudge it to - // actually write (F-018). - if (inferenceResult.response.finishReason !is FinishReason.ToolCall) { - inferenceResult = pushBack(writeNudge) - continue - } - // emit_artifact is a meta-tool: the LLM calls it with the artifact's fields as arguments. - // Capture those arguments as the artifact content and exit the loop straight to validation - // (the post-loop capture honours this override instead of the empty assistant text). - val emitCall = inferenceResult.response.toolCalls.firstOrNull { it.function.name == EMIT_ARTIFACT_TOOL } - if (emitCall != null && llmEmittedSlots.isNotEmpty()) { - llmArtifactOverride = emitCall.function.arguments - break - } - // stage_complete is a meta-tool: the LLM calls it to signal the stage's goal is met. - // Skip tool dispatch — the loop exits and normal validation/transition follows - // (emitToolArtifacts + verifyProduces run on the success path after the loop). - if (inferenceResult.response.toolCalls.any { it.function.name == STAGE_COMPLETE_TOOL }) { - // Premature-completion guard: don't accept stage_complete while the stage still - // owes an artifact. For an LLM-emitted slot, nudge to emit the JSON (F-010); for a - // file_written slot, nudge to invoke the write tool (F-018). Otherwise complete. - val owedLlm = llmEmittedSlots.firstOrNull()?.takeIf { - inferenceResult.response.text.isBlank() && - artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() - } - when { - owedLlm != null -> inferenceResult = pushBack( - "ERROR: stage_complete is not allowed yet — you have not emitted the " + - "required '${owedLlm.name.value}' artifact. Call the emit_artifact tool with " + - "the fields filled in (or respond with the artifact as a single JSON message " + - "matching the provided schema).", - ) - owesFileWrite() -> inferenceResult = pushBack(writeNudge) - owesTaskDecompose() -> inferenceResult = pushBack( - "ERROR: stage_complete is not allowed yet — this stage requires a successful " + - "task_decompose (or task_create) call before it can complete. Call task_decompose " + - "with a parent epic and DEPENDS_ON-linked children representing the full task graph. " + - "Do not call stage_complete until task_decompose has returned successfully.", - ) - else -> break - } - continue - } - val toolEntries = dispatchToolCalls(sessionId, stageId, - inferenceResult.response.toolCalls, stageConfig, effectives, - approvalModeFor(session.state.boundProfile?.approvalMode), - inferenceResult.response.reasoning) - val fatalEntry = toolEntries.firstOrNull { it.content.startsWith("FATAL:") } - if (fatalEntry != null) { - emitProcessResultEvents(sessionId, stageId, stageConfig) - return StageExecutionResult.Failure( - fatalEntry.content.removePrefix("FATAL: "), - retryable = false, - ) - } - // Recoverable (ERROR:) tool failures are NOT terminal — they flow back into the - // loop as tool-result context so the model can see the error and adapt (bounded by - // MAX_TOOL_ROUNDS). Only FATAL: failures (handled above) abort the stage. - accumulatedEntries = accumulatedEntries + toolEntries - // Read-loop breaker: this round called only read-only tools yet the stage still owes a - // file_written artifact. Left alone the model keeps reading until MAX_TOOL_ROUNDS and - // never writes (F-018 nudges only cover a prose turn or a premature stage_complete, not - // an endless read-loop). After READ_LOOP_NUDGE_THRESHOLD such rounds, force the write - // nudge; a real write resets the counter so multi-file stages are unaffected. - if (owesFileWrite() && - inferenceResult.response.toolCalls.none { it.function.name in WRITE_TOOL_NAMES } - ) { - val roundFingerprints = inferenceResult.response.toolCalls - .map { "${it.function.name}:${it.function.arguments}" } - val madeProgress = roundFingerprints.any { it !in seenReadFingerprints } - seenReadFingerprints += roundFingerprints - if (madeProgress) { - consecutiveReadOnlyRounds = 0 - } else { - consecutiveReadOnlyRounds++ - if (consecutiveReadOnlyRounds >= tuning.readLoopNudgeThreshold) { - consecutiveReadOnlyRounds = 0 - inferenceResult = pushBack(readLoopNudge, forceWriteOnly = true) - continue - } - } - } else { - consecutiveReadOnlyRounds = 0 - seenReadFingerprints.clear() - } - // Rejection-loop breaker (stage-type agnostic): every tool result this round was a - // rejection (plane-2 BLOCKED or a stage/policy ERROR) and nothing succeeded. A stage that - // owes no file_written artifact (discovery, analyst) never trips the read-loop breaker, so - // without this it re-issues the same rejected call until MAX_TOOL_ROUNDS. After the - // threshold, tell it to stop retrying and produce its output; any success resets. - val toolResults = toolEntries.filter { it.role == EntryRole.TOOL } - val allRejected = toolResults.isNotEmpty() && - toolResults.all { it.content.startsWith("BLOCKED:") || it.content.startsWith("ERROR:") } - if (allRejected) { - consecutiveRejectedRounds++ - if (consecutiveRejectedRounds >= tuning.rejectionLoopNudgeThreshold) { - consecutiveRejectedRounds = 0 - inferenceResult = pushBack( - "STOP. Your last tool calls were all rejected and retrying the same paths will " + - "keep failing. Do not repeat them. Proceed with the information you already " + - "have: produce this stage's required output now (call stage_complete, or emit " + - "the required artifact). If a path you wanted does not exist yet, that is a " + - "fact to record — do not keep probing for it.", - ) - continue - } - } else { - consecutiveRejectedRounds = 0 - } - // Refresh the remaining-delta checklist only when a write actually landed this round - // (cache-until-write, §1b): reads can never change the contract verdict, so re-evaluating - // on them would just burn filesystem work and re-emit identical events. On a successful - // write, recompute contract-minus-reality and swap the pinned entry so the model watches - // the list shrink toward the empty=done signal it previously lacked. - val wroteThisRound = inferenceResult.response.toolCalls.any { it.function.name in WRITE_TOOL_NAMES } && - toolEntries.any { - it.role == EntryRole.TOOL && - !it.content.startsWith("ERROR:") && !it.content.startsWith("BLOCKED:") - } - if (wroteThisRound) { - remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives) - val refreshed = remainingDeltaResults?.let { buildRemainingDeltaEntry(contractFailureItems(it)) } - accumulatedEntries = accumulatedEntries.filterNot { it.sourceType == "remainingDelta" } + - listOfNotNull(refreshed) - } - currentContext = contextPackBuilder.build( - id = ContextPackId(UUID.randomUUID().toString()), - sessionId = sessionId, - stageId = stageId, - entries = accumulatedEntries, - budget = TokenBudget(limit = stageConfig.tokenBudget), - ) - emitContextTruncationIfNeeded(sessionId, stageId, currentContext) - inferenceResult = runInference( - sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, - ) - toolRounds++ - } - // Final clean emission. Producing the artifact while tools are in the request is unreliable on - // some local chat templates (Gemma leaks `<|channel>` markers into the text; other models keep - // tool-calling until the round cap and never emit). For an LLM-emitted stage that hasn't already - // produced its artifact via the emit_artifact tool, fire ONE tools-less inference — verified to - // yield clean JSON deterministically. Skipped when the model already emitted clean final text. - if (inferenceResult is InferenceResult.Success && llmEmittedSlots.isNotEmpty() && llmArtifactOverride == null) { - val last = inferenceResult.response - val needsCleanEmission = last.finishReason is FinishReason.ToolCall || - last.text.isBlank() || last.text.contains("<|channel") - if (needsCleanEmission && !isCancelled(sessionId)) { - val nudge = "Stop calling tools. Output the required '${llmEmittedSlots.first().name.value}' " + - "artifact now as a single JSON object matching the schema — no tool calls, no commentary." - accumulatedEntries = accumulatedEntries + ContextEntry( - id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L2, - sourceType = "toolResult", - sourceId = UUID.randomUUID().toString(), - content = nudge, - tokenEstimate = estimateTokens(nudge), - role = EntryRole.TOOL, - ) - currentContext = contextPackBuilder.build( - id = ContextPackId(UUID.randomUUID().toString()), - sessionId = sessionId, - stageId = stageId, - entries = accumulatedEntries, - budget = TokenBudget(limit = stageConfig.tokenBudget), - ) - inferenceResult = runInference( - sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, - responseFormat, effectives, withTools = false, - ) - } - } - return when (inferenceResult) { - is InferenceResult.Cancelled -> StageExecutionResult.Failure("CANCELLED", retryable = false) - is InferenceResult.Failed -> - StageExecutionResult.Failure(inferenceResult.reason, retryable = isRetryableFailure(inferenceResult.reason)) - is InferenceResult.Success -> { - if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false) - // Capture the LLM-emitted artifact's JSON so transition conditions - // (e.g. ArtifactFieldEquals on a reviewer's verdict) can read it. An emit_artifact - // tool call (llmArtifactOverride) supplies the content directly; otherwise the - // artifact is the final assistant message text. - val rawArtifactText = llmArtifactOverride ?: inferenceResult.response.text - val slot = llmEmittedSlots.firstOrNull() - // Artifact-emission ladder: deterministic repair (prose/fences/trailing commas), then - // — rarely, gated by the failure classifier — one grammar-constrained LLM-repair rung, - // before the text is cached/validated. On a hard failure the policy decision - // short-circuits the stage. Only the nondeterministic rung records events (spec 2026-07-04 §6). - val artifactText = if (slot != null) { - when (val res = artifactExtractionPipeline.run(rawArtifactText, slot.kind.deriveJsonSchema())) { - is ArtifactExtractionPipeline.ExtractionResult.Resolved -> { - if (res.repaired) { - emitArtifactRepairAttempted(sessionId, stageId, slot, ArtifactFailure.FORMATTING, "DETERMINISTIC") - emitArtifactRepairResolved(sessionId, stageId, slot, res.canonicalJson.toString()) - } - res.canonicalJson.toString() - } - is ArtifactExtractionPipeline.ExtractionResult.Unresolved -> - when (val ladder = repairArtifact(sessionId, stageId, slot, res, stageConfig, effectives, config.stageTimeoutMs)) { - is ArtifactLadderOutcome.Text -> ladder.text - is ArtifactLadderOutcome.Reject -> return ladder.failure - } - } - } else { - rawArtifactText - } - val artifactHash = if (artifactText != rawArtifactText || llmArtifactOverride != null) { - artifactStore.put(artifactText.toByteArray()) - } else { - inferenceResult.responseArtifactId - } - slot?.let { slot -> - artifactContentCache["${sessionId.value}:${slot.name.value}"] = artifactText - artifactHash?.let { hash -> - emit( - sessionId, - ArtifactContentStoredEvent( - artifactId = slot.name, - contentHash = hash, - sessionId = sessionId, - stageId = stageId, - ), - ) - } - } - val validationCtx = ValidationContext( - graph = graph, - sessionState = session.state, - availableTools = effectives.registry?.all()?.map { it.name }?.toSet(), - producedArtifactContent = graph.stages.values.flatMap { it.produces }.mapNotNull { slot -> - artifactContentCache["${sessionId.value}:${slot.name.value}"]?.let { slot.name.value to it } - }.toMap(), - ) - when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) { - is StageExecutionResult.Success -> { - emitToolArtifacts(sessionId, stageId, stageConfig) - emitLlmArtifacts(sessionId, stageId, stageConfig) - runPostStageGates( - sessionId, stageId, stageConfig, effectives, - session.state.boundProjectProfile?.commands ?: emptyMap(), - ) - } - - is StageExecutionResult.Failure -> outcome - } - } - } - } - - private sealed interface ArtifactLadderOutcome { - data class Text(val text: String) : ArtifactLadderOutcome - data class Reject(val failure: StageExecutionResult.Failure) : ArtifactLadderOutcome - } /** * Handles an [ArtifactExtractionPipeline.ExtractionResult.Unresolved]: attempts one grammar-constrained * LLM-repair rung (only for SCHEMA / MISSING_INFO — the classifier gates the expensive rung), then * consults the policy [decide] to map a hard failure onto an orchestrator action. Records the §6 events. */ - private suspend fun repairArtifact( + internal suspend fun repairArtifact( sessionId: SessionId, stageId: StageId, slot: TypedArtifactSlot, @@ -1022,7 +446,7 @@ abstract class SessionOrchestrator( } /** One tools-less, grammar-constrained inference asking the model to repair its own malformed artifact. */ - private suspend fun runArtifactRepairInference( + internal suspend fun runArtifactRepairInference( sessionId: SessionId, stageId: StageId, slot: TypedArtifactSlot, @@ -1058,7 +482,7 @@ abstract class SessionOrchestrator( return (result as? InferenceResult.Success)?.response?.text } - private suspend fun emitArtifactRepairAttempted( + internal suspend fun emitArtifactRepairAttempted( sessionId: SessionId, stageId: StageId, slot: TypedArtifactSlot, @@ -1069,7 +493,7 @@ abstract class SessionOrchestrator( ArtifactRepairAttemptedEvent(slot.name, classification.name, rung, sessionId, stageId), ) - private suspend fun emitArtifactRepairResolved( + internal suspend fun emitArtifactRepairResolved( sessionId: SessionId, stageId: StageId, slot: TypedArtifactSlot, @@ -1079,7 +503,7 @@ abstract class SessionOrchestrator( emit(sessionId, ArtifactRepairResolvedEvent(slot.name, hash, sessionId, stageId)) } - private fun error( + internal fun error( stageId: StageId, path: String, throwable: Throwable, @@ -1091,7 +515,7 @@ abstract class SessionOrchestrator( * is a single fenced block. Conservative: only acts when the trimmed text both starts and ends * with a fence, so prose responses that merely contain an inline block are left untouched. */ - private fun stripCodeFence(text: String): String { + internal fun stripCodeFence(text: String): String { val trimmed = text.trim() if (!trimmed.startsWith("```") || !trimmed.endsWith("```")) return text val withoutClose = trimmed.removeSuffix("```").trimEnd() @@ -1105,7 +529,7 @@ abstract class SessionOrchestrator( // Maps the operator profile's free-text approval_mode onto the engine's [ApprovalMode]. // Unset/unknown falls back to PROMPT so an absent profile keeps the human-in-the-loop // default; only an explicit auto/yolo opts a session into unattended approval. - private fun approvalModeFor(profileMode: String?): ApprovalMode = + internal fun approvalModeFor(profileMode: String?): ApprovalMode = when (profileMode?.trim()?.lowercase()) { "deny" -> ApprovalMode.DENY "auto" -> ApprovalMode.AUTO @@ -1121,12 +545,11 @@ abstract class SessionOrchestrator( // arguments" as "retry the same call" rather than "try a different path" — the arguments were // valid, the path simply doesn't exist. The schema reference stays so malformed calls can still // self-correct; the model must infer the tactic from the tool's error text. - private fun toolArgsHint(tool: Tool?): String = + internal fun toolArgsHint(tool: Tool?): String = tool?.let { "\nAccepted parameters for '${it.name}': ${it.parametersSchema}" }.orEmpty() - private data class RenderedToolResult(val content: String, val fullOutputHash: String?) /** * Render a tool result into the consistently-framed, bounded text the model sees. Success output @@ -1137,7 +560,7 @@ abstract class SessionOrchestrator( * rest. Failures keep their `ERROR:`/`FATAL:` sentinels unchanged — the all-rejected loop breaker * keys on those prefixes. */ - private suspend fun renderToolResult(toolName: String, tool: Tool?, result: ToolResult): RenderedToolResult = + internal suspend fun renderToolResult(toolName: String, tool: Tool?, result: ToolResult): RenderedToolResult = when (result) { is ToolResult.Failure -> RenderedToolResult( if (!result.recoverable) "FATAL: ${result.reason}" else "ERROR: ${result.reason}${toolArgsHint(tool)}", @@ -1158,7 +581,7 @@ abstract class SessionOrchestrator( } } - private suspend fun dispatchToolCalls( + internal suspend fun dispatchToolCalls( sessionId: SessionId, stageId: StageId, toolCalls: List, @@ -1555,13 +978,13 @@ abstract class SessionOrchestrator( * scope — block the task so the loop advances instead of the implementer retrying the same * out-of-scope write. Only fires for the scope-proposal tool; other denied tools are unaffected. */ - private suspend fun blockTaskOnScopeRejection(sessionId: SessionId, toolName: String, reason: String) { + internal suspend fun blockTaskOnScopeRejection(sessionId: SessionId, toolName: String, reason: String) { if (toolName == SCOPE_PROPOSAL_TOOL) { taskClaimCoordinator?.blockActiveTask(sessionId, "scope amendment rejected: $reason") } } - private suspend fun runPlane2Assessment( + internal suspend fun runPlane2Assessment( sessionId: SessionId, stageId: StageId, invocationId: ToolInvocationId, @@ -1634,7 +1057,7 @@ abstract class SessionOrchestrator( * touches the filesystem. Symlink escapes and privileged locations are caught upstream by the * plane-2 PathContainmentRule (BLOCK before any prompt), and by the tool's own realize() jail. */ - private fun outsideWorkspaceTarget(parameters: Map, workspaceRoot: Path?): Path? { + internal fun outsideWorkspaceTarget(parameters: Map, workspaceRoot: Path?): Path? { workspaceRoot ?: return null val raw = parameters["path"] as? String ?: return null val candidate = runCatching { Path.of(raw) }.getOrNull() ?: return null @@ -1647,7 +1070,7 @@ abstract class SessionOrchestrator( * Paths outside the workspace the operator has approved this session (folded from * [OutsidePathAccessGrantedEvent]). Replay-safe: reads the log, never re-prompts or re-probes. */ - private fun grantedOutsidePaths(sessionId: SessionId): Set = + internal fun grantedOutsidePaths(sessionId: SessionId): Set = eventStore.read(sessionId) .mapNotNull { it.payload as? OutsidePathAccessGrantedEvent } .filter { it.sessionId == sessionId } @@ -1659,7 +1082,7 @@ abstract class SessionOrchestrator( * Active from the moment a READ_BEFORE_WRITE block lands until a FILE_READ completion follows it. * Derived purely from the event log — no separate flag stored. */ - private fun isReadOnlyMode(sessionId: SessionId): Boolean { + internal fun isReadOnlyMode(sessionId: SessionId): Boolean { var blocked = false val pendingReadIds = mutableSetOf() for (event in eventStore.read(sessionId)) { @@ -1697,7 +1120,7 @@ abstract class SessionOrchestrator( return blocked } - private suspend fun buildSchemaEntries( + internal suspend fun buildSchemaEntries( responseFormat: ResponseFormat, stageId: StageId, ): List { @@ -1719,7 +1142,7 @@ abstract class SessionOrchestrator( ) } - private suspend fun buildSteeringNoteEntries(sessionId: SessionId): List { + internal suspend fun buildSteeringNoteEntries(sessionId: SessionId): List { val events = eventStore.read(sessionId) var locked = false return events.mapNotNull { event -> @@ -1775,7 +1198,7 @@ abstract class SessionOrchestrator( * 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 { + internal suspend fun buildClarificationAnswerEntries(sessionId: SessionId): List { val events = eventStore.read(sessionId) val answered = events.mapNotNull { it.payload as? ClarificationAnsweredEvent } if (answered.isEmpty()) return emptyList() @@ -1846,7 +1269,7 @@ abstract class SessionOrchestrator( * (2026-07-05 context-poisoning root cause). Reads the latest [RepoMapComputedEvent] from * the log (replay-safe — never re-scans the FS). */ - private suspend fun buildRepoMapEntries(sessionId: SessionId, stagePrompt: String = ""): List { + internal suspend fun buildRepoMapEntries(sessionId: SessionId, stagePrompt: String = ""): List { val map = eventStore.read(sessionId) .mapNotNull { it.payload as? RepoMapComputedEvent } .lastOrNull() ?: return emptyList() @@ -1890,7 +1313,7 @@ abstract class SessionOrchestrator( * The purpose-built recovery arbiter (role=recovery) is excluded: it is also ticket-entered but * its own prompt IS a tuned repair prompt, so it keeps it. Replay-safe (recorded events only). */ - private fun mandateSuppressedByTicket(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): Boolean { + internal fun mandateSuppressedByTicket(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): Boolean { if (stageConfig.metadata["role"] == "recovery") return false return eventStore.read(sessionId) .mapNotNull { it.payload as? TransitionExecutedEvent } @@ -1904,7 +1327,7 @@ abstract class SessionOrchestrator( * `com/correx/apps/server/...`) still resolves. Null when nothing shares the filename, so we * only ever suggest a concrete, existing file. Reads the recorded map (replay-safe). */ - private fun suggestClosestPath(sessionId: SessionId, guessed: String): String? { + internal fun suggestClosestPath(sessionId: SessionId, guessed: String): String? { val map = eventStore.read(sessionId) .mapNotNull { it.payload as? RepoMapComputedEvent } .lastOrNull() ?: return null @@ -1919,7 +1342,7 @@ abstract class SessionOrchestrator( * similarity scores never reflect genuine relevance (2026-07-08 finding: no file scored a * clear match, everything clustered in the embedder's baseline noise band for short strings). */ - private fun resolveStagePromptText(stageId: StageId, stageConfig: StageConfig): String { + internal fun resolveStagePromptText(stageId: StageId, stageConfig: StageConfig): String { stageConfig.metadata["promptInline"]?.trim()?.takeIf { it.isNotBlank() }?.let { return it } stageConfig.metadata["prompt"]?.let { path -> runCatching { promptResolver.resolve(path) }.getOrNull() @@ -1938,7 +1361,7 @@ abstract class SessionOrchestrator( * finding: without it, the query matched a tool script's own symbol names over the files the * user's request actually named). */ - private fun repoKnowledgeQuery(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): String { + internal fun repoKnowledgeQuery(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): String { val intent = eventStore.read(sessionId) .mapNotNull { it.payload as? InitialIntentEvent } .firstOrNull()?.intent?.trim() @@ -1946,7 +1369,7 @@ abstract class SessionOrchestrator( return if (intent.isNullOrBlank()) stagePrompt else "$intent\n\n$stagePrompt" } - private suspend fun buildContextualRepoEntries( + internal suspend fun buildContextualRepoEntries( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -1981,7 +1404,7 @@ abstract class SessionOrchestrator( * [buildRepoMapEntries]/[repoEntriesOrMapFloor], which still exclude doc *content* from the source * layout for non-doc prompts. Reads the recorded map (replay-safe — never re-scans the FS). */ - private suspend fun buildDocsCatalogEntry(sessionId: SessionId): List { + internal suspend fun buildDocsCatalogEntry(sessionId: SessionId): List { val map = eventStore.read(sessionId) .mapNotNull { it.payload as? RepoMapComputedEvent } .lastOrNull() ?: return emptyList() @@ -2020,7 +1443,7 @@ abstract class SessionOrchestrator( * above real source for a code task, re-poisoning the context the floor's gate was added to fix * (2026-07-05 root cause). They only survive when the stage prompt explicitly asks about docs. */ - private suspend fun repoEntriesOrMapFloor( + internal suspend fun repoEntriesOrMapFloor( sessionId: SessionId, hits: List, stagePrompt: String, @@ -2041,11 +1464,11 @@ abstract class SessionOrchestrator( * journal, profiles, vocabulary, transcript) is OPTIONAL and shares the remainder under * per-type ceilings. */ - private fun stampBuckets(entries: List): List = entries.map { + internal fun stampBuckets(entries: List): List = entries.map { if (it.sourceType in REQUIRED_SOURCE_TYPES) it.copy(bucket = ContextBucket.REQUIRED) else it } - private suspend fun buildNeedsArtifactEntries( + internal suspend fun buildNeedsArtifactEntries( sessionId: SessionId, stageConfig: StageConfig, criticIds: Set = emptySet(), @@ -2079,7 +1502,7 @@ abstract class SessionOrchestrator( } } - private fun parseFileWrittenArtifact(json: String): FileWrittenArtifact? = + internal fun parseFileWrittenArtifact(json: String): FileWrittenArtifact? = runCatching { Json.decodeFromString(FileWrittenArtifact.serializer(), json) }.getOrNull() /** @@ -2089,7 +1512,7 @@ abstract class SessionOrchestrator( * over existing events — no new artifact kind, no producer change, replay-safe. Null when no * writes are on record (caller falls back to the cached single-file JSON). */ - private fun fileWrittenManifest(sessionId: SessionId, needed: ArtifactId): String? { + internal fun fileWrittenManifest(sessionId: SessionId, needed: ArtifactId): String? { val events = eventStore.read(sessionId) val stageIds = events.mapNotNull { it.payload as? ArtifactContentStoredEvent } .filter { it.artifactId == needed } @@ -2111,7 +1534,7 @@ abstract class SessionOrchestrator( }.trimEnd() } - private suspend fun emitToolArtifacts( + internal suspend fun emitToolArtifacts( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2139,7 +1562,7 @@ abstract class SessionOrchestrator( } } - private suspend fun recordToolExecution( + internal suspend fun recordToolExecution( sessionId: SessionId, stageId: StageId, toolCall: ToolCallRequest, @@ -2204,7 +1627,7 @@ abstract class SessionOrchestrator( * store it as the stage's file_written slot content, and record [ArtifactContentStoredEvent]. * This ties the file_written artifact to a real write so validation can no longer be faked. */ - private suspend fun materializeFileWritten( + internal suspend fun materializeFileWritten( sessionId: SessionId, stageId: StageId, request: ToolRequest, @@ -2240,7 +1663,7 @@ abstract class SessionOrchestrator( } } - private suspend fun emitLlmArtifacts( + internal suspend fun emitLlmArtifacts( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2270,7 +1693,7 @@ abstract class SessionOrchestrator( } } - private fun verifyProduces( + internal fun verifyProduces( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2317,7 +1740,7 @@ abstract class SessionOrchestrator( * by [verifyProduces]; this method only EMITS — a [StageCheckpointFailedEvent] accompanies that * halt and surfaces *why* the plan diverged. */ - private suspend fun emitStageCheckpoint( + internal suspend fun emitStageCheckpoint( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2343,7 +1766,7 @@ abstract class SessionOrchestrator( * (invariant #9) — replay reads it back, never re-probes. A hallucinated path fails the stage * (retryable), feeding the missing paths back so the model re-grounds on the next attempt. */ - private suspend fun groundBriefReferences( + internal suspend fun groundBriefReferences( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2394,7 +1817,7 @@ abstract class SessionOrchestrator( * falls back to the single needs artifact present. When there are multiple needs and none is * named "analysis", the gate skips (ambiguous brief) and returns Success. */ - private suspend fun checkBriefEcho( + internal suspend fun checkBriefEcho( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2452,7 +1875,7 @@ abstract class SessionOrchestrator( * the heavier process-spawning static-analysis step last, so it only runs once a stage is * otherwise sound. */ - private suspend fun runPostStageGates( + internal suspend fun runPostStageGates( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2501,7 +1924,7 @@ abstract class SessionOrchestrator( * ([buildRemainingDeltaEntry]) so both read the same contract-minus-reality computation from a * single evaluation + single event emission. */ - private suspend fun evaluateStageContract( + internal suspend fun evaluateStageContract( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2536,10 +1959,10 @@ abstract class SessionOrchestrator( } /** The currently-failing assertions as (target, assertionId, evidence) triples for the checklist. */ - private fun contractFailureItems(results: List): List> = + internal fun contractFailureItems(results: List): List> = results.filterNot { it.passed }.map { Triple(it.target, it.assertionId, it.evidence) } - private suspend fun runContractGate( + internal suspend fun runContractGate( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2573,7 +1996,7 @@ abstract class SessionOrchestrator( * recorded in a [PlanCompileCheckedEvent] (invariant #9) so replay reads it back and never * re-compiles. */ - private suspend fun runPlanCompileGate( + internal suspend fun runPlanCompileGate( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2618,7 +2041,7 @@ abstract class SessionOrchestrator( * "project" instead of letting it reach COMPLETED. A docs-only plan writes neither, so it is left * alone. */ - private fun sessionProducedBuildTarget(sessionId: SessionId): Boolean = + internal fun sessionProducedBuildTarget(sessionId: SessionId): Boolean = eventStore.read(sessionId) .mapNotNull { it.payload as? FileWrittenEvent } .filter { it.postImageHash != null } @@ -2630,7 +2053,7 @@ abstract class SessionOrchestrator( KindContractTable.assertionsFor(kind, path).any { it.id == "imports_resolve" } } - private fun stageWrittenPaths(sessionId: SessionId, stageId: StageId): List { + internal fun stageWrittenPaths(sessionId: SessionId, stageId: StageId): List { val events = eventStore.read(sessionId) val invocationIds = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent } .filter { it.stageId == stageId } @@ -2653,7 +2076,7 @@ abstract class SessionOrchestrator( * static-clean output ever reaches the downstream LLM reviewer — its context thereby excludes * anything static tools already catch, with no parsing or context plumbing. */ - private suspend fun runStaticAnalysis( + internal suspend fun runStaticAnalysis( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2709,7 +2132,7 @@ abstract class SessionOrchestrator( * run fails the stage retryably with the build output fed back (recorded as a * [StaticAnalysisCompletedEvent], invariant #9, so replay never re-runs it). */ - private suspend fun runExecutionGate( + internal suspend fun runExecutionGate( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2770,7 +2193,7 @@ abstract class SessionOrchestrator( * [StageExecutionResult.Failure] when setup itself fails (a real, actionable error), else null (no * setup configured, or it succeeded — proceed to the build). */ - private suspend fun runSetupCommand( + internal suspend fun runSetupCommand( sessionId: SessionId, stageId: StageId, workspaceRoot: Path, @@ -2809,7 +2232,7 @@ abstract class SessionOrchestrator( * stage; after the budget, findings surface without blocking. Suboptimal/opinion findings never * block. Recorded as an environment observation (invariant #9); replay never re-runs the reviewer. */ - private suspend fun runReviewGate( + internal suspend fun runReviewGate( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2868,7 +2291,7 @@ abstract class SessionOrchestrator( ) } - private suspend fun emitProcessResultEvents( + internal suspend fun emitProcessResultEvents( sessionId: SessionId, stageId: StageId, stageConfig: StageConfig, @@ -2904,7 +2327,7 @@ abstract class SessionOrchestrator( // emitting free-form JSON as a final message. Tool-calling models handle this far more reliably, // and it sidesteps llama.cpp's grammar+tools incompatibility (the artifact can't be grammar- // constrained while tools are present). Empty when the stage has no LLM-emitted slot. - private fun emitArtifactTool(stageConfig: StageConfig): List { + internal fun emitArtifactTool(stageConfig: StageConfig): List { val slot = stageConfig.produces.firstOrNull { it.kind.llmEmitted } ?: return emptyList() val schema = Json.encodeToJsonElement(JsonSchema.serializer(), slot.kind.deriveJsonSchema()).jsonObject return listOf( @@ -2923,7 +2346,7 @@ abstract class SessionOrchestrator( // RetryAttemptedEvent.failureReason that the retry-feedback entry feeds back to the model, so a // weak model is told *what* was wrong (e.g. "required property 'summary' missing") instead of a // bare "validation failed" it can't act on. Capped to keep the feedback prompt small. - private fun summarizeRejection(report: ValidationReport): String { + internal fun summarizeRejection(report: ValidationReport): String { val errors = report.sections.flatMap { section -> section.issues .filter { it.severity == ValidationSeverity.ERROR } @@ -3080,7 +2503,7 @@ abstract class SessionOrchestrator( // request-construction failure — retrying re-sends the identical bad request and exhausts the // budget for nothing (F-002). Treat 4xx as terminal, except 408 (timeout) and 429 (rate-limit) // which are transient. Everything else (network/5xx/unknown) stays retryable. - private fun isRetryableFailure(reason: String): Boolean { + internal fun isRetryableFailure(reason: String): Boolean { val status = Regex("""returned\s+(\d{3})""").find(reason)?.groupValues?.get(1)?.toIntOrNull() ?: return true return status !in 400..499 || status == HTTP_TIMEOUT || status == HTTP_TOO_MANY_REQUESTS @@ -3201,7 +2624,7 @@ abstract class SessionOrchestrator( * recorded (the LLM-side finding emission is a separate, model-gated activation), and idempotent * — it skips when outcomes already exist — so it is safe on every terminal path. */ - private suspend fun correlateCritiqueOutcomes(sessionId: SessionId) { + internal suspend fun correlateCritiqueOutcomes(sessionId: SessionId) { val events = eventStore.read(sessionId) if (events.any { it.payload is CritiqueOutcomeCorrelatedEvent }) return val reviews = events.mapNotNull { it.payload as? CritiqueFindingsRecordedEvent } @@ -3229,7 +2652,7 @@ abstract class SessionOrchestrator( // Surface context-budget overflow as an event on the workflow path. The builder already // enforces the budget (drops oldest/compressible entries, pinning steering + event history); // this records that it happened so replay and operators can see it, mirroring the router path. - private suspend fun emitContextTruncationIfNeeded( + internal suspend fun emitContextTruncationIfNeeded( sessionId: SessionId, stageId: StageId, contextPack: ContextPack, @@ -3290,7 +2713,7 @@ abstract class SessionOrchestrator( // --- token estimation --- - protected open suspend fun estimateTokens(content: String): Int { + internal open suspend fun estimateTokens(content: String): Int { val t = tokenizer if (t != null) { // A zero count for non-blank content means the tokenizer endpoint is lying @@ -3304,7 +2727,7 @@ abstract class SessionOrchestrator( return fallbackTokenEstimate(content) } - private fun fallbackTokenEstimate(content: String): Int { + internal fun fallbackTokenEstimate(content: String): Int { return (content.length / 4).coerceAtLeast(1) } @@ -3373,7 +2796,7 @@ abstract class SessionOrchestrator( // --- private functions --- - private suspend fun handleApproval( + internal suspend fun handleApproval( sessionId: SessionId, stageId: StageId, outcome: ValidationOutcome.NeedsApproval, @@ -3421,7 +2844,7 @@ abstract class SessionOrchestrator( } } - private suspend fun buildApprovalRequest( + internal suspend fun buildApprovalRequest( sessionId: SessionId, stageId: StageId, outcome: ValidationOutcome.NeedsApproval, @@ -3453,7 +2876,7 @@ abstract class SessionOrchestrator( ) } - private suspend fun emitDecisionResolved( + internal suspend fun emitDecisionResolved( sessionId: SessionId, domainRequest: DomainApprovalRequest, decision: ApprovalDecision, @@ -3483,7 +2906,7 @@ abstract class SessionOrchestrator( * This function performs blocking I/O (file reads) and must be called from a suspend context * that will dispatch it on [Dispatchers.IO]. */ -private suspend fun computeToolPreview( +internal suspend fun computeToolPreview( toolName: String, parameters: Map, workspaceRoot: java.nio.file.Path?, @@ -3502,7 +2925,7 @@ private suspend fun computeToolPreview( return buildDiffString(path, existingContent, proposedContent) } -private suspend fun readFileIfExists(path: String, workspaceRoot: java.nio.file.Path?): String? = +internal suspend fun readFileIfExists(path: String, workspaceRoot: java.nio.file.Path?): String? = withContext(Dispatchers.IO) { runCatching { // Resolve relative paths against the session's workspace root, same as the tools do — @@ -3522,7 +2945,7 @@ private suspend fun readFileIfExists(path: String, workspaceRoot: java.nio.file. * to the raw-JSON-args fallback — simulating `patch -p1` application here isn't worth the risk of * the preview silently disagreeing with what the tool actually does. */ -private suspend fun computeFileEditPreview( +internal suspend fun computeFileEditPreview( parameters: Map, workspaceRoot: java.nio.file.Path?, ): String? { @@ -3550,7 +2973,7 @@ private suspend fun computeFileEditPreview( * line — rather than the raw `arguments.take(200)` fallback — keeps the command both * untruncated and human-readable; the TUI's deterministic parser re-tokenizes it. */ -private fun shellCommandPreview(parameters: Map): String? { +internal fun shellCommandPreview(parameters: Map): String? { val argv = when (val raw = parameters["argv"]) { is List<*> -> raw.filterIsInstance() is String -> runCatching { Json.decodeFromString>(raw) }.getOrElse { emptyList() } @@ -3561,21 +2984,21 @@ private fun shellCommandPreview(parameters: Map): String? { } /** Single-quote a token when it carries whitespace or shell metacharacters. */ -private fun shellQuoteToken(token: String): String { +internal fun shellQuoteToken(token: String): String { if (token.isEmpty()) return "''" val needsQuote = token.any { it.isWhitespace() || it in "\"'\\|&;<>(){}\$`*?[]~#!" } return if (needsQuote) "'" + token.replace("'", "'\\''") + "'" else token } -private enum class DiffOpKind { EQUAL, DELETE, INSERT } -private data class DiffOp(val kind: DiffOpKind, val line: String) +internal enum class DiffOpKind { EQUAL, DELETE, INSERT } +internal data class DiffOp(val kind: DiffOpKind, val line: String) /** * Build a unified-diff string for a full file replacement, using a line-level LCS diff * so a small edit to a large file shows as a tight hunk rather than a full-file replacement. * When the file doesn't exist yet (new file), only `+` lines are shown. */ -private fun buildDiffString(path: String, existingContent: String?, proposedContent: String): String { +internal fun buildDiffString(path: String, existingContent: String?, proposedContent: String): String { val existingLines = existingContent?.lines() ?: emptyList() val proposedLines = proposedContent.lines() if (existingLines == proposedLines) return " (no change)" @@ -3589,7 +3012,7 @@ private fun buildDiffString(path: String, existingContent: String?, proposedCont } /** Line-level LCS diff producing an edit script of EQUAL/DELETE/INSERT ops. */ -private fun diffLines(oldLines: List, newLines: List): List { +internal fun diffLines(oldLines: List, newLines: List): List { val n = oldLines.size val m = newLines.size val lcs = Array(n + 1) { IntArray(m + 1) } @@ -3634,7 +3057,7 @@ private fun diffLines(oldLines: List, newLines: List): List, context: Int = 3): String { +internal fun renderHunks(ops: List, context: Int = 3): String { val changedIndices = ops.indices.filter { ops[it].kind != DiffOpKind.EQUAL } if (changedIndices.isEmpty()) return "" @@ -3740,7 +3163,7 @@ internal sealed interface InferenceResult { // values. Strict parse is tried first; only genuinely-off syntax hits this. Triple-quoted // Python-style values (`"""..."""`) still fail here — that ambiguity isn't worth a fragile // hand-rolled repair; the loud "unparseable" path below tells the model to fix its format. -private val lenientJson = Json { isLenient = true } +internal val lenientJson = Json { isLenient = true } internal fun parseToolArguments(arguments: String): Map = runCatching { val element = runCatching { Json.parseToJsonElement(arguments) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt new file mode 100644 index 00000000..96fd7d31 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt @@ -0,0 +1,741 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.GRANT_LEDGER_SESSION_ID +import com.correx.core.approvals.ProjectIdentity +import com.correx.core.approvals.Tier +import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.domain.ApprovalEngine +import com.correx.core.approvals.isAtMost +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.approvals.model.DomainApprovalRequest +import com.correx.core.artifacts.ArtifactState +import com.correx.core.artifacts.kind.ArtifactKindRegistry +import com.correx.core.artifacts.kind.KindContractTable +import com.correx.core.artifacts.kind.KindInference +import com.correx.core.artifacts.kind.JsonSchema +import com.correx.core.artifacts.kind.FileWrittenArtifact +import com.correx.core.artifacts.kind.ProcessResultArtifact +import com.correx.core.artifacts.kind.TypedArtifactSlot +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.context.builder.ContextPackBuilder +import com.correx.core.context.builder.RequiredContextOverflowException +import com.correx.core.context.model.ContextBucket +import com.correx.core.context.model.ContextEntry +import com.correx.core.context.model.ContextLayer +import com.correx.core.context.model.ContextPack +import com.correx.core.context.model.TokenBudget +import com.correx.core.context.model.EntryRole +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.ArtifactContentStoredEvent +import com.correx.core.events.events.FileWrittenEvent +import com.correx.core.events.events.ContractGateEvaluatedEvent +import com.correx.core.events.events.PlanCompileCheckedEvent +import com.correx.core.events.events.ReviewFindingsRaisedEvent +import com.correx.core.events.events.ReviewVerdict +import com.correx.core.events.events.ContractAssertionResult +import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ArtifactRepairAttemptedEvent +import com.correx.core.events.events.ArtifactRepairFailedEvent +import com.correx.core.events.events.ArtifactRepairResolvedEvent +import com.correx.core.events.events.CritiqueFindingsRecordedEvent +import com.correx.core.events.events.CritiqueOutcomeCorrelatedEvent +import com.correx.core.events.events.BriefEchoMismatchEvent +import com.correx.core.events.events.BriefGroundingCheckedEvent +import com.correx.core.events.events.StaticAnalysisCompletedEvent +import com.correx.core.events.events.StaticAnalysisFinding +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 +import com.correx.core.events.events.ArtifactValidatingEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.RepoKnowledgeHit +import com.correx.core.events.events.InitialIntentEvent +import com.correx.core.events.events.RepoKnowledgeRetrievedEvent +import com.correx.core.events.events.RepoMapComputedEvent +import com.correx.core.events.events.InferenceFailedEvent +import com.correx.core.events.events.InferenceStartedEvent +import com.correx.core.events.events.InferenceTimeoutEvent +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.OutsidePathAccessGrantedEvent +import com.correx.core.events.events.RiskAssessedEvent +import com.correx.core.events.events.ExecutionPlanLockedEvent +import com.correx.core.events.events.StageCheckpointFailedEvent +import com.correx.core.events.events.StageCheckpointPassedEvent +import com.correx.core.events.events.SteeringNoteAddedEvent +import com.correx.core.events.events.ToolExecutionCompletedEvent +import com.correx.core.events.events.ToolExecutionFailedEvent +import com.correx.core.events.events.ToolExecutionRejectedEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.ToolReceipt +import com.correx.core.events.events.ToolRequest +import com.correx.core.events.risk.RiskAction +import com.correx.core.events.risk.RiskSummary +import com.correx.core.toolintent.SessionContextProjection +import com.correx.core.toolintent.ToolCallAssessmentInput +import com.correx.core.toolintent.ToolCallAssessor +import com.correx.core.toolintent.WorkspacePolicy +import com.correx.core.toolintent.WorldProbe +import com.correx.core.toolintent.toAssessedIssues +import com.correx.core.toolintent.toRiskSummary +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.sessions.projections.EgressAllowlistProjection +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 +import com.correx.core.events.types.InferenceRequestId +import com.correx.core.events.types.RiskSummaryId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.ToolInvocationId +import com.correx.core.events.types.ValidationReportId +import com.correx.core.inference.FinishReason +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceRequest +import com.correx.core.inference.InferenceResponse +import com.correx.core.inference.InferenceRouter +import com.correx.core.inference.PromptRenderer +import com.correx.core.inference.ResponseFormat +import com.correx.core.inference.Tokenizer +import com.correx.core.inference.ToolCallRequest +import com.correx.core.inference.ToolDefinition +import com.correx.core.inference.ModelCapability +import com.correx.core.sessions.ApprovalMode +import com.correx.core.inference.ToolFunction +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.risk.RiskAssessor +import com.correx.core.risk.RiskContext +import com.correx.core.risk.toApprovalTier +import com.correx.core.sessions.Session +import com.correx.core.tools.compression.ToolOutputContext +import com.correx.core.tools.contract.FileAffectingTool +import com.correx.core.tools.contract.Tool +import com.correx.core.tools.contract.ToolCapability +import com.correx.core.tools.contract.ToolExecutor +import com.correx.core.tools.contract.ToolResult +import com.correx.core.tools.registry.ToolRegistry +import com.correx.core.transitions.evaluation.EvaluationContext +import com.correx.core.transitions.evaluation.PromptResolver +import com.correx.core.transitions.execution.StageExecutionResult +import com.correx.core.transitions.graph.BuildExpectation +import com.correx.core.transitions.graph.StageConfig +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.core.transitions.resolution.TransitionDecision +import com.correx.core.transitions.resolution.TransitionResolver +import com.correx.core.validation.artifact.ArtifactExtractionPipeline +import com.correx.core.validation.artifact.ArtifactFailure +import com.correx.core.validation.model.ValidationContext +import com.correx.core.validation.model.ValidationReport +import com.correx.core.validation.model.ValidationSeverity +import com.correx.core.validation.pipeline.ValidationOutcome +import com.correx.core.validation.pipeline.ValidationPipeline +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.TimeoutCancellationException +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeout +import kotlinx.datetime.Clock +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.jsonObject +import com.correx.core.journal.DecisionJournalRenderer +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.kernel.orchestration.subagent.SubagentRunner +import org.slf4j.LoggerFactory +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.Paths +import java.util.* +import java.util.concurrent.* +import java.util.concurrent.atomic.* +import kotlin.coroutines.cancellation.CancellationException + +@Suppress("CyclomaticComplexMethod") +internal suspend fun SessionOrchestrator.executeStage( + sessionId: SessionId, + stageId: StageId, + graph: WorkflowGraph, + session: Session, + config: OrchestrationConfig, + effectives: RunEffectives, +): StageExecutionResult { + val stageConfig = requireNotNull(graph.stages[stageId]) { + "Stage '${stageId.value}' not declared in workflow graph" + } + val systemPrompt = stageConfig.metadata["systemPrompt"] + ?.let { path -> + runCatching { promptResolver.resolve(path) } + .onFailure { + log.error(error(stageId, path, it)) + } + .getOrNull() + } + ?.takeIf { it.isNotBlank() } + ?.let { text -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = text, + sourceType = "systemPrompt", + sourceId = stageId.value, + tokenEstimate = estimateTokens(text), + role = EntryRole.SYSTEM, + ), + ) + } ?: config.defaultSystemPromptPath + ?.let { path -> + runCatching { promptResolver.resolve(path) } + .onFailure { + log.error(error(stageId, path, it)) + } + .getOrNull() + } + ?.takeIf { it.isNotBlank() } + ?.let { text -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = text, + sourceType = "systemPrompt", + sourceId = stageId.value, + tokenEstimate = estimateTokens(text), + role = EntryRole.SYSTEM, + ), + ) + } ?: emptyList() + // A stage re-entered to repair a gate failure has its own (often generative/"scaffold") prompt + // SUPPRESSED: that mandate is what drove it to overwrite real files with stubs on re-entry. The + // recovery-ticket entry (buildRecoveryTicketEntry) is the sole mandate here — it already carries + // "make the smallest change, don't rewrite from scratch" plus the gate evidence. See + // mandateSuppressedByTicket. + val promptEntries = if (mandateSuppressedByTicket(sessionId, stageId, stageConfig)) { + emptyList() + } else { + stageConfig.metadata["promptInline"] + ?.takeIf { it.isNotBlank() } + ?.let { text -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L1, + content = text, + sourceType = "agentPrompt", + sourceId = stageId.value, + tokenEstimate = estimateTokens(text), + role = EntryRole.USER, + ), + ) + } + ?: stageConfig.metadata["prompt"] + ?.let { path -> + val resolvedText = runCatching { promptResolver.resolve(path) } + .onFailure { + log.error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "failed to load prompt '$path': ${it.message}", + ) + } + .getOrNull() + val text = resolvedText?.takeIf { it.isNotBlank() } + ?: error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "declared prompt '$path' could not be resolved", + ) + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L1, + content = text, + sourceType = "agentPrompt", + sourceId = stageId.value, + tokenEstimate = estimateTokens(text), + role = EntryRole.USER, + ), + ) + } + ?: emptyList() + } + + val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted } + require(llmEmittedSlots.size <= 1) { + "Stage '${stageId.value}' declares ${llmEmittedSlots.size} LLM-emitted artifact kinds; " + + "only 0 or 1 is supported" + } + val responseFormat = llmEmittedSlots.firstOrNull() + ?.let { ResponseFormat.Json(it.kind.deriveJsonSchema()) } + ?: ResponseFormat.Text + + 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 + // across rounds; instead we grow our own ordered list and let the builder restamp + // ordinals from it each round. + val journalState = decisionJournalRepository.getJournal(sessionId) + val summaryText = journalState.summaryArtifactId?.let { id -> + artifactStore.get(id)?.toString(Charsets.UTF_8) + ?: run { + log.warn("Journal summary artifact {} not found in store — rendering without summary", id.value) + null + } + } + val journalText = decisionJournalRenderer.render(journalState, summaryText) + val journalEntries = if (journalText.isBlank()) emptyList() else listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = journalText, + sourceType = "decisionJournal", + sourceId = "decision-journal", + tokenEstimate = estimateTokens(journalText), + role = EntryRole.SYSTEM, + ), + ) + val repoMapEntries = buildContextualRepoEntries(sessionId, stageId, stageConfig) + val sessionEvents = eventStore.read(sessionId) + val criticIds = criticArtifactIds(sessionEvents, graph, stageId) + val criticFrom = sessionEvents + .mapNotNull { it.payload as? RefinementIterationEvent } + .lastOrNull { it.cycleKey.endsWith("->${stageId.value}") } + ?.cycleKey?.substringBefore("->") + val needsEntries = buildNeedsArtifactEntries(sessionId, stageConfig, criticIds, criticFrom) + val profileEntries = session.state.boundProfile?.about + ?.takeIf { it.isNotBlank() } + ?.let { about -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = "## Operator profile\n$about", + sourceType = "operatorProfile", + sourceId = "operator-profile", + tokenEstimate = about.length / 4, + role = EntryRole.SYSTEM, + ), + ) + } ?: emptyList() + val projectProfileEntries = session.state.boundProjectProfile + ?.let { listOf(buildProjectProfileEntry(it)) } ?: emptyList() + // AGENTS.md injection is disabled for now. The workspace-root AGENTS.md (the DOX framework) + // carries a mandatory "Read Before Editing" walk — "read the whole AGENTS.md chain, do not + // rely on memory, re-read before editing" — which drives a small model into an endless + // read-loop that never writes (2026-07-05 root cause). Reincorporating DOX properly (scoped + // to the stage's subproject, not the repo root; not read-mandating) is deferred. + val agentInstructionsEntries = emptyList() + val retryFeedbackEntries = buildRetryFeedbackEntry(sessionEvents, stageId) + ?.let { listOf(it) } ?: emptyList() + val recoveryTicketEntries = buildRecoveryTicketEntry(sessionEvents, stageId) + ?.let { listOf(it) } ?: emptyList() + val vocabularyEntries = artifactKindRegistry + ?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" } + ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList() + // Deterministic claim: a stage flagged claimTask gets the kernel to claim the next ready task + // (or re-surface the one already claimed) and inject its context bundle as L0 — the loop + // advances by recorded fact, not by the agent remembering to claim. + val claimedTaskEntries = if (stageConfig.metadata["claimTask"] == "true") { + taskClaimCoordinator?.claimNext(sessionId)?.let { bundle -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L0, + content = bundle, + sourceType = "claimedTask", + sourceId = stageId.value, + tokenEstimate = estimateTokens(bundle), + role = EntryRole.SYSTEM, + ), + ) + } ?: emptyList() + } else { + emptyList() + } + // Remaining-delta checklist (stage-termination design 2026-07-11 §1a): the stage's currently + // unmet file-contract assertions as a forward-looking, shrinking TODO. Injected pinned + // (neverDrop) so it survives the budget/dedup passes that otherwise wipe the model's memory of + // progress; refreshed inside the tool loop each time a write lands (§1b, cache-until-write). + var remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives) + val remainingDeltaEntries = remainingDeltaResults + ?.let { buildRemainingDeltaEntry(contractFailureItems(it)) } + ?.let { listOf(it) } ?: emptyList() + var accumulatedEntries = stampBuckets( + systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries + + journalEntries + repoMapEntries + claimedTaskEntries + + needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + + clarificationEntries + retryFeedbackEntries + recoveryTicketEntries + remainingDeltaEntries, + ) + val contextPack = runCatching { + contextPackBuilder.build( + id = ContextPackId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = stageId, + entries = accumulatedEntries, + budget = TokenBudget(limit = stageConfig.tokenBudget), + ) + }.getOrElse { e -> + if (e is CancellationException) throw e + if (e !is RequiredContextOverflowException) throw e + // Required context doesn't fit — retrying with the same inputs can't succeed. + log.error("[Orchestrator] stage={}: {}", stageId.value, e.message) + return StageExecutionResult.Failure(e.message ?: "required context overflow", retryable = false) + } + emitContextTruncationIfNeeded(sessionId, stageId, contextPack) + + var currentContext = contextPack + var inferenceResult = runInference( + sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, + ) + var toolRounds = 0 + var consecutiveReadOnlyRounds = 0 + var consecutiveRejectedRounds = 0 + // Fingerprints ("tool:arguments") of every read-only call made during the current + // no-write streak. A round only counts against the read-loop breaker if it repeats calls + // already seen — a stage reading N distinct files before writing is legitimate context + // gathering, not a loop, and shouldn't trip the same counter as re-reading the same file. + val seenReadFingerprints = mutableSetOf() + // Set when the model produces its artifact via the emit_artifact tool instead of a final + // JSON message; overrides the post-loop capture of the (then-empty) assistant text. + var llmArtifactOverride: String? = null + + val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" } + fun owesFileWrite(): Boolean = fileWrittenSlots.isNotEmpty() && + fileWrittenSlots.any { artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() } + + // Gate: if the stage declares require_task_decompose, block stage_complete until at least one + // task_decompose (or task_create) call has completed successfully in this stage. Checked via + // ToolInvocationRequestedEvent + ToolExecutionCompletedEvent in the session event log. + // ponytail: scans events on every stage_complete attempt; negligible cost (in-memory list, rare call). + fun owesTaskDecompose(): Boolean { + if (stageConfig.metadata["requireTaskDecompose"] != "true") return false + val events = eventStore.read(sessionId) + val requestedIds = events + .mapNotNull { it.payload as? ToolInvocationRequestedEvent } + .filter { it.stageId == stageId && it.toolName in setOf("task_decompose", "task_create") } + .map { it.invocationId } + .toSet() + if (requestedIds.isEmpty()) return true + val completedIds = events + .mapNotNull { it.payload as? ToolExecutionCompletedEvent } + .filter { it.toolName in setOf("task_decompose", "task_create") && it.invocationId in requestedIds } + .map { it.invocationId } + .toSet() + return completedIds.isEmpty() + } + + // Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS). + // Returns the new result rather than mutating inferenceResult, to preserve smart casts. + suspend fun pushBack(nudge: String, forceWriteOnly: Boolean = false): InferenceResult { + accumulatedEntries = accumulatedEntries + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L2, + sourceType = "toolResult", + sourceId = UUID.randomUUID().toString(), + content = nudge, + tokenEstimate = estimateTokens(nudge), + role = EntryRole.TOOL, + ) + currentContext = contextPackBuilder.build( + id = ContextPackId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = stageId, + entries = accumulatedEntries, + budget = TokenBudget(limit = stageConfig.tokenBudget), + ) + emitContextTruncationIfNeeded(sessionId, stageId, currentContext) + toolRounds++ + return runInference( + sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, + forceWriteOnly = forceWriteOnly, + ) + } + + val writeNudge = "ERROR: you described the change but did not apply it. You MUST call the " + + "file_edit or file_write tool to write the change to the file before finishing — do " + + "not output the file contents as a message." + + val readLoopNudge = "STOP reading. You have read enough files and this stage has produced " + + "nothing yet. You MUST now call the file_write tool to create the required file(s) for " + + "this stage. Do not call file_read or list_dir again before you have written a file." + + while ( + inferenceResult is InferenceResult.Success && + toolRounds < tuning.maxToolRounds && + (inferenceResult.response.finishReason is FinishReason.ToolCall || owesFileWrite()) + ) { + // Content turn (no tool call) but the stage still owes a file_written artifact: the + // model emitted the change as prose instead of invoking the write tool. Nudge it to + // actually write (F-018). + if (inferenceResult.response.finishReason !is FinishReason.ToolCall) { + inferenceResult = pushBack(writeNudge) + continue + } + // emit_artifact is a meta-tool: the LLM calls it with the artifact's fields as arguments. + // Capture those arguments as the artifact content and exit the loop straight to validation + // (the post-loop capture honours this override instead of the empty assistant text). + val emitCall = inferenceResult.response.toolCalls.firstOrNull { it.function.name == EMIT_ARTIFACT_TOOL } + if (emitCall != null && llmEmittedSlots.isNotEmpty()) { + llmArtifactOverride = emitCall.function.arguments + break + } + // stage_complete is a meta-tool: the LLM calls it to signal the stage's goal is met. + // Skip tool dispatch — the loop exits and normal validation/transition follows + // (emitToolArtifacts + verifyProduces run on the success path after the loop). + if (inferenceResult.response.toolCalls.any { it.function.name == STAGE_COMPLETE_TOOL }) { + // Premature-completion guard: don't accept stage_complete while the stage still + // owes an artifact. For an LLM-emitted slot, nudge to emit the JSON (F-010); for a + // file_written slot, nudge to invoke the write tool (F-018). Otherwise complete. + val owedLlm = llmEmittedSlots.firstOrNull()?.takeIf { + inferenceResult.response.text.isBlank() && + artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() + } + when { + owedLlm != null -> inferenceResult = pushBack( + "ERROR: stage_complete is not allowed yet — you have not emitted the " + + "required '${owedLlm.name.value}' artifact. Call the emit_artifact tool with " + + "the fields filled in (or respond with the artifact as a single JSON message " + + "matching the provided schema).", + ) + owesFileWrite() -> inferenceResult = pushBack(writeNudge) + owesTaskDecompose() -> inferenceResult = pushBack( + "ERROR: stage_complete is not allowed yet — this stage requires a successful " + + "task_decompose (or task_create) call before it can complete. Call task_decompose " + + "with a parent epic and DEPENDS_ON-linked children representing the full task graph. " + + "Do not call stage_complete until task_decompose has returned successfully.", + ) + else -> break + } + continue + } + val toolEntries = dispatchToolCalls(sessionId, stageId, + inferenceResult.response.toolCalls, stageConfig, effectives, + approvalModeFor(session.state.boundProfile?.approvalMode), + inferenceResult.response.reasoning) + val fatalEntry = toolEntries.firstOrNull { it.content.startsWith("FATAL:") } + if (fatalEntry != null) { + emitProcessResultEvents(sessionId, stageId, stageConfig) + return StageExecutionResult.Failure( + fatalEntry.content.removePrefix("FATAL: "), + retryable = false, + ) + } + // Recoverable (ERROR:) tool failures are NOT terminal — they flow back into the + // loop as tool-result context so the model can see the error and adapt (bounded by + // MAX_TOOL_ROUNDS). Only FATAL: failures (handled above) abort the stage. + accumulatedEntries = accumulatedEntries + toolEntries + // Read-loop breaker: this round called only read-only tools yet the stage still owes a + // file_written artifact. Left alone the model keeps reading until MAX_TOOL_ROUNDS and + // never writes (F-018 nudges only cover a prose turn or a premature stage_complete, not + // an endless read-loop). After READ_LOOP_NUDGE_THRESHOLD such rounds, force the write + // nudge; a real write resets the counter so multi-file stages are unaffected. + if (owesFileWrite() && + inferenceResult.response.toolCalls.none { it.function.name in WRITE_TOOL_NAMES } + ) { + val roundFingerprints = inferenceResult.response.toolCalls + .map { "${it.function.name}:${it.function.arguments}" } + val madeProgress = roundFingerprints.any { it !in seenReadFingerprints } + seenReadFingerprints += roundFingerprints + if (madeProgress) { + consecutiveReadOnlyRounds = 0 + } else { + consecutiveReadOnlyRounds++ + if (consecutiveReadOnlyRounds >= tuning.readLoopNudgeThreshold) { + consecutiveReadOnlyRounds = 0 + inferenceResult = pushBack(readLoopNudge, forceWriteOnly = true) + continue + } + } + } else { + consecutiveReadOnlyRounds = 0 + seenReadFingerprints.clear() + } + // Rejection-loop breaker (stage-type agnostic): every tool result this round was a + // rejection (plane-2 BLOCKED or a stage/policy ERROR) and nothing succeeded. A stage that + // owes no file_written artifact (discovery, analyst) never trips the read-loop breaker, so + // without this it re-issues the same rejected call until MAX_TOOL_ROUNDS. After the + // threshold, tell it to stop retrying and produce its output; any success resets. + val toolResults = toolEntries.filter { it.role == EntryRole.TOOL } + val allRejected = toolResults.isNotEmpty() && + toolResults.all { it.content.startsWith("BLOCKED:") || it.content.startsWith("ERROR:") } + if (allRejected) { + consecutiveRejectedRounds++ + if (consecutiveRejectedRounds >= tuning.rejectionLoopNudgeThreshold) { + consecutiveRejectedRounds = 0 + inferenceResult = pushBack( + "STOP. Your last tool calls were all rejected and retrying the same paths will " + + "keep failing. Do not repeat them. Proceed with the information you already " + + "have: produce this stage's required output now (call stage_complete, or emit " + + "the required artifact). If a path you wanted does not exist yet, that is a " + + "fact to record — do not keep probing for it.", + ) + continue + } + } else { + consecutiveRejectedRounds = 0 + } + // Refresh the remaining-delta checklist only when a write actually landed this round + // (cache-until-write, §1b): reads can never change the contract verdict, so re-evaluating + // on them would just burn filesystem work and re-emit identical events. On a successful + // write, recompute contract-minus-reality and swap the pinned entry so the model watches + // the list shrink toward the empty=done signal it previously lacked. + val wroteThisRound = inferenceResult.response.toolCalls.any { it.function.name in WRITE_TOOL_NAMES } && + toolEntries.any { + it.role == EntryRole.TOOL && + !it.content.startsWith("ERROR:") && !it.content.startsWith("BLOCKED:") + } + if (wroteThisRound) { + remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives) + val refreshed = remainingDeltaResults?.let { buildRemainingDeltaEntry(contractFailureItems(it)) } + accumulatedEntries = accumulatedEntries.filterNot { it.sourceType == "remainingDelta" } + + listOfNotNull(refreshed) + } + currentContext = contextPackBuilder.build( + id = ContextPackId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = stageId, + entries = accumulatedEntries, + budget = TokenBudget(limit = stageConfig.tokenBudget), + ) + emitContextTruncationIfNeeded(sessionId, stageId, currentContext) + inferenceResult = runInference( + sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives, + ) + toolRounds++ + } + // Final clean emission. Producing the artifact while tools are in the request is unreliable on + // some local chat templates (Gemma leaks `<|channel>` markers into the text; other models keep + // tool-calling until the round cap and never emit). For an LLM-emitted stage that hasn't already + // produced its artifact via the emit_artifact tool, fire ONE tools-less inference — verified to + // yield clean JSON deterministically. Skipped when the model already emitted clean final text. + if (inferenceResult is InferenceResult.Success && llmEmittedSlots.isNotEmpty() && llmArtifactOverride == null) { + val last = inferenceResult.response + val needsCleanEmission = last.finishReason is FinishReason.ToolCall || + last.text.isBlank() || last.text.contains("<|channel") + if (needsCleanEmission && !isCancelled(sessionId)) { + val nudge = "Stop calling tools. Output the required '${llmEmittedSlots.first().name.value}' " + + "artifact now as a single JSON object matching the schema — no tool calls, no commentary." + accumulatedEntries = accumulatedEntries + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L2, + sourceType = "toolResult", + sourceId = UUID.randomUUID().toString(), + content = nudge, + tokenEstimate = estimateTokens(nudge), + role = EntryRole.TOOL, + ) + currentContext = contextPackBuilder.build( + id = ContextPackId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = stageId, + entries = accumulatedEntries, + budget = TokenBudget(limit = stageConfig.tokenBudget), + ) + inferenceResult = runInference( + sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, + responseFormat, effectives, withTools = false, + ) + } + } + return when (inferenceResult) { + is InferenceResult.Cancelled -> StageExecutionResult.Failure("CANCELLED", retryable = false) + is InferenceResult.Failed -> + StageExecutionResult.Failure(inferenceResult.reason, retryable = isRetryableFailure(inferenceResult.reason)) + is InferenceResult.Success -> { + if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false) + // Capture the LLM-emitted artifact's JSON so transition conditions + // (e.g. ArtifactFieldEquals on a reviewer's verdict) can read it. An emit_artifact + // tool call (llmArtifactOverride) supplies the content directly; otherwise the + // artifact is the final assistant message text. + val rawArtifactText = llmArtifactOverride ?: inferenceResult.response.text + val slot = llmEmittedSlots.firstOrNull() + // Artifact-emission ladder: deterministic repair (prose/fences/trailing commas), then + // — rarely, gated by the failure classifier — one grammar-constrained LLM-repair rung, + // before the text is cached/validated. On a hard failure the policy decision + // short-circuits the stage. Only the nondeterministic rung records events (spec 2026-07-04 §6). + val artifactText = if (slot != null) { + when (val res = artifactExtractionPipeline.run(rawArtifactText, slot.kind.deriveJsonSchema())) { + is ArtifactExtractionPipeline.ExtractionResult.Resolved -> { + if (res.repaired) { + emitArtifactRepairAttempted(sessionId, stageId, slot, ArtifactFailure.FORMATTING, "DETERMINISTIC") + emitArtifactRepairResolved(sessionId, stageId, slot, res.canonicalJson.toString()) + } + res.canonicalJson.toString() + } + is ArtifactExtractionPipeline.ExtractionResult.Unresolved -> + when (val ladder = repairArtifact(sessionId, stageId, slot, res, stageConfig, effectives, config.stageTimeoutMs)) { + is ArtifactLadderOutcome.Text -> ladder.text + is ArtifactLadderOutcome.Reject -> return ladder.failure + } + } + } else { + rawArtifactText + } + val artifactHash = if (artifactText != rawArtifactText || llmArtifactOverride != null) { + artifactStore.put(artifactText.toByteArray()) + } else { + inferenceResult.responseArtifactId + } + slot?.let { slot -> + artifactContentCache["${sessionId.value}:${slot.name.value}"] = artifactText + artifactHash?.let { hash -> + emit( + sessionId, + ArtifactContentStoredEvent( + artifactId = slot.name, + contentHash = hash, + sessionId = sessionId, + stageId = stageId, + ), + ) + } + } + val validationCtx = ValidationContext( + graph = graph, + sessionState = session.state, + availableTools = effectives.registry?.all()?.map { it.name }?.toSet(), + producedArtifactContent = graph.stages.values.flatMap { it.produces }.mapNotNull { slot -> + artifactContentCache["${sessionId.value}:${slot.name.value}"]?.let { slot.name.value to it } + }.toMap(), + ) + when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) { + is StageExecutionResult.Success -> { + emitToolArtifacts(sessionId, stageId, stageConfig) + emitLlmArtifacts(sessionId, stageId, stageConfig) + runPostStageGates( + sessionId, stageId, stageConfig, effectives, + session.state.boundProjectProfile?.commands ?: emptyMap(), + ) + } + + is StageExecutionResult.Failure -> outcome + } + } + } +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorTypes.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorTypes.kt new file mode 100644 index 00000000..7798304c --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorTypes.kt @@ -0,0 +1,22 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.toolintent.WorkspacePolicy +import com.correx.core.tools.contract.ToolExecutor +import com.correx.core.tools.registry.ToolRegistry +import com.correx.core.transitions.execution.StageExecutionResult + +/** Per-run effective tool wiring (workspace-scoped registry/executor/policy or the orchestrator defaults). */ +internal data class RunEffectives( + val registry: ToolRegistry?, + val executor: ToolExecutor?, + val policy: WorkspacePolicy?, +) + +/** Result of the artifact extraction/repair ladder: usable text, or a hard stage failure. */ +internal sealed interface ArtifactLadderOutcome { + data class Text(val text: String) : ArtifactLadderOutcome + data class Reject(val failure: StageExecutionResult.Failure) : ArtifactLadderOutcome +} + +/** A tool result rendered for the model, with the CAS hash of the full (pre-truncation) output if spilled. */ +internal data class RenderedToolResult(val content: String, val fullOutputHash: String?)