diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 6b2236d5..cfcf332f 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -371,7 +371,11 @@ fun main() { // Plan-compile gate: wrap the ExecutionPlanCompiler as a post-stage check so a compile-invalid // architect plan is handed back through the retry-feedback loop instead of parking the session in // ACTIVE (post-planning compile dead-end). Returns null on success, the compiler message on failure. - val planCompiler = ExecutionPlanCompiler(artifactKindRegistry, toolRegistry.all().map { it.name }.toSet()) + val planCompiler = ExecutionPlanCompiler( + artifactKindRegistry, + toolRegistry.all().map { it.name }.toSet(), + injectRecovery = true, + ) val orchestrator = DefaultSessionOrchestrator( repositories = repositories, engines = engines.copy( diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt index 20644a31..221c4e99 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt @@ -69,9 +69,45 @@ data class RetryAttemptedEvent( val charged: Boolean = true, ) : EventPayload -/** Outcome of the hybrid-exhaustion salvage judgement for the semantic review gate. */ +/** + * Outcome of the hybrid-exhaustion salvage judgement for the semantic review gate. + * CONTINUE resets the gate budget and retries the reviewed stage in place; FAIL is terminal; + * RECOVER hands the failure to the recovery stage (same destination as the deterministic-gate + * agency guard) — safe under invariant #7 because recovery output re-enters verification and so + * cannot corrupt state, it only chooses which bounded remediation to spend budget on. + */ @Serializable -enum class SalvageDecision { CONTINUE, FAIL } +enum class SalvageDecision { CONTINUE, FAIL, RECOVER } + +/** + * A verification gate failed on a stage that lacks the capability to change the failure condition + * (the retry-agency invariant: recovery/ticket model). Rather than retry the same write-less stage + * in place — which is futile by construction (live repro: freestyle `final_verification` compiled + * with `allowedTools=[shell]` re-ran the identical broken `npm run build` until the budget drained, + * never able to touch a file) — the orchestrator opens a failure ticket and routes to a recovery + * stage that DOES hold the capability. [category] and [requiredCapability] are derived + * deterministically from [gate] (no LLM), so this drives routing without violating invariant #7. + * [evidence] is the gate's captured failure output; it is recorded here at ticket-open time and + * replay reads it back (invariant #9) — the gate command is never re-run on replay. + */ +@Serializable +@SerialName("FailureTicketOpened") +data class FailureTicketOpenedEvent( + val sessionId: SessionId, + // The stage that failed the gate but lacks the capability to fix it (the write-less verifier). + val stageId: StageId, + val gate: String, + // Deterministic from [gate]: "implementation" | "planning" | "environment". + val category: String, + // The capability the failing stage lacks (e.g. "file_write"); the recovery stage must hold it. + val requiredCapability: String, + // The recovery stage the ticket is routed to. + val routeTo: StageId, + // The gate's captured failure output (command/exit/stderr/findings) that recovery must resolve. + val evidence: String, + // 1-based count of times this stage has been routed to recovery (own, small route budget). + val routeAttempt: Int, +) : EventPayload /** * Recorded when a gate exhausts its retry budget and (per design decision 3) the salvage judge is diff --git a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt index 6a5365a0..ca49d7d5 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt @@ -25,4 +25,10 @@ data class OrchestrationState( // Gates that have already spent their one hybrid-exhaustion salvage reset (review gate only, // see RetrySalvageDecidedEvent) — a second exhaustion for that gate is terminal. val gateSalvageUsed: Set = emptySet(), + // Recovery-routing budget (retry-agency / ticket model): how many times each stage id has been + // routed to a recovery stage because it failed a gate it lacked the capability to fix (see + // FailureTicketOpenedEvent). Deliberately NOT reset by TransitionExecutedEvent — the + // verification→recovery→verification loop must stay bounded across those transitions, exactly as + // refinementIterations does for back-edge loops. + val recoveryRoutes: Map = emptyMap(), ) diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index 36d35c99..79a813c2 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt @@ -59,6 +59,7 @@ import com.correx.core.events.events.RepoKnowledgeRetrievedEvent import com.correx.core.events.events.RepoMapComputedEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.RetrySalvageDecidedEvent +import com.correx.core.events.events.FailureTicketOpenedEvent import com.correx.core.events.events.WorkspaceStateObservedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.SourceFetchedEvent @@ -146,6 +147,7 @@ val eventModule = SerializersModule { subclass(WorkflowCompletedEvent::class) subclass(RetryAttemptedEvent::class) subclass(RetrySalvageDecidedEvent::class) + subclass(FailureTicketOpenedEvent::class) subclass(RefinementIterationEvent::class) subclass(RepoMapComputedEvent::class) subclass(WorkspaceStateObservedEvent::class) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt index 20afac3b..64ce6b77 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt @@ -8,6 +8,7 @@ import com.correx.core.events.events.RepoKnowledgeHit import com.correx.core.context.model.ContextEntry import com.correx.core.context.model.ContextLayer import com.correx.core.context.model.EntryRole +import com.correx.core.events.events.FailureTicketOpenedEvent import com.correx.core.events.events.RefinementIterationEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.StoredEvent @@ -38,6 +39,33 @@ fun buildRetryFeedbackEntry(events: List, stageId: StageId): Contex ) } +/** + * Feeds the open failure ticket into the recovery stage's context. The recovery stage was entered + * by kernel routing (not a normal edge) precisely because an upstream write-less stage failed a gate + * it lacked the capability to fix; without the ticket the recovery stage has no idea what to repair. + * Surfaces the failing stage, the gate, and the captured gate output (evidence, recorded at + * ticket-open per invariant #9) so recovery can act on recorded fact, not a re-observation. + */ +fun buildRecoveryTicketEntry(events: List, stageId: StageId): ContextEntry? { + val ticket = events + .mapNotNull { it.payload as? FailureTicketOpenedEvent } + .lastOrNull { it.routeTo == stageId } ?: return null + val content = "## Recovery ticket\n" + + "Stage '${ticket.stageId.value}' failed the '${ticket.gate}' gate (${ticket.category}) and " + + "lacks the '${ticket.requiredCapability}' capability to fix it, so the failure was routed to " + + "you. Repair the underlying cause, then control returns to verification.\n" + + "### Gate output\n${ticket.evidence}" + return ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L1, + content = content, + sourceType = "recoveryTicket", + sourceId = stageId.value, + tokenEstimate = content.length / 4, + role = EntryRole.SYSTEM, + ) +} + fun criticArtifactIds( events: List, graph: WorkflowGraph, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt index f30095e0..19750f46 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt @@ -7,6 +7,7 @@ import com.correx.core.events.events.RefinementIterationEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.RetrySalvageDecidedEvent import com.correx.core.events.events.SalvageDecision +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.events.WorkflowCompletedEvent @@ -94,6 +95,12 @@ class DefaultOrchestrationReducer : OrchestrationReducer { refinementIterations = state.refinementIterations + (p.cycleKey to p.iteration), ) + // Charge the failing stage's recovery-route budget. routeAttempt is the 1-based count the + // orchestrator computed, so fold it directly (idempotent under replay). + is FailureTicketOpenedEvent -> state.copy( + recoveryRoutes = state.recoveryRoutes + (p.stageId.value to p.routeAttempt), + ) + else -> state } } 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 cd2aac9f..bcac8dfb 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 @@ -15,6 +15,7 @@ import com.correx.core.events.events.ArtifactContentStoredEvent import com.correx.core.events.events.ClarificationAnswer import com.correx.core.events.events.ClarificationAnsweredEvent import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.FailureTicketOpenedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RefinementIterationEvent import com.correx.core.events.events.RetrySalvageDecidedEvent @@ -29,6 +30,7 @@ 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 @@ -54,6 +56,25 @@ private val log = LoggerFactory.getLogger(DefaultSessionOrchestrator::class.java // Fallback refinement-loop cap when a stage does not declare maxRetries. private const val DEFAULT_MAX_REFINEMENT = 3 +// Own, small budget for the retry-agency recovery route: how many times a single stage may be +// routed to a recovery stage before the run fails terminally. Deliberately smaller than the +// per-gate retry budget — recovery is a purpose-built repair-then-reverify loop, not open-ended. +private const val RECOVERY_ROUTE_BUDGET = 2 + +// 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( + "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) { + "plan_compile" -> "planning" + else -> "implementation" +} + class DefaultSessionOrchestrator( private val repositories: OrchestratorRepositories, engines: OrchestratorEngines, @@ -138,6 +159,12 @@ class DefaultSessionOrchestrator( val resolved = resolveTransition( enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent, ) + // Recovery return-to-sender: a recovery stage is entered by kernel routing from whichever + // write-less stage failed a gate, so its exit must go BACK to that exact stage (re-run its + // gate) — not follow a static edge that would 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 = recoveryReturnMove(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 @@ -161,9 +188,9 @@ class DefaultSessionOrchestrator( "[Orchestrator] redirect blocked session={} to={} reason={}", enriched.sessionId.value, outcome.event.toStageId.value, outcome.event.reason, ) - resolved + baseDecision } - PreemptRedirect.Outcome.None -> resolved + PreemptRedirect.Outcome.None -> baseDecision } log.debug( "[Orchestrator] transition session={} stage={} decision={}", @@ -229,6 +256,114 @@ class DefaultSessionOrchestrator( } + /** + * 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 to the graph's recovery stage, + * bounded by [RECOVERY_ROUTE_BUDGET]. Shared by the deterministic agency guard + * ([maybeRouteToRecovery]) and the review-gate salvage judge ([decideGateExhaustion] on + * [SalvageDecision.RECOVER]) so both routing paths reach the same recovery destination. + * + * Returns null when the graph declares no recovery stage (caller falls back to its legacy path); + * a terminal [StepResult] when the route budget is spent; otherwise the recovery stage's result. + */ + private suspend fun routeToRecovery( + ctx: EnrichedExecutionContext, + stageId: StageId, + gate: String, + requiredCapability: String, + reason: String, + state: OrchestrationState, + ): StepResult? { + val recoveryId = findRecoveryStage(ctx.graph, stageId) ?: return null // no recovery stage: legacy path + val used = state.recoveryRoutes[stageId.value] ?: 0 + if (used >= RECOVERY_ROUTE_BUDGET) { + return StepResult.Terminal( + failWorkflow( + ctx.sessionId, + stageId, + "recovery route budget exhausted for stage ${stageId.value} (gate=$gate): $reason", + retryExhausted = true, + ), + ) + } + emit( + ctx.sessionId, + FailureTicketOpenedEvent( + sessionId = ctx.sessionId, + stageId = stageId, + gate = gate, + category = ticketCategory(gate), + requiredCapability = requiredCapability, + routeTo = recoveryId, + evidence = reason, + routeAttempt = used + 1, + ), + ) + log.info( + "[Orchestrator] failure ticket session={} stage={} gate={} lacks={} -> route to recovery={} ({}/{})", + ctx.sessionId.value, stageId.value, gate, requiredCapability, + recoveryId.value, used + 1, RECOVERY_ROUTE_BUDGET, + ) + val advancedTo = advanceStage( + ctx.sessionId, + stageId, + TransitionDecision.Move(transitionId = TransitionId("recovery-route"), to = recoveryId), + ) + return enterStage(ctx.copy(currentStageId = advancedTo), recoveryId) + } + + /** + * 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 + + /** + * When the current stage is a recovery stage that has finished, route it back to the stage the + * failure came from — the origin of the latest [FailureTicketOpenedEvent] routed here — so the + * gate that triggered the route is re-run. This makes recovery correct for arbitrary graph + * topology (a write-less gate anywhere routes here and returns to exactly its own stage) rather + * than relying on a static recovery→terminal edge that would skip intervening stages. Returns + * null when the current stage is not a recovery stage or no ticket routed here (then the + * resolver's normal edge — the compiler's fallback recovery→terminal — is used). + */ + private fun recoveryReturnMove(ctx: EnrichedExecutionContext): TransitionDecision.Move? { + val cfg = ctx.graph.stages[ctx.currentStageId] ?: return null + val isRecovery = cfg.metadata["role"] == "recovery" || ctx.currentStageId.value == "recovery" + if (!isRecovery) return null + val origin = repositories.eventStore.read(ctx.sessionId) + .mapNotNull { it.payload as? FailureTicketOpenedEvent } + .lastOrNull { it.routeTo == ctx.currentStageId } + ?.stageId ?: return null + return TransitionDecision.Move(transitionId = TransitionId("recovery-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}"] @@ -450,6 +585,10 @@ class DefaultSessionOrchestrator( 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, @@ -461,10 +600,9 @@ class DefaultSessionOrchestrator( when (gateDecision) { RetryDecision.Retry -> Unit // retry — loop and re-execute RetryDecision.Exhausted -> { - val terminal = decideGateExhaustion( - ctx.sessionId, stageId, result.gate, result.reason, refreshedState, - ) - if (terminal != null) return StepResult.Terminal(terminal) + 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. } @@ -482,17 +620,19 @@ class DefaultSessionOrchestrator( * 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 the terminal [WorkflowResult] to fail with, or null to retry (loop and re-execute). + * 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( - sessionId: SessionId, + ctx: EnrichedExecutionContext, stageId: StageId, gate: String, reason: String, state: OrchestrationState, - ): WorkflowResult? { + ): StepResult? { + val sessionId = ctx.sessionId if (gate != "review" || state.gateSalvageUsed.contains(gate)) { - return failWorkflow(sessionId, stageId, reason, retryExhausted = true) + 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. @@ -504,7 +644,12 @@ class DefaultSessionOrchestrator( emit(sessionId, RetrySalvageDecidedEvent(sessionId, stageId, gate, judgment.decision, judgment.rationale)) return when (judgment.decision) { SalvageDecision.CONTINUE -> null - SalvageDecision.FAIL -> failWorkflow(sessionId, stageId, reason, retryExhausted = true) + 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)) } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 09b718a1..6df8b8ef 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -505,6 +505,8 @@ abstract class SessionOrchestrator( val agentInstructionsEntries = emptyList() val retryFeedbackEntries = buildRetryFeedbackEntry(sessionEvents, stageId) ?.let { listOf(it) } ?: emptyList() + val recoveryTicketEntries = buildRecoveryTicketEntry(sessionEvents, stageId) + ?.let { listOf(it) } ?: emptyList() val vocabularyEntries = artifactKindRegistry ?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" } ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList() @@ -532,7 +534,7 @@ abstract class SessionOrchestrator( systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries + journalEntries + repoMapEntries + claimedTaskEntries + needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + - clarificationEntries + retryFeedbackEntries, + clarificationEntries + retryFeedbackEntries + recoveryTicketEntries, ) val contextPack = runCatching { contextPackBuilder.build( @@ -1021,7 +1023,7 @@ abstract class SessionOrchestrator( ), ) val deniedByStage = toolCall.function.name != STAGE_COMPLETE_TOOL && - toolCall.function.name !in stageConfig.allowedTools + toolCall.function.name !in stageConfig.effectiveAllowedTools val deniedByReadOnly = !deniedByStage && isReadOnlyMode(sessionId) && (tool?.requiredCapabilities?.contains(ToolCapability.FILE_WRITE) == true) if (deniedByStage || deniedByReadOnly) { @@ -2563,7 +2565,7 @@ abstract class SessionOrchestrator( tools = if (!withTools) { emptyList() } else { - stageConfig.allowedTools + stageConfig.effectiveAllowedTools .mapNotNull { effectives.registry?.resolve(it) } .filter { tool -> // ponytail: filter write tools while read-before-write block is active; restored once a read completes diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt index 28877963..6204be57 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt @@ -52,4 +52,20 @@ data class StageConfig( // inference), so a docs-only plan is left untouched. Explicit build_expectation always wins. val autoBuildGate: Boolean = false, val metadata: Map = emptyMap(), -) +) { + /** + * The tools actually permitted at run time: the stage's declared [allowedTools] plus the + * always-available read-only tools ([ALWAYS_AVAILABLE_READ_TOOLS]). A stage that grants any tool + * can always inspect the workspace (read a file, list a dir) without the planner having to + * remember to list them — this is why a write-less verifier could `shell ls -R` (flooding its + * own context) instead of a scoped `list_dir`. A stage that declares NO tools stays toolless (a + * pure-reasoning stage is left as designed — read tools would flip it into tool-calling mode). + */ + val effectiveAllowedTools: Set + get() = if (allowedTools.isEmpty()) allowedTools else allowedTools + ALWAYS_AVAILABLE_READ_TOOLS + + companion object { + /** Read-only tools every tool-granting stage may call regardless of its declared set. */ + val ALWAYS_AVAILABLE_READ_TOOLS: Set = setOf("file_read", "list_dir") + } +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt index 7857b420..9f7d531a 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt @@ -15,6 +15,20 @@ import com.fasterxml.jackson.module.kotlin.kotlinModule import com.fasterxml.jackson.module.kotlin.readValue private const val TERMINAL = "done" +private const val RECOVERY_STAGE = "recovery" + +// Synthesized recovery stage prompt. Freestyle plans are LLM-emitted and never declare a recovery +// stage, but the kernel's retry-agency guard routes a write-less stage's unfixable gate failure to +// one when it exists. So the compiler always injects it (inert unless a gate actually routes): a +// write-capable stage that reads the failure ticket from context, repairs the cause, and hands +// control back to verification. +private const val RECOVERY_PROMPT = + "You are the recovery stage. An upstream stage failed a verification gate it could not fix " + + "(it lacked write access). The failure ticket — the failing stage, the gate, and its exact " + + "output — is in your context under \"## Recovery ticket\". Read the gate output, inspect the " + + "relevant files (file_read / list_dir), and make the minimal edits that fix the reported " + + "cause. Do not re-scaffold or rewrite working code. When done, control returns to the stage " + + "that failed so its gate re-runs." // Freestyle plans don't specify a per-stage token budget, so without this every compiled stage // inherits StageConfig's 4096 default — 1/8th of what the static role_pipeline execution stages @@ -32,6 +46,12 @@ class ExecutionPlanCompiler( // freestyle driver feeds back for a retry). Empty = the caller didn't supply the tool // universe, so validation is skipped (preserves the bare `ExecutionPlanCompiler(registry)`). private val knownTools: Set = emptySet(), + // When true, synthesize a write-capable `recovery` stage (+ a recovery->terminal edge) into every + // compiled graph. The kernel's retry-agency guard routes a write-less stage's unfixable gate + // failure there; without it the guard finds no recovery stage and falls back to futile in-place + // retries (Vikunja #41). Off by default so the compiler's own unit tests see only plan stages; + // the server's freestyle path (Main.kt) turns it on. + private val injectRecovery: Boolean = false, ) { private val mapper = JsonMapper.builder() .addModule(kotlinModule()) @@ -100,6 +120,32 @@ class ExecutionPlanCompiler( metadata = mapOf("promptInline" to s.prompt), ) } + // Inject a recovery stage unless the plan already declares one. Reached only by the kernel's + // failure-ticket routing (no incoming plan edge), so it is excluded from the reachability + // check below. At run time the kernel returns recovery to the exact stage that failed + // (recoveryReturnMove, from the ticket origin); the recovery->terminal edge synthesized here + // is only a fallback for the impossible case of a recovery stage reached with no ticket, so + // the resolver always has a valid outgoing edge and can never dead-end. + val terminal = terminalStageId(plan) + val recoveryStage: Pair? = + if (!injectRecovery || RECOVERY_STAGE in stageMap.keys.map { it.value }) { + null + } else { + StageId(RECOVERY_STAGE) to StageConfig( + allowedTools = setOf("file_write", "shell"), + tokenBudget = DEFAULT_STAGE_TOKEN_BUDGET, + metadata = mapOf("role" to "recovery", "promptInline" to RECOVERY_PROMPT), + ) + } + val recoveryEdge: TransitionEdge? = recoveryStage?.let { + TransitionEdge( + id = TransitionId("$RECOVERY_STAGE->$terminal"), + from = StageId(RECOVERY_STAGE), + to = StageId(terminal), + condition = ConditionSpec(type = "always_true").toCondition(), + ) + } + val fullStageMap = stageMap + listOfNotNull(recoveryStage) val declared = stageMap.keys.map { it.value }.toSet() val transitions = plan.edges.map { e -> @@ -123,11 +169,13 @@ class ExecutionPlanCompiler( operator = e.condition.operator, ).toCondition(), ) - }.toSet() + }.toSet() + listOfNotNull(recoveryEdge) val start = StageId(plan.stages.first().id) - validateReachability(declared, transitions, start) - return WorkflowGraph(id = workflowId, stages = stageMap, transitions = transitions, start = start) + // The recovery stage is reached by kernel routing, not a plan edge, so exclude it from the + // "unreachable from start" check (it would otherwise always trip it). + validateReachability(declared, transitions, start, excluded = setOfNotNull(recoveryStage?.first?.value)) + return WorkflowGraph(id = workflowId, stages = fullStageMap, transitions = transitions, start = start) } /** @@ -174,7 +222,12 @@ class ExecutionPlanCompiler( private fun isGlob(path: String): Boolean = path.any { it == '*' || it == '?' || it == '[' } - private fun validateReachability(declared: Set, transitions: Set, start: StageId) { + private fun validateReachability( + declared: Set, + transitions: Set, + start: StageId, + excluded: Set = emptySet(), + ) { // A declared stage with no path from start can never run — typical LLM topology // failure: every stage wired straight to "done" instead of chained, which silently // truncates the plan to its first stage. @@ -187,7 +240,7 @@ class ExecutionPlanCompiler( .filter { it != TERMINAL && reachable.add(it) } .toSet() } - val unreachable = declared - reachable + val unreachable = declared - reachable - excluded if (unreachable.isNotEmpty()) { throw WorkflowValidationException( "stages unreachable from start '${start.value}': ${unreachable.sorted().joinToString(", ")} — " + diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt index d6ce2d9a..bb9bd5f7 100644 --- a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt @@ -18,6 +18,7 @@ class ExecutionPlanCompilerTest { it.register(ConfigArtifactKind(id = "patch", schema = trivialSchema, llmEmitted = true)) } private val compiler = ExecutionPlanCompiler(registry) + private val recoveringCompiler = ExecutionPlanCompiler(registry, injectRecovery = true) private val validPlan = """ { @@ -72,6 +73,42 @@ class ExecutionPlanCompilerTest { assertEquals(setOf("patch"), reviewStage.needs.map { it.value }.toSet()) } + @Test + fun `injectRecovery synthesizes a write-capable recovery stage looping back to the terminal`() { + val graph = recoveringCompiler.compile(validPlan, "recovery-workflow") + + // Two plan stages + the synthesized recovery stage; still reachable-valid (recovery is + // excluded from the reachability check because it's entered by kernel routing, not an edge). + assertEquals(3, graph.stages.size) + val recovery = graph.stages[StageId("recovery")] + assertNotNull(recovery) + assertEquals("recovery", recovery.metadata["role"]) + assertTrue(recovery.allowedTools.contains("file_write")) + // file_read / list_dir are always available to any tool-granting stage. + assertTrue(recovery.effectiveAllowedTools.contains("file_read")) + + // recovery -> review (the plan's terminal stage, the one whose edge points at 'done'). + val recoveryEdge = graph.transitions.single { it.from == StageId("recovery") } + assertEquals(StageId("review"), recoveryEdge.to) + } + + @Test + fun `injectRecovery is a no-op when the plan already declares a recovery stage`() { + val plan = """ + { + "goal": "x", + "stages": [ + { "id": "recovery", "prompt": "fix", "produces": "patch", "needs": [], "tools": ["file_write"] } + ], + "edges": [ + { "from": "recovery", "to": "done", "condition": { "type": "always_true" } } + ] + } + """.trimIndent() + val graph = recoveringCompiler.compile(plan, "wf") + assertEquals(1, graph.stages.size) + } + @Test fun `plan-declared writes are derived into the stage write manifest`() { val plan = """ diff --git a/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt b/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt index cc797237..85b27796 100644 --- a/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt +++ b/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt @@ -14,6 +14,7 @@ import com.correx.core.events.events.ReviewFinding import com.correx.core.events.events.ReviewFindingsRaisedEvent import com.correx.core.events.events.ReviewSeverity import com.correx.core.events.events.ReviewVerdict +import com.correx.core.events.events.FailureTicketOpenedEvent import com.correx.core.events.events.RetrySalvageDecidedEvent import com.correx.core.events.events.SalvageDecision import com.correx.core.events.events.WorkflowFailedEvent @@ -311,6 +312,61 @@ class GateRetryBudgetExhaustionTest { assertTrue(blockedReviews >= 2, "expected at least 2 blocked review attempts (pre- and post-salvage)") } + // Stage A opts into the (always-failing) review gate; a sibling recovery stage holds file_write. + // Recovery has NO outgoing edge — the kernel returns it to the ticket origin (A) dynamically. + private fun recoveryReviewGraph(): WorkflowGraph = WorkflowGraph( + id = "gate-review-recover-test", + stages = mapOf( + StageId("A") to StageConfig(allowedTools = setOf("file_write"), semanticReview = true), + StageId("recovery") to StageConfig( + allowedTools = setOf("file_write"), + metadata = mapOf("role" to "recovery"), + ), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + + @Test + fun `review gate exhaustion with RECOVER salvage routes to the recovery stage and exhausts the route budget`(): Unit = + runBlocking { + // Slice 2: the salvage judge chooses RECOVER instead of CONTINUE/FAIL, so a review-gate + // exhaustion hands off to the recovery stage (same destination as the deterministic agency + // guard) rather than retrying A in place or failing outright. Bounded by RECOVERY_ROUTE_BUDGET. + val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, _ -> + SalvageJudgment(SalvageDecision.RECOVER, "hand this to recovery") + } + val (orchestrator, eventStore) = buildOrchestrator(alwaysFailingReviewer, salvageJudge) + val sessionId = SessionId("gate-review-recover") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0, perGateMaxAttempts = mapOf("review" to 1)), + ) + + val result = orchestrator.run(sessionId, recoveryReviewGraph(), config) + + assertTrue(result is WorkflowResult.Failed, "route budget must exhaust terminally") + assertTrue((result as WorkflowResult.Failed).retryExhausted) + + val events = eventStore.read(sessionId) + val tickets = events.mapNotNull { it.payload as? FailureTicketOpenedEvent } + assertEquals(2, tickets.size, "one ticket per route, bounded by RECOVERY_ROUTE_BUDGET") + tickets.forEach { + assertEquals("A", it.stageId.value) + assertEquals("review", it.gate) + assertEquals("implementation", it.category) + assertEquals("file_write", it.requiredCapability) + assertEquals("recovery", it.routeTo.value) + } + assertEquals(listOf(1, 2), tickets.map { it.routeAttempt }) + // Every salvage decision on this path is RECOVER (judge consulted afresh each re-entry + // because TransitionExecuted resets gateSalvageUsed). + val salvageDecisions = events.mapNotNull { it.payload as? RetrySalvageDecidedEvent } + assertTrue(salvageDecisions.size >= 2, "judge consulted on each review exhaustion") + assertTrue(salvageDecisions.all { it.decision == SalvageDecision.RECOVER }) + } + @Test fun `review gate exhaustion with FAIL salvage is terminal immediately`(): Unit = runBlocking { val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, _ -> diff --git a/testing/integration/src/test/kotlin/RecoveryRoutingTest.kt b/testing/integration/src/test/kotlin/RecoveryRoutingTest.kt new file mode 100644 index 00000000..db8148a5 --- /dev/null +++ b/testing/integration/src/test/kotlin/RecoveryRoutingTest.kt @@ -0,0 +1,257 @@ +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.domain.ApprovalEngine +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.approvals.model.ApprovalGrant +import com.correx.core.approvals.model.DomainApprovalRequest +import com.correx.core.artifacts.DefaultArtifactReducer +import com.correx.core.events.EventDispatcher +import com.correx.core.events.events.FailureTicketOpenedEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.types.ProviderId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId +import com.correx.core.inference.CapabilityScore +import com.correx.core.inference.FinishReason +import com.correx.core.inference.InferenceProvider +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceRequest +import com.correx.core.inference.InferenceResponse +import com.correx.core.inference.InferenceRouter +import com.correx.core.inference.InferenceState +import com.correx.core.inference.ModelCapability +import com.correx.core.inference.ProviderHealth +import com.correx.core.inference.Tokenizer +import com.correx.core.inference.ToolCallFunction +import com.correx.core.inference.ToolCallRequest +import com.correx.core.inference.TokenUsage +import com.correx.core.journal.DecisionJournalProjector +import com.correx.core.journal.DefaultDecisionJournalReducer +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.kernel.orchestration.OrchestrationProjector +import com.correx.core.kernel.orchestration.OrchestrationRepository +import com.correx.core.kernel.orchestration.OrchestratorEngines +import com.correx.core.kernel.orchestration.OrchestratorRepositories +import com.correx.core.kernel.orchestration.StaticAnalysisRunResult +import com.correx.core.kernel.orchestration.StaticAnalysisRunner +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import com.correx.core.sessions.DefaultSessionReducer +import com.correx.core.sessions.DefaultSessionRepository +import com.correx.core.sessions.SessionProjector +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.core.toolintent.WorkspacePolicy +import com.correx.core.tools.contract.Tool +import com.correx.core.tools.registry.ToolRegistry +import com.correx.core.transitions.graph.StageConfig +import com.correx.core.transitions.graph.TransitionEdge +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.core.transitions.resolution.DefaultTransitionResolver +import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.tools.SandboxedToolExecutor +import com.correx.infrastructure.tools.shell.ShellTool +import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore +import com.correx.testing.fixtures.context.ContextFixtures +import com.correx.testing.fixtures.cyclePolicyMissingValidator +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Instant +import java.nio.file.Files +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Retry-agency / recovery-routing coverage. A write-less `verify` stage runs `shell true` (so it + * passes the producer gate, exactly as #41's `final_verification` ran shell) then fails a + * `static_analysis` gate (fake runner always non-clean). Because `verify`'s allowedTools lack the + * `file_write` capability that gate requires, the orchestrator must NOT retry it in place (futile) — + * it opens a [FailureTicketOpenedEvent] and routes to the `recovery` stage (marked + * `metadata["role"]="recovery"`), which loops back to `verify`. After RECOVERY_ROUTE_BUDGET routes + * the run fails terminally. + */ +class RecoveryRoutingTest { + + private val workspace = Files.createTempDirectory("recovery-routing-test") + + /** Odd calls issue a `shell true` tool call; even calls end the round — repeats across re-entries. */ + private class ShellThenStopProvider : InferenceProvider { + override val id = ProviderId("shell-caller") + override val name = "shell-caller" + override val tokenizer: Tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + private var callCount = 0 + override suspend fun infer(request: InferenceRequest): InferenceResponse { + callCount++ + return if (callCount % 2 == 1) { + InferenceResponse( + requestId = request.requestId, + text = "", + finishReason = FinishReason.ToolCall, + tokensUsed = TokenUsage(10, 5), + latencyMs = 0, + toolCalls = listOf( + ToolCallRequest( + id = "tc-$callCount", + function = ToolCallFunction(name = "shell", arguments = """{"argv":["true"]}"""), + ), + ), + ) + } else { + InferenceResponse( + requestId = request.requestId, + text = "done", + finishReason = FinishReason.Stop, + tokensUsed = TokenUsage(10, 5), + latencyMs = 0, + ) + } + } + override suspend fun healthCheck(): ProviderHealth = ProviderHealth.Healthy + override fun capabilities(): Set = setOf(CapabilityScore(ModelCapability.General, 1.0)) + } + + private class SingleToolRegistry(private val tool: Tool) : ToolRegistry { + override fun resolve(name: String): Tool? = if (name == tool.name) tool else null + override fun all(): List = listOf(tool) + } + + private fun buildOrchestrator(): Pair { + val eventStore = InMemoryEventStore() + val shell = ShellTool(allowedExecutables = setOf("true"), workingDir = workspace) + val toolRegistry = SingleToolRegistry(shell) + val executor = SandboxedToolExecutor( + delegate = shell, + registry = toolRegistry, + eventDispatcher = EventDispatcher(eventStore), + workDir = workspace, + artifactStore = NoopArtifactStore(), + ) + val provider = ShellThenStopProvider() + val inferenceRouter = object : InferenceRouter { + override suspend fun route(stageId: StageId, requiredCapabilities: Set) = provider + } + val failingStaticAnalysis = + StaticAnalysisRunner { _, _ -> StaticAnalysisRunResult(exitCode = 1, output = "TS5023: bad option") } + + val repositories = OrchestratorRepositories( + eventStore = eventStore, + inferenceRepository = InferenceRepository(object : EventReplayer { + override fun rebuild(sessionId: SessionId) = InferenceState() + }), + orchestrationRepository = OrchestrationRepository( + DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())), + ), + sessionRepository = DefaultSessionRepository( + DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer())), + ), + artifactRepository = com.correx.infrastructure.persistence.artifact.LiveArtifactRepository( + eventStore, DefaultArtifactReducer(), + ), + approvalRepository = DefaultApprovalRepository( + DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())), + ), + ) + + val engines = OrchestratorEngines( + transitionResolver = DefaultTransitionResolver { _, _ -> true }, + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = inferenceRouter, + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + riskAssessor = DefaultRiskAssessor(), + toolExecutor = executor, + toolRegistry = toolRegistry, + workspacePolicy = WorkspacePolicy(workspace), + staticAnalysisRunner = failingStaticAnalysis, + approvalEngine = object : ApprovalEngine { + override fun evaluate( + request: DomainApprovalRequest, + context: ApprovalContext, + grants: List, + now: Instant, + ) = ApprovalDecision( + id = null, + requestId = request.id, + outcome = ApprovalOutcome.AUTO_APPROVED, + state = ApprovalStatus.COMPLETED, + tier = request.tier, + contextSnapshot = context, + resolutionTimestamp = now, + reason = "test-auto-approve", + ) + }, + ) + + val decisionJournalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ) + val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = NoopArtifactStore(), + decisionJournalRepository = decisionJournalRepository, + ) + return orchestrator to eventStore + } + + // verify (write-less, failing static gate) --t-verify--> done + // recovery (role=recovery, holds file_write) has NO outgoing edge on purpose: the kernel returns + // it to the ticket's origin stage (recoveryReturnMove), so the recovery->verify loop is driven + // dynamically. If the return-to-sender path regresses, recovery dead-ends and the route count is + // wrong — this graph would fail rather than silently pass on a stale static edge. + private fun recoveryGraph(): WorkflowGraph = WorkflowGraph( + id = "recovery-routing-test", + stages = mapOf( + StageId("verify") to StageConfig( + allowedTools = setOf("shell"), + staticAnalysis = listOf("typecheck"), + ), + StageId("recovery") to StageConfig( + allowedTools = setOf("shell", "file_write"), + metadata = mapOf("role" to "recovery"), + ), + ), + transitions = setOf( + TransitionEdge(TransitionId("t-verify"), StageId("verify"), StageId("done"), condition = { true }), + ), + start = StageId("verify"), + ) + + @Test + fun `write-less gate failure routes to recovery and exhausts its own route budget`(): Unit = runBlocking { + val (orchestrator, eventStore) = buildOrchestrator() + val sessionId = SessionId("recovery-route") + // Generous in-place per-gate budget proves routing pre-empts it: without the agency guard, + // static_analysis would retry verify in place (never able to change the outcome) instead. + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 10, backoffMs = 0)) + + val result = orchestrator.run(sessionId, recoveryGraph(), config) + + assertTrue(result is WorkflowResult.Failed, "route budget must exhaust terminally") + assertTrue((result as WorkflowResult.Failed).retryExhausted) + + val events = eventStore.read(sessionId) + val tickets = events.mapNotNull { it.payload as? FailureTicketOpenedEvent } + assertEquals(2, tickets.size, "one ticket per route, bounded by RECOVERY_ROUTE_BUDGET") + tickets.forEach { + assertEquals("verify", it.stageId.value) + assertEquals("static_analysis", it.gate) + assertEquals("implementation", it.category) + assertEquals("file_write", it.requiredCapability) + assertEquals("recovery", it.routeTo.value) + } + assertEquals(listOf(1, 2), tickets.map { it.routeAttempt }) + assertTrue(events.any { it.payload is WorkflowFailedEvent }) + } +}