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.
This commit is contained in:
+13
-546
@@ -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<String, String> = mapOf(
|
||||
internal val GATE_REQUIRED_CAPABILITY: Map<String, String> = 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<ArtifactId> = stageConfig?.produces?.map { it.name }?.toSet() ?: emptySet()
|
||||
val stageArtifacts: Map<ArtifactId, ArtifactState> = 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<ArtifactId, String> = 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<StoredEvent>, 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<StoredEvent>, target: StageId): Boolean =
|
||||
internal fun isBackEdge(events: List<StoredEvent>, 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,
|
||||
|
||||
+238
@@ -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<StoredEvent>, 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)
|
||||
}
|
||||
+324
@@ -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<ArtifactId> = stageConfig?.produces?.map { it.name }?.toSet() ?: emptySet()
|
||||
val stageArtifacts: Map<ArtifactId, ArtifactState> = 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<ArtifactId, String> = 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))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user