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.
This commit is contained in:
@@ -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<CapabilityScore> = 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<Tool> = 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<ApprovalGrant>,
|
||||
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<DefaultSessionOrchestrator, InMemoryEventStore> {
|
||||
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<ModelCapability>) = provider
|
||||
}
|
||||
|
||||
val approvalRepository = DefaultApprovalRepository(
|
||||
DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())),
|
||||
)
|
||||
val repositories = OrchestratorRepositories(
|
||||
eventStore = eventStore,
|
||||
inferenceRepository = InferenceRepository(object : EventReplayer<InferenceState> {
|
||||
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<String>()
|
||||
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())
|
||||
Reference in New Issue
Block a user