From 1f794dad63801924b9f781947b4cfa5717b648f6 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 13 Jul 2026 12:06:20 +0400 Subject: [PATCH] refactor(kernel): decompose DefaultSessionOrchestrator into per-concern extensions Moves the 11 private step/recovery helpers off the concrete orchestrator into internal extension-fun files (Step: step, executeMove, enterStage, decideGateExhaustion; Recovery: retry/route/ticket helpers). Class keeps the override seams (run, cancel, submitApprovalDecision), the cross-module public API (rehydrate, resume, submitClarification, submitSteering), enrich, and validatedArtifactContent. File-private helpers the moved funs need are promoted to internal. Pure relocation; clears TooManyFunctions on the class. kernel + testing:kernel green; server/cli compile clean. --- .../DefaultSessionOrchestrator.kt | 559 +----------------- .../DefaultSessionOrchestratorRecovery.kt | 238 ++++++++ .../DefaultSessionOrchestratorStep.kt | 324 ++++++++++ 3 files changed, 575 insertions(+), 546 deletions(-) create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorRecovery.kt create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorStep.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 8224b548..60be6ecc 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 @@ -5,23 +5,13 @@ import com.correx.core.approvals.ApprovalStatus import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.inference.Tokenizer import com.correx.core.journal.DefaultDecisionJournalRepository -import com.correx.core.journal.DecisionJournalRenderer -import com.correx.core.artifacts.ArtifactState import com.correx.core.artifacts.kind.ArtifactKindRegistry import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.ApprovalDecisionResolvedEvent -import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.ArtifactContentStoredEvent -import com.correx.core.events.events.FileWrittenEvent -import com.correx.core.events.events.ToolInvocationRequestedEvent import com.correx.core.events.events.ClarificationAnswer import com.correx.core.events.events.ClarificationAnsweredEvent -import com.correx.core.events.events.ArtifactValidatedEvent -import com.correx.core.events.events.FailureTicketOpenedEvent import com.correx.core.events.events.OrchestrationResumedEvent -import com.correx.core.events.events.RefinementIterationEvent -import com.correx.core.events.events.RetrySalvageDecidedEvent -import com.correx.core.events.events.SalvageDecision import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.TransitionExecutedEvent @@ -30,21 +20,15 @@ import com.correx.core.events.types.ApprovalDecisionId import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.ClarificationRequestId -import com.correx.core.events.types.ArtifactLifecyclePhase import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.events.types.TransitionId import com.correx.core.kernel.execution.WorkflowResult import com.correx.core.kernel.orchestration.subagent.InSessionSubagentRunner -import com.correx.core.kernel.orchestration.subagent.SubagentRunRequest import com.correx.core.kernel.orchestration.subagent.SubagentRunner -import com.correx.core.kernel.retry.FailureFingerprint import com.correx.core.kernel.retry.RetryCoordinator -import com.correx.core.kernel.retry.RetryDecision import com.correx.core.sessions.Session -import com.correx.core.transitions.execution.StageExecutionResult import com.correx.core.transitions.graph.WorkflowGraph -import com.correx.core.transitions.resolution.TransitionDecision import kotlinx.datetime.Clock import org.slf4j.LoggerFactory import java.util.UUID @@ -70,37 +54,37 @@ private val log = LoggerFactory.getLogger(DefaultSessionOrchestrator::class.java // repair owner (TICKET_ROUTE); the owner, once done, hands control straight back to the gate that // opened the ticket (TICKET_RETURN). Both are kernel-driven, not plan edges — see ticketReturnMove. internal val TICKET_ROUTE = TransitionId("ticket-route") -private val TICKET_RETURN = TransitionId("ticket-return") +internal val TICKET_RETURN = TransitionId("ticket-return") // Shortest evidence token treated as a file reference — guards against matching a written path's // suffix on a trivially short token (an extension fragment, a one-char name). -private const val MIN_EVIDENCE_TOKEN = 5 +internal const val MIN_EVIDENCE_TOKEN = 5 // File-like tokens in gate evidence (tsc/build output): "src/App.tsx", "frontend/x.ts:12:5". -private val EVIDENCE_PATH_RE = Regex("""[\w./@-]+\.[A-Za-z0-9]+""") +internal val EVIDENCE_PATH_RE = Regex("""[\w./@-]+\.[A-Za-z0-9]+""") // Deterministic gate id → capability the failing stage must hold to change the failure condition. // A gate not listed here is not eligible for recovery routing (retries in place as before). -private val GATE_REQUIRED_CAPABILITY: Map = mapOf( +internal val GATE_REQUIRED_CAPABILITY: Map = mapOf( "execution" to "file_write", "contract" to "file_write", "static_analysis" to "file_write", ) // Deterministic failure category from the gate id (no LLM — keeps routing off the untrusted path). -private fun ticketCategory(gate: String): String = when (gate) { +internal fun ticketCategory(gate: String): String = when (gate) { "plan_compile" -> "planning" else -> "implementation" } class DefaultSessionOrchestrator( - private val repositories: OrchestratorRepositories, + internal val repositories: OrchestratorRepositories, engines: OrchestratorEngines, - private val retryCoordinator: RetryCoordinator, + internal val retryCoordinator: RetryCoordinator, artifactStore: ArtifactStore, tokenizer: Tokenizer? = null, decisionJournalRepository: DefaultDecisionJournalRepository, - private val compactionService: JournalCompactionService? = null, + internal val compactionService: JournalCompactionService? = null, artifactKindRegistry: ArtifactKindRegistry? = null, repoKnowledgeRetriever: RepoKnowledgeRetriever? = null, readyTaskCounter: ReadyTaskCounter? = null, @@ -114,7 +98,7 @@ class DefaultSessionOrchestrator( // Hybrid-exhaustion salvage judge (T6/T7): consulted only when the "review" gate exhausts its // per-gate retry budget. Null = deterministic allow-one-reset-then-fail fallback (see // decideGateExhaustion) so the feature degrades safely without inference. - private val salvageJudge: SalvageJudge? = engines.salvageJudge + internal val salvageJudge: SalvageJudge? = engines.salvageJudge override val subagentRunner: SubagentRunner = InSessionSubagentRunner( executeStage = { sid, stg, graph, session, cfg -> @@ -140,336 +124,6 @@ class DefaultSessionOrchestrator( } } - @Suppress("LongMethod") - private tailrec suspend fun step(ctx: EnrichedExecutionContext): WorkflowResult { - log.debug( - "[Orchestrator] step session={} stage={} stageCount={}", - ctx.sessionId.value, ctx.currentStageId.value, ctx.stageCount, - ) - if (isCancelled(ctx.sessionId)) return handleCancellation(ctx.sessionId, ctx.currentStageId) - - val enriched = EnrichedExecutionContext( - graph = ctx.graph, - sessionId = ctx.sessionId, - stageCount = ctx.stageCount, - currentStageId = ctx.currentStageId, - config = ctx.config, - state = orchestrationRepository.getState(ctx.sessionId), - session = repositories.sessionRepository.getSession(ctx.sessionId), - ) - - val stageConfig = enriched.graph.stages[enriched.currentStageId] - val targetIds: Set = stageConfig?.produces?.map { it.name }?.toSet() ?: emptySet() - val stageArtifacts: Map = if (targetIds.isEmpty()) { - emptyMap() - } else { - val validated = repositories.eventStore.read(enriched.sessionId) - .mapNotNull { (it.payload as? ArtifactValidatedEvent)?.artifactId } - .filter { it in targetIds } - .toSet() - validated.associateWith { ArtifactState(phase = ArtifactLifecyclePhase.VALIDATED) } - } - - val artifactContent: Map = targetIds.mapNotNull { id -> - val cacheKey = "${enriched.sessionId.value}:${id.value}" - artifactContentCache[cacheKey]?.let { content -> id to content } - }.toMap() - - val resolved = resolveTransition( - enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent, - ) - // Ticket return-to-sender: a repair stage (the file's author, or the fallback recovery stage) - // is entered by kernel routing from whichever stage failed a gate, so its exit must go BACK to - // that exact gate (re-run it) — not follow a plan edge that would walk a regenerating - // implementer or skip intervening stages. Derived from the latest failure ticket routed here - // (replay-deterministic). Overrides the resolved edge; an operator preempt below still wins. - val baseDecision = ticketReturnMove(enriched) ?: resolved - // Preemptive redirect override (freestyle graph re-routing): if the operator confirmed a - // jump (a recorded, unconsumed PreemptRedirectEvent), it overrides the resolver's edge — - // unless the target is unknown or its needs aren't satisfied, in which case it is blocked - // and the normal edge is taken. The decision is pure over the event log (replay-deterministic); - // only the block-event emission is a side effect here. - val decision = when ( - val outcome = PreemptRedirect.decide( - events = repositories.eventStore.read(enriched.sessionId), - graph = enriched.graph, - sessionId = enriched.sessionId, - artifactAvailable = { id -> - !artifactContentCache["${enriched.sessionId.value}:${id.value}"].isNullOrBlank() - }, - nowMs = Clock.System.now().toEpochMilliseconds(), - ) - ) { - is PreemptRedirect.Outcome.Override -> outcome.move - is PreemptRedirect.Outcome.Block -> { - emit(enriched.sessionId, outcome.event) - log.info( - "[Orchestrator] redirect blocked session={} to={} reason={}", - enriched.sessionId.value, outcome.event.toStageId.value, outcome.event.reason, - ) - baseDecision - } - PreemptRedirect.Outcome.None -> baseDecision - } - log.debug( - "[Orchestrator] transition session={} stage={} decision={}", - enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName, - ) - return when (decision) { - is TransitionDecision.Move -> when (val r = executeMove(enriched, decision)) { - is StepResult.Continue -> step(r.ctx) - is StepResult.Terminal -> r.result - } - - is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) { - completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount, enriched.graph.id) - } else { - // No outgoing edge matched: the stage finished but produced nothing that satisfies a - // transition (e.g. the analyst never emitted a validated `analysis`). Treat it as a - // retryable stage failure — a fresh attempt lets the model gather more before writing — - // rather than killing the run. Only terminal once the retry budget is spent. - retryStageOrFail( - enriched, - "no transition condition matched from stage ${enriched.currentStageId.value}", - ) - } - - is TransitionDecision.Blocked -> failWorkflow( - sessionId = enriched.sessionId, - stageId = enriched.currentStageId, - reason = decision.reason, - retryExhausted = false, - ) - - is TransitionDecision.NoMatch -> retryStageOrFail( - enriched, - "no matching transition from stage ${enriched.currentStageId.value}", - ) - } - } - - /** - * A non-terminal stage whose outcome matched no outgoing transition (Stay/NoMatch) produced no - * usable artifact. Route it back through the stage's retry budget instead of failing the run: - * re-execute the same stage if attempts remain, else fail terminally with retryExhausted=true. - */ - private suspend fun retryStageOrFail( - ctx: EnrichedExecutionContext, - reason: String, - ): WorkflowResult { - val attempt = orchestrationRepository.getState(ctx.sessionId).retryCount - val shouldRetry = retryCoordinator.shouldRetry( - sessionId = ctx.sessionId, - stageId = ctx.currentStageId, - currentAttempt = attempt, - policy = ctx.config.retryPolicy, - failureReason = reason, - ) - if (!shouldRetry) { - return failWorkflow(ctx.sessionId, ctx.currentStageId, reason, retryExhausted = true) - } - return when (val r = enterStage(ctx, ctx.currentStageId)) { - is StepResult.Continue -> step(r.ctx) - is StepResult.Terminal -> r.result - } - } - - - /** - * Retry-agency guard (ticket + recovery model). When [failure]'s gate requires a capability the - * failing stage does not hold (e.g. a build/contract gate needs `file_write` on a write-less - * verifier), retrying the stage in place can never change the outcome. Instead open a - * [FailureTicketOpenedEvent] and route to the graph's recovery stage — which does hold the - * capability — bounded by the stage's own [RECOVERY_ROUTE_BUDGET]. - * - * Returns a [StepResult] when it took over the failure (routed, or budget-exhausted terminal), - * or null to fall through to the normal per-gate retry path. Null in every backward-compatible - * case: gate not capability-gated, stage already has the capability, or no recovery stage exists. - */ - private suspend fun maybeRouteToRecovery( - ctx: EnrichedExecutionContext, - stageId: StageId, - failure: StageExecutionResult.Failure, - state: OrchestrationState, - ): StepResult? { - val requiredCapability = GATE_REQUIRED_CAPABILITY[failure.gate] ?: return null - val stageTools = ctx.graph.stages[stageId]?.allowedTools ?: emptySet() - if (requiredCapability in stageTools) return null // stage has agency — retry in place is valid - return routeToRecovery(ctx, stageId, failure.gate, requiredCapability, failure.reason, state) - } - - /** - * Open a [FailureTicketOpenedEvent] and route [stageId]'s failure down a two-tier repair ladder. - * - * **Tier 1 (owner):** when the gate's evidence names a file a stage actually wrote, the ticket goes - * to that author ([resolveTicketOwner]) — it owns the code and the contract it declared, so it - * patches its own layer. Bounded by [RECOVERY_ROUTE_BUDGET]. - * - * **Tier 2 (arbiter / intent-holder):** when the owner loop spends its budget without settling the - * failure, the failure is a cross-file CONTRACT dispute no single owner can resolve (each side is - * internally consistent; they disagree with each other). The ticket escalates to the synthesized - * recovery stage ([findRecoveryStage]) recast as the intent-holder — given the initial intent as - * authority (see buildRecoveryTicketEntry) and told to reconcile ALL the named files in one pass. - * Bounded by its own independent [INTENT_ROUTE_BUDGET]. An orphan failure (no author) starts here. - * - * Both tiers charge progress-awarely: a route whose failure fingerprint changed from the previous - * one moved the needle, so it is FREE; only a route reproducing the same failure charges the tier's - * budget. Terminal only when every reachable tier is spent. Returns null when neither an owner nor a - * recovery stage exists (caller falls back to the legacy per-gate retry path). Shared by the agency - * guard ([maybeRouteToRecovery]) and the review-gate salvage judge ([decideGateExhaustion]). - */ - private suspend fun routeToRecovery( - ctx: EnrichedExecutionContext, - stageId: StageId, - gate: String, - requiredCapability: String, - reason: String, - state: OrchestrationState, - ): StepResult? { - val events = repositories.eventStore.read(ctx.sessionId) - val owner = resolveTicketOwner(events, reason) - ?.takeIf { it != stageId && requiredCapability in (ctx.graph.stages[it]?.allowedTools ?: emptySet()) } - val arbiter = findRecoveryStage(ctx.graph, stageId) - val fingerprint = FailureFingerprint.of(reason) - val ownerKey = stageId.value - val intentKey = stageId.value + INTENT_BUDGET_SUFFIX - - // (target, budgetKey, budget, escalated). Prefer the owner tier while it has budget; escalate to - // the arbiter tier once the owner loop is spent; null = no tier available now. - val route = when { - owner != null && !budgetExhausted(state, ownerKey, fingerprint, tuning.recoveryRouteBudget) -> - RouteTier(owner, ownerKey, tuning.recoveryRouteBudget, escalated = false) - arbiter != null && !budgetExhausted(state, intentKey, fingerprint, tuning.intentRouteBudget) -> - RouteTier(arbiter, intentKey, tuning.intentRouteBudget, escalated = true) - else -> null - } - if (route == null) { - if (owner == null && arbiter == null) return null // nothing to route to: legacy retry path - return StepResult.Terminal( - failWorkflow( - ctx.sessionId, - stageId, - "repair ladder exhausted for stage ${stageId.value} (gate=$gate): $reason", - retryExhausted = true, - ), - ) - } - - val prev = state.recoveryFailureFingerprints[route.budgetKey] - val progressed = prev != null && prev != fingerprint - val used = state.recoveryRoutes[route.budgetKey] ?: 0 - val routeAttempt = if (progressed) used else used + 1 - emit( - ctx.sessionId, - FailureTicketOpenedEvent( - sessionId = ctx.sessionId, - stageId = stageId, - gate = gate, - category = ticketCategory(gate), - requiredCapability = requiredCapability, - routeTo = route.target, - evidence = reason, - routeAttempt = routeAttempt, - fingerprint = fingerprint, - escalated = route.escalated, - ), - ) - log.info( - "[Orchestrator] failure ticket session={} stage={} gate={} lacks={} -> {}={} " + - "(charged={}/{}, progressed={})", - ctx.sessionId.value, stageId.value, gate, requiredCapability, - if (route.escalated) "arbiter" else "owner", route.target.value, - routeAttempt, route.budget, progressed, - ) - val advancedTo = advanceStage( - ctx.sessionId, - stageId, - TransitionDecision.Move(transitionId = TICKET_ROUTE, to = route.target), - ) - return enterStage(ctx.copy(currentStageId = advancedTo), route.target) - } - - /** A chosen rung of the repair ladder: where to route, which budget it charges, and its tier. */ - private data class RouteTier( - val target: StageId, - val budgetKey: String, - val budget: Int, - val escalated: Boolean, - ) - - /** - * Progress-aware budget check for one ladder tier. A route whose failure fingerprint changed from - * the previous route on the same [key] made progress (fixed one cause, surfaced another) → NOT - * exhausted (progress is free). Only a no-progress streak that reaches [budget] is exhausted. - */ - private fun budgetExhausted( - state: OrchestrationState, - key: String, - fingerprint: String, - budget: Int, - ): Boolean { - val prev = state.recoveryFailureFingerprints[key] - val progressed = prev != null && prev != fingerprint - val used = state.recoveryRoutes[key] ?: 0 - return !progressed && used >= budget - } - - /** - * The graph's recovery stage: a stage marked `metadata["role"] == "recovery"` (or, as a - * convenience, one whose id is literally "recovery"), other than [failingStageId] itself. - * Null when the graph declares none — recovery routing is opt-in per workflow. - */ - private fun findRecoveryStage(graph: WorkflowGraph, failingStageId: StageId): StageId? = - graph.stages.entries.firstOrNull { (id, cfg) -> - id != failingStageId && (cfg.metadata["role"] == "recovery" || id.value == "recovery") - }?.key - - /** - * Route-to-owner resolution: map the failing gate's [evidence] (build/tsc output that names - * files, e.g. "src/App.tsx: TS2322") to the stage that most recently WROTE one of those files, - * via the recorded manifest chain (ToolInvocationRequestedEvent.stageId ← FileWrittenEvent by - * invocationId → path). The last write of a named file wins — its author is who to hand the - * ticket to. Pure over recorded events (replay-deterministic). Null when the evidence names no - * written file (an orphan failure — caller falls back to the synthesized recovery stage). - */ - private fun resolveTicketOwner(events: List, evidence: String): StageId? { - val tokens = EVIDENCE_PATH_RE.findAll(evidence) - .map { it.value.substringBefore(':').replace('\\', '/') } - .filter { it.length >= MIN_EVIDENCE_TOKEN } - .toSet() - if (tokens.isEmpty()) return null - val invToStage = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent } - .associate { it.invocationId to it.stageId } - return events.mapNotNull { it.payload as? FileWrittenEvent } - .filter { it.postImageHash != null } - .lastOrNull { fw -> - val norm = fw.path.replace('\\', '/') - invToStage.containsKey(fw.invocationId) && tokens.any { norm == it || norm.endsWith("/$it") } - } - ?.let { invToStage[it.invocationId] } - } - - /** - * Return-to-sender for a routed repair stage. When the current stage was entered by a failure - * ticket ([TICKET_ROUTE]) rather than a plan edge, its exit goes BACK to the gate that opened the - * ticket (origin of the latest [FailureTicketOpenedEvent] routed here) so that gate re-runs. This - * holds for any repair destination — the file's author OR the fallback recovery stage — and for - * arbitrary topology, instead of following the plan's forward edge (which would walk a - * regenerating implementer, or skip intervening stages). Returns null when the stage was NOT - * entered via a ticket (a normal forward visit — the resolver's edge is used). - */ - private fun ticketReturnMove(ctx: EnrichedExecutionContext): TransitionDecision.Move? { - val events = repositories.eventStore.read(ctx.sessionId) - val enteredViaTicket = events.mapNotNull { it.payload as? TransitionExecutedEvent } - .lastOrNull { it.to == ctx.currentStageId } - ?.transitionId == TICKET_ROUTE - if (!enteredViaTicket) return null - val origin = events.mapNotNull { it.payload as? FailureTicketOpenedEvent } - .lastOrNull { it.routeTo == ctx.currentStageId } - ?.stageId ?: return null - return TransitionDecision.Move(transitionId = TICKET_RETURN, to = origin) - } - /** Returns the cached validated artifact content for the given session + artifact id, or null if absent. */ fun validatedArtifactContent(sessionId: SessionId, artifactId: ArtifactId): String? = artifactContentCache["${sessionId.value}:${artifactId.value}"] @@ -599,193 +253,6 @@ class DefaultSessionOrchestrator( ) } - private suspend fun executeMove( - ctx: EnrichedExecutionContext, - decision: TransitionDecision.Move, - ): StepResult { - val nextStageId = decision.to - - if (isTerminal(ctx.graph, nextStageId)) { - // Terminal stage is a sentinel — not executed, so don't count it. - return StepResult.Terminal( - completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id), - ) - } - - // Runtime refinement guard: a back-edge (re-entering a stage already visited this - // run, e.g. reviewer→implementer) increments a per-cycle counter recorded as an - // event, so the guard is replay-deterministic. Exceeding the cap escalates to a - // terminal failure instead of looping forever. - if (isBackEdge(repositories.eventStore.read(ctx.sessionId), nextStageId)) { - val cycleKey = "${ctx.currentStageId.value}->${nextStageId.value}" - val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: tuning.defaultMaxRefinement - val iteration = (orchestrationRepository.getState(ctx.sessionId).refinementIterations[cycleKey] ?: 0) + 1 - emit(ctx.sessionId, RefinementIterationEvent(ctx.sessionId, cycleKey, iteration, maxIterations)) - if (iteration > maxIterations) { - return StepResult.Terminal( - failWorkflow( - ctx.sessionId, - nextStageId, - "refinement loop '$cycleKey' exceeded $maxIterations iterations — escalating", - retryExhausted = true, - ), - ) - } - } - - // Emit the transition before executing the next stage. Otherwise the next stage's - // events (InferenceStarted, artifacts, …) are recorded ahead of the TransitionExecuted - // that marks entering it, so the log shows the stage running before it was entered. - val advancedTo = advanceStage(ctx.sessionId, ctx.currentStageId, decision) - return when (val result = enterStage(ctx.copy(currentStageId = advancedTo), nextStageId)) { - is StepResult.Continue -> StepResult.Continue(result.ctx) - is StepResult.Terminal -> result - } - } - - @Suppress("ReturnCount") - private suspend fun enterStage( - ctx: EnrichedExecutionContext, - stageId: StageId, - ): StepResult { - log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") - - if (ctx.graph.stages[stageId]?.metadata?.get("requiresApproval") == "true") { - val alreadyApproved = repositories.eventStore.read(ctx.sessionId).let { events -> - val stageRequestIds = events - .mapNotNull { it.payload as? ApprovalRequestedEvent } - .filter { it.stageId == stageId && it.toolName == null } - .map { it.requestId } - .toSet() - stageRequestIds.isNotEmpty() && events.any { - val decision = it.payload as? ApprovalDecisionResolvedEvent - // A REJECTED decision must NOT satisfy the gate on retry/resume, else the stage - // runs unapproved. Only APPROVED/AUTO_APPROVED count as a prior approval. - decision?.requestId in stageRequestIds && - decision?.outcome != ApprovalOutcome.REJECTED - } - } - if (!alreadyApproved) { - val gateArtifact = ctx.graph.stages[stageId]?.needs?.firstOrNull()?.value - val preview = gateArtifact - ?.let { artifactContentCache["${ctx.sessionId.value}:$it"] } - ?: "Upstream stage has completed. Review and approve to continue to stage ${stageId.value}." - val approved = requestStageApproval(ctx.sessionId, stageId, preview) - if (!approved) { - return StepResult.Terminal( - failWorkflow(ctx.sessionId, stageId, "approval rejected for stage ${stageId.value}", retryExhausted = false) - ) - } - } - } - - while (true) { - if (isCancelled(ctx.sessionId)) { - return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId)) - } - val result = subagentRunner.run( - SubagentRunRequest(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config), - ).outcome - when (result) { - is StageExecutionResult.Success -> { - if (requestClarificationIfNeeded(ctx.sessionId, stageId, ctx.graph)) { - // The stage raised open questions and the operator answered them; loop to - // re-run the stage with the answers injected (no failure-retry budget spent). - continue - } - compactionService?.let { svc -> - val journalState = decisionJournalRepository.getJournal(ctx.sessionId) - val journalText = DecisionJournalRenderer().render(journalState) - val tokenEstimate = estimateTokens(journalText) - svc.compactIfNeeded( - sessionId = ctx.sessionId, - state = journalState, - renderedTokenEstimate = tokenEstimate, - emit = { payload -> emit(ctx.sessionId, payload) }, - ) - } - return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) - } - - is StageExecutionResult.Failure -> { - log.debug( - "[Orchestrator] stage failed session=${ctx.sessionId.value} " + - "stage=${stageId.value} reason=${result.reason} retryable=${result.retryable}", - ) - if (!result.retryable) { - // Non-retryable: let the transition resolver handle the outcome - // (e.g., back-edge via artifact_field_equals on failure) - return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) - } - val refreshedState = orchestrationRepository.getState(ctx.sessionId) - // Retry-agency invariant: a gate this stage lacks the capability to fix must not - // be retried in place (futile). Route to a recovery stage that holds the - // capability, if one exists and the route budget remains. - maybeRouteToRecovery(ctx, stageId, result, refreshedState)?.let { return it } - val gateDecision = retryCoordinator.decide( - sessionId = ctx.sessionId, - stageId = stageId, - gate = result.gate, - failureReason = result.reason, - state = refreshedState, - policy = ctx.config.retryPolicy, - ) - when (gateDecision) { - RetryDecision.Retry -> Unit // retry — loop and re-execute - RetryDecision.Exhausted -> { - decideGateExhaustion( - ctx, stageId, result.gate, result.reason, refreshedState, - )?.let { return it } - // review-gate salvage said CONTINUE — budget was reset by the reducer; - // loop and re-execute the stage with a fresh budget. - } - } - } - } - } - } - - /** - * Hybrid exhaustion (design 2026-07-06-per-gate-retry-budgets.md §3/T6): a deterministic gate - * exhausting its budget fails the stage terminally, unchanged from today's behaviour (just - * per-gate now). The "review" gate instead gets one salvageability judgement — CONTINUE resets - * its budget (via the reducer folding [RetrySalvageDecidedEvent]) and the caller loops to retry; - * FAIL is terminal. A gate that already spent its one salvage reset (state.gateSalvageUsed) fails - * immediately on a second exhaustion, regardless of what the judge would say. - * - * Returns a terminal [StepResult] to fail/route with, the recovery stage's result on RECOVER, - * or null to retry in place (loop and re-execute). - */ - private suspend fun decideGateExhaustion( - ctx: EnrichedExecutionContext, - stageId: StageId, - gate: String, - reason: String, - state: OrchestrationState, - ): StepResult? { - val sessionId = ctx.sessionId - if (gate != "review" || state.gateSalvageUsed.contains(gate)) { - return StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) - } - // No judge wired: degrade safely with a deterministic allow-one-reset-then-fail policy — - // the gateSalvageUsed check above already ensures this fires at most once per gate. - val judgment = salvageJudge?.judge(sessionId, stageId, reason) - ?: SalvageJudgment( - SalvageDecision.CONTINUE, - "no salvage judge wired — deterministic one-time reset", - ) - emit(sessionId, RetrySalvageDecidedEvent(sessionId, stageId, gate, judgment.decision, judgment.rationale)) - return when (judgment.decision) { - SalvageDecision.CONTINUE -> null - SalvageDecision.FAIL -> StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) - // The judge chose recovery: route to the recovery stage (file_write is the capability it - // provides). Degrade to terminal if the graph declares no recovery stage. - SalvageDecision.RECOVER -> - routeToRecovery(ctx, stageId, gate, "file_write", reason, state) - ?: StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) - } - } - private fun ExecutionContext.enrich() = EnrichedExecutionContext( graph, sessionId, stageCount, currentStageId, config, session = repositories.sessionRepository.getSession(sessionId), @@ -795,15 +262,15 @@ class DefaultSessionOrchestrator( // A back-edge re-enters a stage already transitioned into this run (e.g. reviewer→implementer). // Top-level (not a member) to keep the orchestrator off the TooManyFunctions threshold. -private fun isBackEdge(events: List, target: StageId): Boolean = +internal fun isBackEdge(events: List, target: StageId): Boolean = events.any { (it.payload as? TransitionExecutedEvent)?.to == target } -private sealed class StepResult { +internal sealed class StepResult { data class Continue(val ctx: EnrichedExecutionContext) : StepResult() data class Terminal(val result: WorkflowResult) : StepResult() } -private data class ExecutionContext( +internal data class ExecutionContext( val graph: WorkflowGraph, val sessionId: SessionId, val stageCount: Int, @@ -813,7 +280,7 @@ private data class ExecutionContext( val state: OrchestrationState?, ) -private data class EnrichedExecutionContext( +internal data class EnrichedExecutionContext( val graph: WorkflowGraph, val sessionId: SessionId, val stageCount: Int, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorRecovery.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorRecovery.kt new file mode 100644 index 00000000..e8c9c866 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorRecovery.kt @@ -0,0 +1,238 @@ +package com.correx.core.kernel.orchestration +import com.correx.core.events.events.FileWrittenEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.FailureTicketOpenedEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.orchestration.OrchestrationState +import com.correx.core.events.types.StageId +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.kernel.retry.FailureFingerprint +import com.correx.core.transitions.execution.StageExecutionResult +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.core.transitions.resolution.TransitionDecision +import java.util.concurrent.* +import java.util.concurrent.atomic.* + +internal suspend fun DefaultSessionOrchestrator.retryStageOrFail( + ctx: EnrichedExecutionContext, + reason: String, +): WorkflowResult { + val attempt = orchestrationRepository.getState(ctx.sessionId).retryCount + val shouldRetry = retryCoordinator.shouldRetry( + sessionId = ctx.sessionId, + stageId = ctx.currentStageId, + currentAttempt = attempt, + policy = ctx.config.retryPolicy, + failureReason = reason, + ) + if (!shouldRetry) { + return failWorkflow(ctx.sessionId, ctx.currentStageId, reason, retryExhausted = true) + } + return when (val r = enterStage(ctx, ctx.currentStageId)) { + is StepResult.Continue -> step(r.ctx) + is StepResult.Terminal -> r.result + } +} + + +/** + * Retry-agency guard (ticket + recovery model). When [failure]'s gate requires a capability the + * failing stage does not hold (e.g. a build/contract gate needs `file_write` on a write-less + * verifier), retrying the stage in place can never change the outcome. Instead open a + * [FailureTicketOpenedEvent] and route to the graph's recovery stage — which does hold the + * capability — bounded by the stage's own [RECOVERY_ROUTE_BUDGET]. + * + * Returns a [StepResult] when it took over the failure (routed, or budget-exhausted terminal), + * or null to fall through to the normal per-gate retry path. Null in every backward-compatible + * case: gate not capability-gated, stage already has the capability, or no recovery stage exists. + */ + +internal suspend fun DefaultSessionOrchestrator.maybeRouteToRecovery( + ctx: EnrichedExecutionContext, + stageId: StageId, + failure: StageExecutionResult.Failure, + state: OrchestrationState, +): StepResult? { + val requiredCapability = GATE_REQUIRED_CAPABILITY[failure.gate] ?: return null + val stageTools = ctx.graph.stages[stageId]?.allowedTools ?: emptySet() + if (requiredCapability in stageTools) return null // stage has agency — retry in place is valid + return routeToRecovery(ctx, stageId, failure.gate, requiredCapability, failure.reason, state) +} + +/** + * Open a [FailureTicketOpenedEvent] and route [stageId]'s failure down a two-tier repair ladder. + * + * **Tier 1 (owner):** when the gate's evidence names a file a stage actually wrote, the ticket goes + * to that author ([resolveTicketOwner]) — it owns the code and the contract it declared, so it + * patches its own layer. Bounded by [RECOVERY_ROUTE_BUDGET]. + * + * **Tier 2 (arbiter / intent-holder):** when the owner loop spends its budget without settling the + * failure, the failure is a cross-file CONTRACT dispute no single owner can resolve (each side is + * internally consistent; they disagree with each other). The ticket escalates to the synthesized + * recovery stage ([findRecoveryStage]) recast as the intent-holder — given the initial intent as + * authority (see buildRecoveryTicketEntry) and told to reconcile ALL the named files in one pass. + * Bounded by its own independent [INTENT_ROUTE_BUDGET]. An orphan failure (no author) starts here. + * + * Both tiers charge progress-awarely: a route whose failure fingerprint changed from the previous + * one moved the needle, so it is FREE; only a route reproducing the same failure charges the tier's + * budget. Terminal only when every reachable tier is spent. Returns null when neither an owner nor a + * recovery stage exists (caller falls back to the legacy per-gate retry path). Shared by the agency + * guard ([maybeRouteToRecovery]) and the review-gate salvage judge ([decideGateExhaustion]). + */ + +internal suspend fun DefaultSessionOrchestrator.routeToRecovery( + ctx: EnrichedExecutionContext, + stageId: StageId, + gate: String, + requiredCapability: String, + reason: String, + state: OrchestrationState, +): StepResult? { + val events = repositories.eventStore.read(ctx.sessionId) + val owner = resolveTicketOwner(events, reason) + ?.takeIf { it != stageId && requiredCapability in (ctx.graph.stages[it]?.allowedTools ?: emptySet()) } + val arbiter = findRecoveryStage(ctx.graph, stageId) + val fingerprint = FailureFingerprint.of(reason) + val ownerKey = stageId.value + val intentKey = stageId.value + INTENT_BUDGET_SUFFIX + + // (target, budgetKey, budget, escalated). Prefer the owner tier while it has budget; escalate to + // the arbiter tier once the owner loop is spent; null = no tier available now. + val route = when { + owner != null && !budgetExhausted(state, ownerKey, fingerprint, tuning.recoveryRouteBudget) -> + RouteTier(owner, ownerKey, tuning.recoveryRouteBudget, escalated = false) + arbiter != null && !budgetExhausted(state, intentKey, fingerprint, tuning.intentRouteBudget) -> + RouteTier(arbiter, intentKey, tuning.intentRouteBudget, escalated = true) + else -> null + } + if (route == null) { + if (owner == null && arbiter == null) return null // nothing to route to: legacy retry path + return StepResult.Terminal( + failWorkflow( + ctx.sessionId, + stageId, + "repair ladder exhausted for stage ${stageId.value} (gate=$gate): $reason", + retryExhausted = true, + ), + ) + } + + val prev = state.recoveryFailureFingerprints[route.budgetKey] + val progressed = prev != null && prev != fingerprint + val used = state.recoveryRoutes[route.budgetKey] ?: 0 + val routeAttempt = if (progressed) used else used + 1 + emit( + ctx.sessionId, + FailureTicketOpenedEvent( + sessionId = ctx.sessionId, + stageId = stageId, + gate = gate, + category = ticketCategory(gate), + requiredCapability = requiredCapability, + routeTo = route.target, + evidence = reason, + routeAttempt = routeAttempt, + fingerprint = fingerprint, + escalated = route.escalated, + ), + ) + log.info( + "[Orchestrator] failure ticket session={} stage={} gate={} lacks={} -> {}={} " + + "(charged={}/{}, progressed={})", + ctx.sessionId.value, stageId.value, gate, requiredCapability, + if (route.escalated) "arbiter" else "owner", route.target.value, + routeAttempt, route.budget, progressed, + ) + val advancedTo = advanceStage( + ctx.sessionId, + stageId, + TransitionDecision.Move(transitionId = TICKET_ROUTE, to = route.target), + ) + return enterStage(ctx.copy(currentStageId = advancedTo), route.target) +} + +/** A chosen rung of the repair ladder: where to route, which budget it charges, and its tier. */ +private data class RouteTier( + val target: StageId, + val budgetKey: String, + val budget: Int, + val escalated: Boolean, +) + +/** + * Progress-aware budget check for one ladder tier. A route whose failure fingerprint changed from + * the previous route on the same [key] made progress (fixed one cause, surfaced another) → NOT + * exhausted (progress is free). Only a no-progress streak that reaches [budget] is exhausted. + */ + +internal fun DefaultSessionOrchestrator.budgetExhausted( + state: OrchestrationState, + key: String, + fingerprint: String, + budget: Int, +): Boolean { + val prev = state.recoveryFailureFingerprints[key] + val progressed = prev != null && prev != fingerprint + val used = state.recoveryRoutes[key] ?: 0 + return !progressed && used >= budget +} + +/** + * The graph's recovery stage: a stage marked `metadata["role"] == "recovery"` (or, as a + * convenience, one whose id is literally "recovery"), other than [failingStageId] itself. + * Null when the graph declares none — recovery routing is opt-in per workflow. + */ + +internal fun DefaultSessionOrchestrator.findRecoveryStage(graph: WorkflowGraph, failingStageId: StageId): StageId? = + graph.stages.entries.firstOrNull { (id, cfg) -> + id != failingStageId && (cfg.metadata["role"] == "recovery" || id.value == "recovery") + }?.key + +/** + * Route-to-owner resolution: map the failing gate's [evidence] (build/tsc output that names + * files, e.g. "src/App.tsx: TS2322") to the stage that most recently WROTE one of those files, + * via the recorded manifest chain (ToolInvocationRequestedEvent.stageId ← FileWrittenEvent by + * invocationId → path). The last write of a named file wins — its author is who to hand the + * ticket to. Pure over recorded events (replay-deterministic). Null when the evidence names no + * written file (an orphan failure — caller falls back to the synthesized recovery stage). + */ + +internal fun DefaultSessionOrchestrator.resolveTicketOwner(events: List, evidence: String): StageId? { + val tokens = EVIDENCE_PATH_RE.findAll(evidence) + .map { it.value.substringBefore(':').replace('\\', '/') } + .filter { it.length >= MIN_EVIDENCE_TOKEN } + .toSet() + if (tokens.isEmpty()) return null + val invToStage = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent } + .associate { it.invocationId to it.stageId } + return events.mapNotNull { it.payload as? FileWrittenEvent } + .filter { it.postImageHash != null } + .lastOrNull { fw -> + val norm = fw.path.replace('\\', '/') + invToStage.containsKey(fw.invocationId) && tokens.any { norm == it || norm.endsWith("/$it") } + } + ?.let { invToStage[it.invocationId] } +} + +/** + * Return-to-sender for a routed repair stage. When the current stage was entered by a failure + * ticket ([TICKET_ROUTE]) rather than a plan edge, its exit goes BACK to the gate that opened the + * ticket (origin of the latest [FailureTicketOpenedEvent] routed here) so that gate re-runs. This + * holds for any repair destination — the file's author OR the fallback recovery stage — and for + * arbitrary topology, instead of following the plan's forward edge (which would walk a + * regenerating implementer, or skip intervening stages). Returns null when the stage was NOT + * entered via a ticket (a normal forward visit — the resolver's edge is used). + */ + +internal fun DefaultSessionOrchestrator.ticketReturnMove(ctx: EnrichedExecutionContext): TransitionDecision.Move? { + val events = repositories.eventStore.read(ctx.sessionId) + val enteredViaTicket = events.mapNotNull { it.payload as? TransitionExecutedEvent } + .lastOrNull { it.to == ctx.currentStageId } + ?.transitionId == TICKET_ROUTE + if (!enteredViaTicket) return null + val origin = events.mapNotNull { it.payload as? FailureTicketOpenedEvent } + .lastOrNull { it.routeTo == ctx.currentStageId } + ?.stageId ?: return null + return TransitionDecision.Move(transitionId = TICKET_RETURN, to = origin) +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorStep.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorStep.kt new file mode 100644 index 00000000..21eac88c --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestratorStep.kt @@ -0,0 +1,324 @@ +package com.correx.core.kernel.orchestration +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.journal.DecisionJournalRenderer +import com.correx.core.artifacts.ArtifactState +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.RetrySalvageDecidedEvent +import com.correx.core.events.events.SalvageDecision +import com.correx.core.events.orchestration.OrchestrationState +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ArtifactLifecyclePhase +import com.correx.core.events.types.StageId +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.kernel.orchestration.subagent.SubagentRunRequest +import com.correx.core.kernel.retry.RetryDecision +import com.correx.core.transitions.execution.StageExecutionResult +import com.correx.core.transitions.resolution.TransitionDecision +import kotlinx.datetime.Clock +import java.util.concurrent.* +import java.util.concurrent.atomic.* + +@Suppress("LongMethod") +internal tailrec suspend fun DefaultSessionOrchestrator.step(ctx: EnrichedExecutionContext): WorkflowResult { + log.debug( + "[Orchestrator] step session={} stage={} stageCount={}", + ctx.sessionId.value, ctx.currentStageId.value, ctx.stageCount, + ) + if (isCancelled(ctx.sessionId)) return handleCancellation(ctx.sessionId, ctx.currentStageId) + + val enriched = EnrichedExecutionContext( + graph = ctx.graph, + sessionId = ctx.sessionId, + stageCount = ctx.stageCount, + currentStageId = ctx.currentStageId, + config = ctx.config, + state = orchestrationRepository.getState(ctx.sessionId), + session = repositories.sessionRepository.getSession(ctx.sessionId), + ) + + val stageConfig = enriched.graph.stages[enriched.currentStageId] + val targetIds: Set = stageConfig?.produces?.map { it.name }?.toSet() ?: emptySet() + val stageArtifacts: Map = if (targetIds.isEmpty()) { + emptyMap() + } else { + val validated = repositories.eventStore.read(enriched.sessionId) + .mapNotNull { (it.payload as? ArtifactValidatedEvent)?.artifactId } + .filter { it in targetIds } + .toSet() + validated.associateWith { ArtifactState(phase = ArtifactLifecyclePhase.VALIDATED) } + } + + val artifactContent: Map = targetIds.mapNotNull { id -> + val cacheKey = "${enriched.sessionId.value}:${id.value}" + artifactContentCache[cacheKey]?.let { content -> id to content } + }.toMap() + + val resolved = resolveTransition( + enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent, + ) + // Ticket return-to-sender: a repair stage (the file's author, or the fallback recovery stage) + // is entered by kernel routing from whichever stage failed a gate, so its exit must go BACK to + // that exact gate (re-run it) — not follow a plan edge that would walk a regenerating + // implementer or skip intervening stages. Derived from the latest failure ticket routed here + // (replay-deterministic). Overrides the resolved edge; an operator preempt below still wins. + val baseDecision = ticketReturnMove(enriched) ?: resolved + // Preemptive redirect override (freestyle graph re-routing): if the operator confirmed a + // jump (a recorded, unconsumed PreemptRedirectEvent), it overrides the resolver's edge — + // unless the target is unknown or its needs aren't satisfied, in which case it is blocked + // and the normal edge is taken. The decision is pure over the event log (replay-deterministic); + // only the block-event emission is a side effect here. + val decision = when ( + val outcome = PreemptRedirect.decide( + events = repositories.eventStore.read(enriched.sessionId), + graph = enriched.graph, + sessionId = enriched.sessionId, + artifactAvailable = { id -> + !artifactContentCache["${enriched.sessionId.value}:${id.value}"].isNullOrBlank() + }, + nowMs = Clock.System.now().toEpochMilliseconds(), + ) + ) { + is PreemptRedirect.Outcome.Override -> outcome.move + is PreemptRedirect.Outcome.Block -> { + emit(enriched.sessionId, outcome.event) + log.info( + "[Orchestrator] redirect blocked session={} to={} reason={}", + enriched.sessionId.value, outcome.event.toStageId.value, outcome.event.reason, + ) + baseDecision + } + PreemptRedirect.Outcome.None -> baseDecision + } + log.debug( + "[Orchestrator] transition session={} stage={} decision={}", + enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName, + ) + return when (decision) { + is TransitionDecision.Move -> when (val r = executeMove(enriched, decision)) { + is StepResult.Continue -> step(r.ctx) + is StepResult.Terminal -> r.result + } + + is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) { + completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount, enriched.graph.id) + } else { + // No outgoing edge matched: the stage finished but produced nothing that satisfies a + // transition (e.g. the analyst never emitted a validated `analysis`). Treat it as a + // retryable stage failure — a fresh attempt lets the model gather more before writing — + // rather than killing the run. Only terminal once the retry budget is spent. + retryStageOrFail( + enriched, + "no transition condition matched from stage ${enriched.currentStageId.value}", + ) + } + + is TransitionDecision.Blocked -> failWorkflow( + sessionId = enriched.sessionId, + stageId = enriched.currentStageId, + reason = decision.reason, + retryExhausted = false, + ) + + is TransitionDecision.NoMatch -> retryStageOrFail( + enriched, + "no matching transition from stage ${enriched.currentStageId.value}", + ) + } +} + +/** + * A non-terminal stage whose outcome matched no outgoing transition (Stay/NoMatch) produced no + * usable artifact. Route it back through the stage's retry budget instead of failing the run: + * re-execute the same stage if attempts remain, else fail terminally with retryExhausted=true. + */ + +internal suspend fun DefaultSessionOrchestrator.executeMove( + ctx: EnrichedExecutionContext, + decision: TransitionDecision.Move, +): StepResult { + val nextStageId = decision.to + + if (isTerminal(ctx.graph, nextStageId)) { + // Terminal stage is a sentinel — not executed, so don't count it. + return StepResult.Terminal( + completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id), + ) + } + + // Runtime refinement guard: a back-edge (re-entering a stage already visited this + // run, e.g. reviewer→implementer) increments a per-cycle counter recorded as an + // event, so the guard is replay-deterministic. Exceeding the cap escalates to a + // terminal failure instead of looping forever. + if (isBackEdge(repositories.eventStore.read(ctx.sessionId), nextStageId)) { + val cycleKey = "${ctx.currentStageId.value}->${nextStageId.value}" + val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: tuning.defaultMaxRefinement + val iteration = (orchestrationRepository.getState(ctx.sessionId).refinementIterations[cycleKey] ?: 0) + 1 + emit(ctx.sessionId, RefinementIterationEvent(ctx.sessionId, cycleKey, iteration, maxIterations)) + if (iteration > maxIterations) { + return StepResult.Terminal( + failWorkflow( + ctx.sessionId, + nextStageId, + "refinement loop '$cycleKey' exceeded $maxIterations iterations — escalating", + retryExhausted = true, + ), + ) + } + } + + // Emit the transition before executing the next stage. Otherwise the next stage's + // events (InferenceStarted, artifacts, …) are recorded ahead of the TransitionExecuted + // that marks entering it, so the log shows the stage running before it was entered. + val advancedTo = advanceStage(ctx.sessionId, ctx.currentStageId, decision) + return when (val result = enterStage(ctx.copy(currentStageId = advancedTo), nextStageId)) { + is StepResult.Continue -> StepResult.Continue(result.ctx) + is StepResult.Terminal -> result + } +} + +@Suppress("ReturnCount") +internal suspend fun DefaultSessionOrchestrator.enterStage( + ctx: EnrichedExecutionContext, + stageId: StageId, +): StepResult { + log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") + + if (ctx.graph.stages[stageId]?.metadata?.get("requiresApproval") == "true") { + val alreadyApproved = repositories.eventStore.read(ctx.sessionId).let { events -> + val stageRequestIds = events + .mapNotNull { it.payload as? ApprovalRequestedEvent } + .filter { it.stageId == stageId && it.toolName == null } + .map { it.requestId } + .toSet() + stageRequestIds.isNotEmpty() && events.any { + val decision = it.payload as? ApprovalDecisionResolvedEvent + // A REJECTED decision must NOT satisfy the gate on retry/resume, else the stage + // runs unapproved. Only APPROVED/AUTO_APPROVED count as a prior approval. + decision?.requestId in stageRequestIds && + decision?.outcome != ApprovalOutcome.REJECTED + } + } + if (!alreadyApproved) { + val gateArtifact = ctx.graph.stages[stageId]?.needs?.firstOrNull()?.value + val preview = gateArtifact + ?.let { artifactContentCache["${ctx.sessionId.value}:$it"] } + ?: "Upstream stage has completed. Review and approve to continue to stage ${stageId.value}." + val approved = requestStageApproval(ctx.sessionId, stageId, preview) + if (!approved) { + return StepResult.Terminal( + failWorkflow(ctx.sessionId, stageId, "approval rejected for stage ${stageId.value}", retryExhausted = false) + ) + } + } + } + + while (true) { + if (isCancelled(ctx.sessionId)) { + return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId)) + } + val result = subagentRunner.run( + SubagentRunRequest(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config), + ).outcome + when (result) { + is StageExecutionResult.Success -> { + if (requestClarificationIfNeeded(ctx.sessionId, stageId, ctx.graph)) { + // The stage raised open questions and the operator answered them; loop to + // re-run the stage with the answers injected (no failure-retry budget spent). + continue + } + compactionService?.let { svc -> + val journalState = decisionJournalRepository.getJournal(ctx.sessionId) + val journalText = DecisionJournalRenderer().render(journalState) + val tokenEstimate = estimateTokens(journalText) + svc.compactIfNeeded( + sessionId = ctx.sessionId, + state = journalState, + renderedTokenEstimate = tokenEstimate, + emit = { payload -> emit(ctx.sessionId, payload) }, + ) + } + return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) + } + + is StageExecutionResult.Failure -> { + log.debug( + "[Orchestrator] stage failed session=${ctx.sessionId.value} " + + "stage=${stageId.value} reason=${result.reason} retryable=${result.retryable}", + ) + if (!result.retryable) { + // Non-retryable: let the transition resolver handle the outcome + // (e.g., back-edge via artifact_field_equals on failure) + return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) + } + val refreshedState = orchestrationRepository.getState(ctx.sessionId) + // Retry-agency invariant: a gate this stage lacks the capability to fix must not + // be retried in place (futile). Route to a recovery stage that holds the + // capability, if one exists and the route budget remains. + maybeRouteToRecovery(ctx, stageId, result, refreshedState)?.let { return it } + val gateDecision = retryCoordinator.decide( + sessionId = ctx.sessionId, + stageId = stageId, + gate = result.gate, + failureReason = result.reason, + state = refreshedState, + policy = ctx.config.retryPolicy, + ) + when (gateDecision) { + RetryDecision.Retry -> Unit // retry — loop and re-execute + RetryDecision.Exhausted -> { + decideGateExhaustion( + ctx, stageId, result.gate, result.reason, refreshedState, + )?.let { return it } + // review-gate salvage said CONTINUE — budget was reset by the reducer; + // loop and re-execute the stage with a fresh budget. + } + } + } + } + } +} + +/** + * Hybrid exhaustion (design 2026-07-06-per-gate-retry-budgets.md §3/T6): a deterministic gate + * exhausting its budget fails the stage terminally, unchanged from today's behaviour (just + * per-gate now). The "review" gate instead gets one salvageability judgement — CONTINUE resets + * its budget (via the reducer folding [RetrySalvageDecidedEvent]) and the caller loops to retry; + * FAIL is terminal. A gate that already spent its one salvage reset (state.gateSalvageUsed) fails + * immediately on a second exhaustion, regardless of what the judge would say. + * + * Returns a terminal [StepResult] to fail/route with, the recovery stage's result on RECOVER, + * or null to retry in place (loop and re-execute). + */ + +internal suspend fun DefaultSessionOrchestrator.decideGateExhaustion( + ctx: EnrichedExecutionContext, + stageId: StageId, + gate: String, + reason: String, + state: OrchestrationState, +): StepResult? { + val sessionId = ctx.sessionId + if (gate != "review" || state.gateSalvageUsed.contains(gate)) { + return StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) + } + // No judge wired: degrade safely with a deterministic allow-one-reset-then-fail policy — + // the gateSalvageUsed check above already ensures this fires at most once per gate. + val judgment = salvageJudge?.judge(sessionId, stageId, reason) + ?: SalvageJudgment( + SalvageDecision.CONTINUE, + "no salvage judge wired — deterministic one-time reset", + ) + emit(sessionId, RetrySalvageDecidedEvent(sessionId, stageId, gate, judgment.decision, judgment.rationale)) + return when (judgment.decision) { + SalvageDecision.CONTINUE -> null + SalvageDecision.FAIL -> StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) + // The judge chose recovery: route to the recovery stage (file_write is the capability it + // provides). Degrade to terminal if the graph declares no recovery stage. + SalvageDecision.RECOVER -> + routeToRecovery(ctx, stageId, gate, "file_write", reason, state) + ?: StepResult.Terminal(failWorkflow(sessionId, stageId, reason, retryExhausted = true)) + } +}