From 79e2c38a885b403e7680ab0b9dc34c5510b818b1 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 6 Jul 2026 19:20:46 +0400 Subject: [PATCH] feat(retry): per-gate retry budgets + progress-aware charging + hybrid salvage Replaces the single shared per-stage retryCount (reset on TransitionExecuted, shared across all post-stage gates) with a per-gate budget. Motivated by run 2e468f9f, where a promising freestyle run was terminally FAILED because the contract gate burned all 3 shared retries on one trivial miss before the semantic-review gate ever ran. - StageExecutionResult.Failure gains a `gate` id; each gate tags its failures. - OrchestrationState gains gateRetryBudgets / gateFailureFingerprints / gateSalvageUsed (all rebuilt from events, reset per-stage on TransitionExecuted). - FailureFingerprint normalizes+hashes the failure reason; RetryCoordinator.decide charges a gate's budget only when the fingerprint is unchanged (no progress), so a moving run is never penalised. Returns Retry | Exhausted. - Hybrid exhaustion: deterministic gates fail; the review gate consults a SalvageJudge (LLM, or a deterministic allow-one-reset fallback), recorded as RetrySalvageDecidedEvent (invariant #9). CONTINUE resets the gate budget once; a second exhaustion is terminal. - Review gate is now the sole authority on review-retry termination; the old REVIEW_BLOCK_RETRY_CAP is repurposed as a high absolute backstop only. - ServerModule escaped-exception path now records a truthful terminal WorkflowFailed (real stage/reason/retryExhausted) via recordUnhandledFailure. Tests: DefaultRetryCoordinatorTest (fingerprint charging) + GateRetryBudgetExhaustionTest (deterministic exhaust / review CONTINUE-then- terminal / review FAIL / null-judge fallback) + reducer coverage. --- .../com/correx/apps/server/ServerModule.kt | 62 +-- .../core/events/events/OrchestrationEvents.kt | 28 ++ .../core/events/execution/RetryPolicy.kt | 7 + .../orchestration/OrchestrationState.kt | 10 + .../events/serialization/Serialization.kt | 2 + .../DefaultOrchestrationReducer.kt | 25 +- .../DefaultSessionOrchestrator.kt | 69 +++- .../orchestration/OrchestratorEngines.kt | 4 + .../core/kernel/orchestration/SalvageJudge.kt | 24 ++ .../orchestration/SessionOrchestrator.kt | 218 ++++++++++- .../kernel/retry/DefaultRetryCoordinator.kt | 48 +++ .../core/kernel/retry/FailureFingerprint.kt | 21 ++ .../core/kernel/retry/RetryCoordinator.kt | 30 ++ .../execution/StageExecutionResult.kt | 4 + .../kotlin/GateRetryBudgetExhaustionTest.kt | 356 ++++++++++++++++++ .../kotlin/DefaultRetryCoordinatorTest.kt | 94 +++++ .../test/kotlin/OrchestrationReducerTest.kt | 137 +++++++ 17 files changed, 1085 insertions(+), 54 deletions(-) create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SalvageJudge.kt create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/retry/FailureFingerprint.kt create mode 100644 testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index e2549341..b5131788 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -348,33 +348,53 @@ class ServerModule( .onFailure { log.warn("Profile adaptation failed: {}", it.message) } } } - }.onFailure { ex -> - log.error("runSession: session={} failed: {}", sessionId.value, ex.message, ex) - eventStore.append( - NewEvent( - metadata = EventMetadata( - eventId = EventId(java.util.UUID.randomUUID().toString()), - sessionId = sessionId, - timestamp = Clock.System.now(), - schemaVersion = 1, - causationId = null, - correlationId = null, - ), - payload = WorkflowFailedEvent( - sessionId = sessionId, - stageId = graph.start, - reason = ex.message ?: "unexpected orchestrator failure", - retryExhausted = false, - ), - ), - ) - } + }.onFailure { ex -> recordUnhandledFailure(sessionId, graph, sessionConfig, ex) } activeSessionJobs.remove(sessionId) } } } } + // An exception escaped the orchestrator (the run never reached its own terminal WorkflowFailed). + // Record a truthful terminal event: the real failing stage (not graph.start), the real message, + // and a computed retryExhausted — so a headless client sees an honest failure, not a placeholder. + private suspend fun recordUnhandledFailure( + sessionId: SessionId, + graph: com.correx.core.transitions.graph.WorkflowGraph, + sessionConfig: OrchestrationConfig, + ex: Throwable, + ) { + log.error("runSession: session={} failed: {}", sessionId.value, ex.message, ex) + val orchState = orchestrationRepository.getState(sessionId) + val failingStageId = orchState.currentStageId ?: graph.start + val reason = ex.message + ?.trim() + ?.takeIf { it.isNotEmpty() } + ?: "unexpected orchestrator failure: unhandled exception with no message " + + "during stage '${failingStageId.value}'" + val maxAttempts = orchState.retryPolicy?.maxAttempts + ?: sessionConfig.retryPolicy.maxAttempts + val retryExhausted = orchState.retryCount >= maxAttempts + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(java.util.UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = WorkflowFailedEvent( + sessionId = sessionId, + stageId = failingStageId, + reason = reason, + retryExhausted = retryExhausted, + ), + ), + ) + } + /** * Resolves the workflow graph for resuming a freestyle phase-2 session. Phase-2 graphs * (`freestyle-`) are compiled from the locked plan, never registered in the 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 a4168ccd..20644a31 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 @@ -59,4 +59,32 @@ data class RetryAttemptedEvent( val attemptNumber: Int, val maxAttempts: Int, val failureReason: String, + // Per-gate retry budgets (design 2026-07-06-per-gate-retry-budgets.md): budget key this attempt + // is charged against. Defaulted to the legacy shared "stage" bucket for schema back-compat. + val gate: String = "stage", + // Stable hash of the failure's normalised reason, used to detect progress across attempts. + val fingerprint: String = "", + // Whether this attempt consumed the gate's budget (false when the fingerprint changed from the + // previous attempt — i.e. the run is moving, so the retry is free). + val charged: Boolean = true, +) : EventPayload + +/** Outcome of the hybrid-exhaustion salvage judgement for the semantic review gate. */ +@Serializable +enum class SalvageDecision { CONTINUE, FAIL } + +/** + * Recorded when a gate exhausts its retry budget and (per design decision 3) the salvage judge is + * consulted — currently only the "review" gate. The judgement is nondeterministic (LLM-backed, or a + * deterministic fallback when no judge is wired), so invariant #9 requires it be recorded here; + * replay reads this event back and never re-invokes the judge. + */ +@Serializable +@SerialName("RetrySalvageDecided") +data class RetrySalvageDecidedEvent( + val sessionId: SessionId, + val stageId: StageId, + val gate: String, + val decision: SalvageDecision, + val rationale: String, ) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/execution/RetryPolicy.kt b/core/events/src/main/kotlin/com/correx/core/events/execution/RetryPolicy.kt index a2744e20..b9f0ea8b 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/execution/RetryPolicy.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/execution/RetryPolicy.kt @@ -6,8 +6,15 @@ import kotlinx.serialization.Serializable data class RetryPolicy( val maxAttempts: Int, val backoffMs: Long = 1_000L, + // Per-gate override (design 2026-07-06-per-gate-retry-budgets.md §8): a gate id present here uses + // its own budget instead of [maxAttempts]. Empty by default — every gate then independently gets + // [maxAttempts] (un-shared, per T2/T5), preserving today's numbers with no TOML changes required. + val perGateMaxAttempts: Map = emptyMap(), ) { init { require(maxAttempts >= 1) { "maxAttempts must be >= 1" } + require(perGateMaxAttempts.values.all { it >= 1 }) { "perGateMaxAttempts values must be >= 1" } } + + fun maxAttemptsFor(gate: String): Int = perGateMaxAttempts[gate] ?: maxAttempts } 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 7e0a5a4b..6a5365a0 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 @@ -15,4 +15,14 @@ data class OrchestrationState( val failureReason: String? = null, val retryPolicy: RetryPolicy? = null, val refinementIterations: Map = emptyMap(), + // Per-gate retry budget (design 2026-07-06-per-gate-retry-budgets.md): each gate id (produces, + // brief_grounding, brief_echo, contract, plan_compile, static_analysis, execution, review, ...) + // exhausts its own budget rather than sharing retryCount session-wide. Attempts charged so far. + val gateRetryBudgets: Map = emptyMap(), + // Last-seen failure fingerprint per gate, used to tell a no-progress retry (same fingerprint, + // charged) from a genuine-progress retry (changed fingerprint, free). + val gateFailureFingerprints: Map = emptyMap(), + // 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(), ) 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 097246f4..36d35c99 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 @@ -58,6 +58,7 @@ import com.correx.core.events.events.RefinementIterationEvent 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.WorkspaceStateObservedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.SourceFetchedEvent @@ -144,6 +145,7 @@ val eventModule = SerializersModule { subclass(WorkflowFailedEvent::class) subclass(WorkflowCompletedEvent::class) subclass(RetryAttemptedEvent::class) + subclass(RetrySalvageDecidedEvent::class) subclass(RefinementIterationEvent::class) subclass(RepoMapComputedEvent::class) subclass(WorkspaceStateObservedEvent::class) 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 b246cace..f30095e0 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 @@ -5,6 +5,8 @@ import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent 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.StoredEvent import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.events.WorkflowCompletedEvent @@ -59,14 +61,35 @@ class DefaultOrchestrationReducer : OrchestrationReducer { // turning the next stage's first retryable failure into a terminal WorkflowFailed (live repro: // implement burned 3/3, then review's retryable JSON-decode error got 0 retries). Back-edge // loops stay bounded by the separate refinementIterations counter, so this can't loop forever. - is TransitionExecutedEvent -> state.copy(currentStageId = p.to, retryCount = 0) + is TransitionExecutedEvent -> state.copy( + currentStageId = p.to, + retryCount = 0, + gateRetryBudgets = emptyMap(), + gateFailureFingerprints = emptyMap(), + gateSalvageUsed = emptySet(), + ) is RetryAttemptedEvent -> state.copy( status = OrchestrationStatus.RUNNING, retryCount = p.attemptNumber, failureReason = null, + gateRetryBudgets = if (p.charged) { + state.gateRetryBudgets + (p.gate to ((state.gateRetryBudgets[p.gate] ?: 0) + 1)) + } else { + state.gateRetryBudgets + }, + gateFailureFingerprints = state.gateFailureFingerprints + (p.gate to p.fingerprint), ) + is RetrySalvageDecidedEvent -> if (p.decision == SalvageDecision.CONTINUE) { + state.copy( + gateRetryBudgets = state.gateRetryBudgets + (p.gate to 0), + gateSalvageUsed = state.gateSalvageUsed + p.gate, + ) + } else { + state + } + is RefinementIterationEvent -> state.copy( refinementIterations = state.refinementIterations + (p.cycleKey to p.iteration), ) 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 18fd297c..cd2aac9f 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 @@ -17,6 +17,8 @@ import com.correx.core.events.events.ClarificationAnsweredEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.RetrySalvageDecidedEvent +import com.correx.core.events.events.SalvageDecision import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.orchestration.OrchestrationState @@ -32,6 +34,7 @@ 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.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 @@ -68,6 +71,11 @@ class DefaultSessionOrchestrator( override val cancellations: ConcurrentHashMap = ConcurrentHashMap() + // 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 + override val subagentRunner: SubagentRunner = InSessionSubagentRunner( executeStage = { sid, stg, graph, session, cfg -> executeStage(sid, stg, graph, session, cfg, effectivesFor(cfg)) @@ -442,29 +450,64 @@ class DefaultSessionOrchestrator( return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) } val refreshedState = orchestrationRepository.getState(ctx.sessionId) - val shouldRetry = retryCoordinator.shouldRetry( + val gateDecision = retryCoordinator.decide( sessionId = ctx.sessionId, stageId = stageId, - currentAttempt = refreshedState.retryCount, - policy = ctx.config.retryPolicy, + gate = result.gate, failureReason = result.reason, + state = refreshedState, + policy = ctx.config.retryPolicy, ) - if (!shouldRetry) { - return StepResult.Terminal( - failWorkflow( - ctx.sessionId, - stageId, - result.reason, - retryExhausted = true, - ), - ) + 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) + // review-gate salvage said CONTINUE — budget was reset by the reducer; + // loop and re-execute the stage with a fresh budget. + } } - // retry — loop and re-execute } } } } + /** + * 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 the terminal [WorkflowResult] to fail with, or null to retry (loop and re-execute). + */ + private suspend fun decideGateExhaustion( + sessionId: SessionId, + stageId: StageId, + gate: String, + reason: String, + state: OrchestrationState, + ): WorkflowResult? { + if (gate != "review" || state.gateSalvageUsed.contains(gate)) { + return 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 -> failWorkflow(sessionId, stageId, reason, retryExhausted = true) + } + } + private fun ExecutionContext.enrich() = EnrichedExecutionContext( graph, sessionId, stageCount, currentStageId, config, session = repositories.sessionRepository.getSession(sessionId), diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt index 16c2a9fa..a00b2ae5 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt @@ -43,4 +43,8 @@ data class OrchestratorEngines( // Gate 3). Null = no semantic review; the funnel then rests on the deterministic gates only. A // stage opts in via StageConfig.semanticReview. val semanticReviewer: SemanticReviewer? = null, + // Hybrid-exhaustion salvage judge (design 2026-07-06-per-gate-retry-budgets.md §3/T7), consulted + // only when the "review" gate exhausts its retry budget. Null = no LLM judge wired; the + // orchestrator then falls back to a deterministic allow-one-reset-then-fail policy. + val salvageJudge: SalvageJudge? = null, ) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SalvageJudge.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SalvageJudge.kt new file mode 100644 index 00000000..bdc6e0f3 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SalvageJudge.kt @@ -0,0 +1,24 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.events.events.SalvageDecision +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId + +/** Result of a [SalvageJudge.judge] call. */ +data class SalvageJudgment(val decision: SalvageDecision, val rationale: String) + +/** + * Seam for the hybrid-exhaustion salvage judgement (design 2026-07-06-per-gate-retry-budgets.md + * §3/T7): consulted when the "review" gate exhausts its per-gate retry budget. Like the other gate + * seams ([SemanticReviewer], [ContractAssertionEvaluator]), the inference-touching implementation is + * injected so the deterministic core never runs inference itself. + * + * The judgement is nondeterministic (LLM-backed), so invariant #9 requires the caller to record the + * outcome as a [com.correx.core.events.events.RetrySalvageDecidedEvent] — replay reads that event + * back and never re-invokes this judge. When no judge is wired (`null` in [OrchestratorEngines]), + * the orchestrator falls back to a deterministic allow-one-reset-then-fail policy so the feature + * degrades safely without inference (see `DefaultSessionOrchestrator.decideSalvage`). + */ +fun interface SalvageJudge { + suspend fun judge(sessionId: SessionId, stageId: StageId, reason: String): SalvageJudgment +} 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 a06bd4cb..ab977122 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 @@ -223,12 +223,16 @@ private const val MAX_CLARIFICATION_ROUNDS = 3 // Static-analysis output caps: the tail retained in the recorded event vs. the (larger) slice fed // back to the model so it can fix the failure. Tails, because the error summary sits at the end. private const val CONTRACT_EVIDENCE_CAP = 400 -// Gate 3 semantic review: a FAIL blocks only on a correctness finding at/above this confidence, and -// only until this many prior blocks have occurred for the stage — after which findings surface -// without blocking, so a stuck reviewer can never trap a stage. Objective text fed to the reviewer -// is capped to keep the review prompt bounded. +// Gate 3 semantic review: a FAIL blocks only on a correctness finding at/above this confidence. +// Per-gate retry budgeting + hybrid salvage (design 2026-07-06-per-gate-retry-budgets.md) is now the +// authority on when review retries stop — the gate simply reports the block and the RetryCoordinator +// charges/exhausts the "review" budget, then the salvage judge decides continue-vs-fail. This cap +// remains only as an absolute backstop against a pathological run whose review findings *change* +// every attempt (so the budget never charges, i.e. the run looks like perpetual "progress"): after +// this many blocks, findings surface without blocking so a stuck reviewer can never trap a stage +// forever. Set well above any legitimate budget+salvage sequence. Objective text is capped too. private const val REVIEW_BLOCK_MIN_CONFIDENCE = 0.7 -private const val REVIEW_BLOCK_RETRY_CAP = 2 +private const val REVIEW_BLOCK_RETRY_CAP = 20 private const val REVIEW_OBJECTIVE_CAP = 4_000 private const val STATIC_ANALYSIS_SUMMARY_CAP = 2_000 private const val STATIC_ANALYSIS_FEEDBACK_CAP = 6_000 @@ -1929,6 +1933,7 @@ abstract class SessionOrchestrator( return StageExecutionResult.Failure( "stage ${stageId.value} declared no artifacts and ran no tools", retryable = true, + gate = "produces", ) } val producedIds = stageConfig.produces.map { it.name }.toSet() @@ -1945,6 +1950,7 @@ abstract class SessionOrchestrator( return StageExecutionResult.Failure( "stage ${stageId.value} did not produce declared artifacts: $missingIds", retryable = true, + gate = "produces", ) } @@ -2015,6 +2021,7 @@ abstract class SessionOrchestrator( "stage ${stageId.value} references paths that do not exist in the workspace: " + "${missing.joinToString(", ")}. Reference only files you have read.", retryable = true, + gate = "brief_grounding", ) } } @@ -2077,6 +2084,7 @@ abstract class SessionOrchestrator( "files not in the brief: [${div.hallucinatedFiles.joinToString(", ")}]. " + "Restate the brief exactly; do not drop requirements or invent files.", retryable = true, + gate = "brief_echo", ) } else { StageExecutionResult.Success(emptyList()) @@ -2174,6 +2182,7 @@ abstract class SessionOrchestrator( return StageExecutionResult.Failure( "stage ${stageId.value} did not satisfy its file contract. Fix these before review:\n$detail", retryable = true, + gate = "contract", ) } @@ -2209,6 +2218,7 @@ abstract class SessionOrchestrator( "stage ${stageId.value} produced an execution_plan that does not compile. " + "Fix the plan before it can be locked:\n$error", retryable = true, + gate = "plan_compile", ) } @@ -2283,6 +2293,7 @@ abstract class SessionOrchestrator( return StageExecutionResult.Failure( "stage ${stageId.value} did not pass static analysis. Fix these before review:\n\n$detail", retryable = true, + gate = "static_analysis", ) } @@ -2330,6 +2341,7 @@ abstract class SessionOrchestrator( "Fix these before proceeding:\n\n$ $command (exit ${run.exitCode})\n" + run.output.takeLast(STATIC_ANALYSIS_FEEDBACK_CAP), retryable = true, + gate = "execution", ) } @@ -2402,6 +2414,7 @@ abstract class SessionOrchestrator( return StageExecutionResult.Failure( "stage ${stageId.value} failed semantic review — fix these correctness issues:\n$detail", retryable = true, + gate = "review", ) } @@ -2560,6 +2573,7 @@ abstract class SessionOrchestrator( if (isCancelled(sessionId)) return InferenceResult.Cancelled val responseArtifactId = artifactStore.put(response.text.toByteArray()) + val reasoningArtifactId = response.reasoning?.let { artifactStore.put(it.toByteArray()) } emit( sessionId, InferenceCompletedEvent( @@ -2570,6 +2584,7 @@ abstract class SessionOrchestrator( tokensUsed = response.tokensUsed, latencyMs = response.latencyMs, responseArtifactId = responseArtifactId, + reasoningArtifactId = reasoningArtifactId, ), ) @@ -2687,8 +2702,35 @@ abstract class SessionOrchestrator( "[Orchestrator] FAILED session={} stage={} reason={} retryExhausted={}", sessionId.value, stageId.value, reason, retryExhausted, ) - correlateCritiqueOutcomes(sessionId) - emit(sessionId, WorkflowFailedEvent(sessionId, stageId, reason, retryExhausted)) + // failWorkflow IS the clean-termination path: whatever goes wrong while recording it + // (event-store hiccup, correlation lookup, or any other transient failure) must never be + // allowed to escape as a raw exception — that would blow past this function's own + // WorkflowFailedEvent (correct stageId/reason/retryExhausted) and land in the server's + // top-level catch-all, which fabricates its own event with the wrong stage (graph.start) + // and a garbage reason (the escaping throwable's message/toString). Record what we can and + // still return a clean Failed result either way. + // Critique correlation is best-effort bookkeeping; if it throws we log and press on, because + // the terminal WorkflowFailedEvent below MUST still be recorded — skipping it would leave the + // event log with no terminal marker for the session (breaks replay-completeness, invariants + // #1/#8) and let the server catch-all fabricate a wrong one. + runCatching { correlateCritiqueOutcomes(sessionId) }.onFailure { e -> + log.error( + "[Orchestrator] failWorkflow: critique correlation failed session={} stage={}", + sessionId.value, stageId.value, e, + ) + } + // Last-ditch: the one unrecoverable case. If emitting the terminal event itself throws + // (e.g. a serialization edge), we cannot record the failure at all — log it loudly with the + // full throwable so it is never silent, then still return a clean Failed result. + runCatching { + emit(sessionId, WorkflowFailedEvent(sessionId, stageId, reason, retryExhausted)) + }.onFailure { e -> + log.error( + "[Orchestrator] failWorkflow: FAILED to record terminal WorkflowFailedEvent — event " + + "log has no terminal marker for session={} stage={}", + sessionId.value, stageId.value, e, + ) + } cancellations.remove(sessionId) return WorkflowResult.Failed(sessionId, reason, retryExhausted) } @@ -2961,6 +3003,7 @@ abstract class SessionOrchestrator( private suspend fun computeToolPreview(toolName: String, parameters: Map): String? { if (toolName == "shell") return shellCommandPreview(parameters) if (toolName == "task_decompose") return renderDecomposePreview(parameters) + if (toolName == "file_edit") return computeFileEditPreview(parameters) if (toolName != "file_write") return null val path = parameters["path"] as? String ?: return null // file_write no longer carries an `operation` param (delete was split into file_delete), so the @@ -2968,13 +3011,39 @@ private suspend fun computeToolPreview(toolName: String, parameters: Map): String? { + val path = parameters["path"] as? String ?: return null + val operation = parameters["operation"] as? String ?: return null + val existingContent = readFileIfExists(path) ?: return null + + val proposedContent = when (operation) { + "append" -> existingContent + (parameters["content"] as? String ?: return null) + "replace" -> { + val target = parameters["target"] as? String ?: return null + val replacement = parameters["replacement"] as? String ?: return null + if (!existingContent.contains(target)) return null + existingContent.replaceFirst(target, replacement) + } + else -> return null } return buildDiffString(path, existingContent, proposedContent) @@ -3003,25 +3072,136 @@ private fun shellQuoteToken(token: String): String { return if (needsQuote) "'" + token.replace("'", "'\\''") + "'" else token } +private enum class DiffOpKind { EQUAL, DELETE, INSERT } +private data class DiffOp(val kind: DiffOpKind, val line: String) + /** - * Build a unified-diff-style string for a full file replacement. + * Build a unified-diff string for a full file replacement, using a line-level LCS diff + * so a small edit to a large file shows as a tight hunk rather than a full-file replacement. * When the file doesn't exist yet (new file), only `+` lines are shown. - * Uses a simple full-file diff: all old lines as `-`, all new lines as `+`. */ private fun buildDiffString(path: String, existingContent: String?, proposedContent: String): String { val existingLines = existingContent?.lines() ?: emptyList() val proposedLines = proposedContent.lines() if (existingLines == proposedLines) return " (no change)" + val ops = diffLines(existingLines, proposedLines) return buildString { appendLine("--- a/$path") appendLine("+++ b/$path") - appendLine("@@ -1,${existingLines.size.coerceAtLeast(1)} +1,${proposedLines.size.coerceAtLeast(1)} @@") - existingLines.forEach { appendLine("-$it") } - proposedLines.forEach { appendLine("+$it") } + append(renderHunks(ops)) }.trimEnd() } +/** Line-level LCS diff producing an edit script of EQUAL/DELETE/INSERT ops. */ +private fun diffLines(oldLines: List, newLines: List): List { + val n = oldLines.size + val m = newLines.size + val lcs = Array(n + 1) { IntArray(m + 1) } + for (i in n - 1 downTo 0) { + for (j in m - 1 downTo 0) { + lcs[i][j] = if (oldLines[i] == newLines[j]) { + lcs[i + 1][j + 1] + 1 + } else { + maxOf(lcs[i + 1][j], lcs[i][j + 1]) + } + } + } + val ops = mutableListOf() + var i = 0 + var j = 0 + while (i < n && j < m) { + when { + oldLines[i] == newLines[j] -> { + ops += DiffOp(DiffOpKind.EQUAL, oldLines[i]) + i++ + j++ + } + lcs[i + 1][j] >= lcs[i][j + 1] -> { + ops += DiffOp(DiffOpKind.DELETE, oldLines[i]) + i++ + } + else -> { + ops += DiffOp(DiffOpKind.INSERT, newLines[j]) + j++ + } + } + } + while (i < n) { + ops += DiffOp(DiffOpKind.DELETE, oldLines[i]) + i++ + } + while (j < m) { + ops += DiffOp(DiffOpKind.INSERT, newLines[j]) + j++ + } + return ops +} + +/** Groups an edit script into unified-diff hunks with [context] lines of surrounding EQUAL context. */ +private fun renderHunks(ops: List, context: Int = 3): String { + val changedIndices = ops.indices.filter { ops[it].kind != DiffOpKind.EQUAL } + if (changedIndices.isEmpty()) return "" + + data class Hunk(val start: Int, val end: Int) + val hunks = mutableListOf() + var hunkStart = changedIndices.first() + var hunkEnd = changedIndices.first() + for (idx in changedIndices.drop(1)) { + if (idx - hunkEnd <= context * 2) { + hunkEnd = idx + } else { + hunks += Hunk(hunkStart, hunkEnd) + hunkStart = idx + hunkEnd = idx + } + } + hunks += Hunk(hunkStart, hunkEnd) + + var oldLine = 1 + var newLine = 1 + var opIdx = 0 + return buildString { + for (hunk in hunks) { + val from = maxOf(0, hunk.start - context) + val to = minOf(ops.size - 1, hunk.end + context) + while (opIdx < from) { + if (ops[opIdx].kind != DiffOpKind.INSERT) oldLine++ + if (ops[opIdx].kind != DiffOpKind.DELETE) newLine++ + opIdx++ + } + val oldStart = oldLine + val newStart = newLine + var oldCount = 0 + var newCount = 0 + val lines = StringBuilder() + for (k in from..to) { + val op = ops[k] + when (op.kind) { + DiffOpKind.EQUAL -> { + lines.append(' ').append(op.line).append('\n') + oldCount++ + newCount++ + } + DiffOpKind.DELETE -> { + lines.append('-').append(op.line).append('\n') + oldCount++ + } + DiffOpKind.INSERT -> { + lines.append('+').append(op.line).append('\n') + newCount++ + } + } + } + appendLine("@@ -$oldStart,${oldCount.coerceAtLeast(1)} +$newStart,${newCount.coerceAtLeast(1)} @@") + append(lines) + oldLine = oldStart + oldCount + newLine = newStart + newCount + opIdx = to + 1 + } + } +} + /** * The candidate in [paths] that shares [guessed]'s filename and the most trailing path segments, * or null when nothing shares the filename. Lets a wrong package/dir prefix diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/DefaultRetryCoordinator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/DefaultRetryCoordinator.kt index b39394be..36e928bc 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/DefaultRetryCoordinator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/DefaultRetryCoordinator.kt @@ -4,6 +4,7 @@ import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.stores.EventStore import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId @@ -16,6 +17,53 @@ class DefaultRetryCoordinator( private val eventStore: EventStore, ) : RetryCoordinator { + override suspend fun decide( + sessionId: SessionId, + stageId: StageId, + gate: String, + failureReason: String, + state: OrchestrationState, + policy: RetryPolicy, + ): RetryDecision { + val fingerprint = FailureFingerprint.of(failureReason) + val prevFingerprint = state.gateFailureFingerprints[gate] + val progressed = prevFingerprint != fingerprint + val charged = !progressed + val currentCount = state.gateRetryBudgets[gate] ?: 0 + val newCount = if (charged) currentCount + 1 else currentCount + val maxAttempts = policy.maxAttemptsFor(gate) + if (charged && newCount > maxAttempts) { + return RetryDecision.Exhausted + } + + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = RetryAttemptedEvent( + sessionId = sessionId, + stageId = stageId, + attemptNumber = newCount, + maxAttempts = maxAttempts, + failureReason = failureReason, + gate = gate, + fingerprint = fingerprint, + charged = charged, + ), + ), + ) + if (policy.backoffMs > 0L) { + delay(policy.backoffMs) + } + return RetryDecision.Retry + } + override suspend fun shouldRetry( sessionId: SessionId, stageId: StageId, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/FailureFingerprint.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/FailureFingerprint.kt new file mode 100644 index 00000000..e81420bd --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/FailureFingerprint.kt @@ -0,0 +1,21 @@ +package com.correx.core.kernel.retry + +// Collapses whitespace runs and digit runs so line numbers / byte counts / timestamps embedded in a +// gate's failure reason don't make two otherwise-identical failures look like "progress". Gate +// reasons already list the failing assertion/finding ids verbatim (see runContractGate, +// runStaticAnalysis, runReviewGate in SessionOrchestrator), so this normalisation is sufficient to +// treat "same root cause" retries as no-progress without needing per-gate parsing. +private val VOLATILE_RUN = Regex("""[\s\d]+""") + +/** + * Stable fingerprint of a [StageExecutionResult.Failure] reason (design + * 2026-07-06-per-gate-retry-budgets.md §5). Two failures with the same fingerprint are treated as + * "no progress" for per-gate retry budgeting; a changed fingerprint means the run is moving and the + * retry is free. + */ +object FailureFingerprint { + fun of(reason: String): String { + val normalized = reason.trim().lowercase().replace(VOLATILE_RUN, " ").trim() + return Integer.toHexString(normalized.hashCode()) + } +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/RetryCoordinator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/RetryCoordinator.kt index 0780e2c4..9a03e76d 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/RetryCoordinator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/retry/RetryCoordinator.kt @@ -1,10 +1,21 @@ package com.correx.core.kernel.retry import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +/** Outcome of [RetryCoordinator.decide] for a single gate's failure. */ +sealed interface RetryDecision { + data object Retry : RetryDecision + data object Exhausted : RetryDecision +} + interface RetryCoordinator { + /** + * Legacy shared-budget entry point, kept for the non-gate transition-failure paths + * (Stay/NoMatch in DefaultSessionOrchestrator.step) that predate per-gate budgeting. + */ suspend fun shouldRetry( sessionId: SessionId, stageId: StageId, @@ -12,4 +23,23 @@ interface RetryCoordinator { policy: RetryPolicy, failureReason: String, ): Boolean + + /** + * Per-gate, progress-aware retry decision (design 2026-07-06-per-gate-retry-budgets.md §5). + * [failureReason] is fingerprinted internally (see [FailureFingerprint]) and charges [gate]'s + * budget only when the fingerprint is unchanged from the last attempt recorded in [state] for + * that gate (no progress); a changed fingerprint is free. On a charging decision that would + * exceed the gate's budget, returns [RetryDecision.Exhausted] without emitting an event — the + * caller (hybrid exhaustion, T6) decides what happens next. Otherwise emits + * [com.correx.core.events.events.RetryAttemptedEvent] (with the fingerprint and the full + * [failureReason] both recorded), applies backoff, and returns [RetryDecision.Retry]. + */ + suspend fun decide( + sessionId: SessionId, + stageId: StageId, + gate: String, + failureReason: String, + state: OrchestrationState, + policy: RetryPolicy, + ): RetryDecision } diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/execution/StageExecutionResult.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/execution/StageExecutionResult.kt index 3ff5199a..8a4aff6b 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/execution/StageExecutionResult.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/execution/StageExecutionResult.kt @@ -8,5 +8,9 @@ sealed interface StageExecutionResult { data class Failure( val reason: String, val retryable: Boolean, + // Budget key for per-gate retry accounting (see RetryCoordinator.decide). Default keeps + // back-compat for call sites that don't set a stable gate id (e.g. the main producer path) — + // those are budgeted under the single legacy "stage" bucket. + val gate: String = "stage", ) : StageExecutionResult } diff --git a/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt b/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt new file mode 100644 index 00000000..cc797237 --- /dev/null +++ b/testing/integration/src/test/kotlin/GateRetryBudgetExhaustionTest.kt @@ -0,0 +1,356 @@ +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.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.RetrySalvageDecidedEvent +import com.correx.core.events.events.SalvageDecision +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.ReviewOutcome +import com.correx.core.kernel.orchestration.SalvageJudgment +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import kotlinx.datetime.Instant +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.registry.ToolRegistry +import com.correx.core.tools.contract.Tool +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.filesystem.FileWriteTool +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 java.nio.file.Files +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * Integration coverage for the per-gate retry-budget + hybrid-salvage exhaustion routing + * (docs/plans/2026-07-06-per-gate-retry-budgets.md, T9). Drives a real single-stage workflow + * through [DefaultSessionOrchestrator] with a real [FileWriteTool] + [SandboxedToolExecutor] so the + * `review` gate's `stageWrittenPaths` manifest is populated from genuine [com.correx.core.events.events.FileWrittenEvent]s, + * not hand-seeded events. + */ +class GateRetryBudgetExhaustionTest { + + private val workspace = Files.createTempDirectory("gate-retry-budget-test") + + private class FixedToolCallProvider(private val toolName: String) : InferenceProvider { + override val id = ProviderId("fixed-tool-caller") + override val name = "fixed-tool-caller" + override val tokenizer: Tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + private var callCount = 0 + + override suspend fun infer(request: InferenceRequest): InferenceResponse { + callCount++ + // Odd calls issue the tool call; even calls end the round — this repeats identically + // across stage re-entries, so a retried stage goes through the same tool-call cycle. + 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 = toolName, + arguments = """{"path":"out.txt","content":"attempt-$callCount"}""", + ), + ), + ), + ) + } 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) + } + + /** Always returns a FAIL verdict with a high-confidence correctness finding — every review-gate attempt blocks. */ + private val alwaysFailingReviewer = com.correx.core.kernel.orchestration.SemanticReviewer { _, _, _, _, _ -> + ReviewOutcome( + verdict = ReviewVerdict.FAIL, + findings = listOf( + ReviewFinding( + severity = ReviewSeverity.CRITICAL, + confidence = 0.95, + category = "correctness", + target = "out.txt", + message = "off-by-one in the written content", + correctness = true, + ), + ), + ) + } + + // The stage drives a real T2 FileWriteTool; without auto-approval the orchestrator suspends on + // deferred.await() for a tool-call approval that no one resolves in a headless test. Auto-approve + // so execution actually reaches the review gate under test. + private val autoApproveEngine = 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", + ) + } + + private fun buildOrchestrator( + semanticReviewer: com.correx.core.kernel.orchestration.SemanticReviewer?, + salvageJudge: com.correx.core.kernel.orchestration.SalvageJudge?, + ): Pair { + val eventStore = InMemoryEventStore() + val tool = FileWriteTool(allowedPaths = setOf(workspace), workingDir = workspace) + val toolRegistry = SingleToolRegistry(tool) + val executor = SandboxedToolExecutor( + delegate = tool, + registry = toolRegistry, + eventDispatcher = EventDispatcher(eventStore), + workDir = workspace, + artifactStore = NoopArtifactStore(), + ) + val provider = FixedToolCallProvider(tool.name) + val inferenceRouter = object : InferenceRouter { + override suspend fun route(stageId: StageId, requiredCapabilities: Set) = provider + } + + val approvalRepository = DefaultApprovalRepository( + DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())), + ) + 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 = LiveArtifactRepositoryFor(eventStore), + approvalRepository = approvalRepository, + ) + + val engines = OrchestratorEngines( + transitionResolver = DefaultTransitionResolver { _, _ -> true }, + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = inferenceRouter, + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + approvalEngine = autoApproveEngine, + riskAssessor = DefaultRiskAssessor(), + toolExecutor = executor, + toolRegistry = toolRegistry, + workspacePolicy = WorkspacePolicy(workspace), + semanticReviewer = semanticReviewer, + salvageJudge = salvageJudge, + ) + + 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 + } + + private fun reviewGraph(): WorkflowGraph = WorkflowGraph( + id = "gate-retry-budget-test", + stages = mapOf( + StageId("A") to StageConfig(allowedTools = setOf("file_write"), semanticReview = true), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + + @Test + fun `deterministic gate exhaustion fails the workflow terminally with no salvage event`(): Unit = runBlocking { + // "produces" gate: a stage declaring artifacts it never emits fails with a stable + // fingerprint every attempt, so the budget is charged every time and exhausts deterministically. + val (orchestrator, eventStore) = buildOrchestrator(semanticReviewer = null, salvageJudge = null) + val sessionId = SessionId("gate-produces-exhaust") + val graph = WorkflowGraph( + id = "produces-exhaust", + stages = mapOf( + StageId("A") to StageConfig( + produces = listOf( + com.correx.core.artifacts.kind.TypedArtifactSlot( + name = com.correx.core.events.types.ArtifactId("never-produced"), + kind = com.correx.core.artifacts.kind.FileWrittenKind, + ), + ), + ), + ), + transitions = setOf(TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true })), + start = StageId("A"), + ) + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 2, backoffMs = 0)) + + val result = orchestrator.run(sessionId, graph, config) + + assertTrue(result is WorkflowResult.Failed) + assertTrue((result as WorkflowResult.Failed).retryExhausted) + val events = eventStore.read(sessionId) + assertTrue(events.any { it.payload is WorkflowFailedEvent }) + assertTrue(events.none { it.payload is RetrySalvageDecidedEvent }, "deterministic gates never consult the salvage judge") + } + + @Test + fun `review gate exhaustion with CONTINUE salvage resets the budget once then fails terminally on the second exhaustion`(): Unit = + runBlocking { + val salvageCalls = mutableListOf() + val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, reason -> + salvageCalls += reason + SalvageJudgment(SalvageDecision.CONTINUE, "worth one more shot") + } + val (orchestrator, eventStore) = buildOrchestrator(alwaysFailingReviewer, salvageJudge) + val sessionId = SessionId("gate-review-continue-then-fail") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0, perGateMaxAttempts = mapOf("review" to 1)), + ) + + val result = orchestrator.run(sessionId, reviewGraph(), config) + + assertTrue(result is WorkflowResult.Failed, "second exhaustion after a salvage must be terminal") + assertTrue((result as WorkflowResult.Failed).retryExhausted) + // Salvage is consulted exactly once — the second exhaustion skips the judge entirely + // because the gate is already marked salvage-used. + assertEquals(1, salvageCalls.size) + val events = eventStore.read(sessionId) + val salvageEvents = events.mapNotNull { it.payload as? RetrySalvageDecidedEvent } + assertEquals(1, salvageEvents.size) + assertEquals(SalvageDecision.CONTINUE, salvageEvents.single().decision) + assertTrue(events.any { it.payload is WorkflowFailedEvent }) + // The review gate blocked (and was retried) more than once, proving the CONTINUE reset worked. + val blockedReviews = events.mapNotNull { it.payload as? ReviewFindingsRaisedEvent }.count { it.blocked } + assertTrue(blockedReviews >= 2, "expected at least 2 blocked review attempts (pre- and post-salvage)") + } + + @Test + fun `review gate exhaustion with FAIL salvage is terminal immediately`(): Unit = runBlocking { + val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, _ -> + SalvageJudgment(SalvageDecision.FAIL, "not recoverable") + } + val (orchestrator, eventStore) = buildOrchestrator(alwaysFailingReviewer, salvageJudge) + val sessionId = SessionId("gate-review-fail") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0, perGateMaxAttempts = mapOf("review" to 1)), + ) + + val result = orchestrator.run(sessionId, reviewGraph(), config) + + assertTrue(result is WorkflowResult.Failed) + assertTrue((result as WorkflowResult.Failed).retryExhausted) + val events = eventStore.read(sessionId) + val salvageEvents = events.mapNotNull { it.payload as? RetrySalvageDecidedEvent } + assertEquals(1, salvageEvents.size) + assertEquals(SalvageDecision.FAIL, salvageEvents.single().decision) + assertTrue(events.any { it.payload is WorkflowFailedEvent }) + } + + @Test + fun `no salvage judge wired falls back to the deterministic allow-one-reset-then-fail policy`(): Unit = runBlocking { + val (orchestrator, eventStore) = buildOrchestrator(alwaysFailingReviewer, salvageJudge = null) + val sessionId = SessionId("gate-review-no-judge") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0, perGateMaxAttempts = mapOf("review" to 1)), + ) + + val result = orchestrator.run(sessionId, reviewGraph(), config) + + assertTrue(result is WorkflowResult.Failed, "even the null-judge fallback is only a one-time reset") + val events = eventStore.read(sessionId) + val salvageEvents = events.mapNotNull { it.payload as? RetrySalvageDecidedEvent } + assertEquals(1, salvageEvents.size) + assertEquals(SalvageDecision.CONTINUE, salvageEvents.single().decision) + assertTrue(events.any { it.payload is WorkflowFailedEvent }) + } +} + +private fun LiveArtifactRepositoryFor(eventStore: InMemoryEventStore) = + com.correx.infrastructure.persistence.artifact.LiveArtifactRepository(eventStore, DefaultArtifactReducer()) diff --git a/testing/kernel/src/test/kotlin/DefaultRetryCoordinatorTest.kt b/testing/kernel/src/test/kotlin/DefaultRetryCoordinatorTest.kt index 9cbe416a..4df03d02 100644 --- a/testing/kernel/src/test/kotlin/DefaultRetryCoordinatorTest.kt +++ b/testing/kernel/src/test/kotlin/DefaultRetryCoordinatorTest.kt @@ -1,9 +1,12 @@ import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.stores.EventStore import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.kernel.retry.FailureFingerprint +import com.correx.core.kernel.retry.RetryDecision import com.correx.infrastructure.persistence.InMemoryEventStore import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.test.currentTime @@ -165,4 +168,95 @@ class DefaultRetryCoordinatorTest { // virtual time should have advanced by backoffMs assertEquals(1000L, currentTime) } + + // --- decide(): per-gate, progress-aware retry decision --- + + private val sessionId = SessionId(UUID.randomUUID().toString()) + private val stageId = StageId("stage1") + + @Test + fun `decide charges the gate budget when the fingerprint is unchanged from the previous attempt`() = runTest { + val policy = RetryPolicy(maxAttempts = 3, backoffMs = 0L) + val state = OrchestrationState( + gateRetryBudgets = mapOf("contract" to 1), + gateFailureFingerprints = mapOf("contract" to FailureFingerprint.of("same-reason")), + ) + + val decision = retryCoordinator.decide( + sessionId, stageId, gate = "contract", failureReason = "same-reason", state = state, policy = policy, + ) + + assertEquals(RetryDecision.Retry, decision) + val event = eventStore.read(sessionId).single().payload as RetryAttemptedEvent + assertEquals(true, event.charged) + assertEquals(2, event.attemptNumber) + assertEquals("contract", event.gate) + } + + @Test + fun `decide is free when the fingerprint changed — progress on the run`() = runTest { + val policy = RetryPolicy(maxAttempts = 1, backoffMs = 0L) + // Budget already at maxAttempts for this gate — a charged retry would be Exhausted, but a + // changed fingerprint (progress) must still be free and retry. + val state = OrchestrationState( + gateRetryBudgets = mapOf("contract" to 1), + gateFailureFingerprints = mapOf("contract" to "old-reason"), + ) + + val decision = retryCoordinator.decide( + sessionId, stageId, gate = "contract", failureReason = "new-reason", state = state, policy = policy, + ) + + assertEquals(RetryDecision.Retry, decision) + val event = eventStore.read(sessionId).single().payload as RetryAttemptedEvent + assertEquals(false, event.charged) + assertEquals(1, event.attemptNumber) // unchanged — this retry wasn't charged + } + + @Test + fun `decide returns Exhausted without emitting an event once the gate budget is spent`() = runTest { + val policy = RetryPolicy(maxAttempts = 2, backoffMs = 0L) + val state = OrchestrationState( + gateRetryBudgets = mapOf("contract" to 2), + gateFailureFingerprints = mapOf("contract" to FailureFingerprint.of("same-reason")), + ) + + val decision = retryCoordinator.decide( + sessionId, stageId, gate = "contract", failureReason = "same-reason", state = state, policy = policy, + ) + + assertEquals(RetryDecision.Exhausted, decision) + Assertions.assertTrue(eventStore.read(sessionId).isEmpty()) + } + + @Test + fun `decide keys the budget by gate id — a different gate has its own fresh budget`() = runTest { + val policy = RetryPolicy(maxAttempts = 1, backoffMs = 0L) + val state = OrchestrationState( + gateRetryBudgets = mapOf("contract" to 1), + gateFailureFingerprints = mapOf("contract" to FailureFingerprint.of("same-reason")), + ) + + // "contract" is already exhausted at maxAttempts=1, but "review" has never charged. + val decision = retryCoordinator.decide( + sessionId, stageId, gate = "review", failureReason = "some finding", state = state, policy = policy, + ) + + assertEquals(RetryDecision.Retry, decision) + } + + @Test + fun `decide honours perGateMaxAttempts override`() = runTest { + val policy = RetryPolicy(maxAttempts = 1, backoffMs = 0L, perGateMaxAttempts = mapOf("review" to 5)) + val state = OrchestrationState( + gateRetryBudgets = mapOf("review" to 4), + gateFailureFingerprints = mapOf("review" to "same-reason"), + ) + + val decision = retryCoordinator.decide( + sessionId, stageId, gate = "review", failureReason = "same-reason", state = state, policy = policy, + ) + + assertEquals(RetryDecision.Retry, decision) + } } diff --git a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt index a59cb78a..7870fe63 100644 --- a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt +++ b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt @@ -1,6 +1,8 @@ import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent 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.TransitionExecutedEvent import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowFailedEvent @@ -153,6 +155,141 @@ class OrchestrationReducerTest { } + @Test + fun `RetryAttemptedEvent with charged=true bumps the gate's per-gate budget and fingerprint`() { + val retried = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "contract", fingerprint = "fp1", charged = true, + ), + ), + ) + assertEquals(1, retried.gateRetryBudgets["contract"]) + assertEquals("fp1", retried.gateFailureFingerprints["contract"]) + } + + @Test + fun `RetryAttemptedEvent with charged=false leaves the gate's budget untouched but records the fingerprint`() { + val charged = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "contract", fingerprint = "fp1", charged = true, + ), + ), + ) + val free = reducer.reduce( + charged, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "different problem", + gate = "contract", fingerprint = "fp2", charged = false, + ), + ), + ) + assertEquals(1, free.gateRetryBudgets["contract"]) + assertEquals("fp2", free.gateFailureFingerprints["contract"]) + } + + @Test + fun `gate budgets are independent per gate id`() { + val contractCharged = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "contract", fingerprint = "fp1", charged = true, + ), + ), + ) + val reviewCharged = reducer.reduce( + contractCharged, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom2", + gate = "review", fingerprint = "fpr1", charged = true, + ), + ), + ) + assertEquals(1, reviewCharged.gateRetryBudgets["contract"]) + assertEquals(1, reviewCharged.gateRetryBudgets["review"]) + } + + @Test + fun `TransitionExecutedEvent resets per-gate budgets, fingerprints, and salvage-used set`() { + val charged = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "review", fingerprint = "fp1", charged = true, + ), + ), + ) + val salvaged = reducer.reduce( + charged, + stored( + payload = RetrySalvageDecidedEvent(sessionId, stageId, "review", SalvageDecision.CONTINUE, "fixable"), + ), + ) + assertTrue(salvaged.gateSalvageUsed.contains("review")) + + val nextStage = StageId("stage-2") + val advanced = reducer.reduce( + salvaged, + stored(payload = TransitionExecutedEvent(sessionId, stageId, nextStage, TransitionId("t1"))), + ) + assertTrue(advanced.gateRetryBudgets.isEmpty()) + assertTrue(advanced.gateFailureFingerprints.isEmpty()) + assertTrue(advanced.gateSalvageUsed.isEmpty()) + } + + @Test + fun `RetrySalvageDecidedEvent CONTINUE resets the gate's budget and marks salvage used`() { + val charged = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "review", fingerprint = "fp1", charged = true, + ), + ), + ) + assertEquals(1, charged.gateRetryBudgets["review"]) + + val salvaged = reducer.reduce( + charged, + stored( + payload = RetrySalvageDecidedEvent(sessionId, stageId, "review", SalvageDecision.CONTINUE, "fixable"), + ), + ) + assertEquals(0, salvaged.gateRetryBudgets["review"]) + assertTrue(salvaged.gateSalvageUsed.contains("review")) + } + + @Test + fun `RetrySalvageDecidedEvent FAIL does not touch state`() { + val charged = reducer.reduce( + state, + stored( + payload = RetryAttemptedEvent( + sessionId, stageId, 1, 3, "boom", + gate = "review", fingerprint = "fp1", charged = true, + ), + ), + ) + val decided = reducer.reduce( + charged, + stored( + payload = RetrySalvageDecidedEvent(sessionId, stageId, "review", SalvageDecision.FAIL, "not fixable"), + ), + ) + assertEquals(charged, decided) + } + @Test fun `unrelated event does nothing`() { val started = reducer.reduce(