kami
879672a47d
Retry-agency invariant: a stage may only retry a gate it has the capability to change. A write-less stage failing a build/contract/static gate (e.g. freestyle final_verification with allowedTools=[shell]) can never fix it, so retrying in place is futile until the budget drains (Vikunja #41). Instead: open a FailureTicketOpenedEvent (category + requiredCapability derived deterministically from the gate id, no LLM) and route to a recovery stage that holds the capability, bounded by a small per-stage route budget. - Slice 2: SalvageDecision.RECOVER (ternary CONTINUE/RECOVER/FAIL) lets the review-gate salvage judge hand off to recovery. Shared routeToRecovery() unifies the deterministic agency guard and the judge on one destination; decideGateExhaustion now returns StepResult?. - Return-to-sender: recovery's exit is dynamic (recoveryReturnMove) — it goes back to the exact ticket-origin stage to re-run its gate, so a write-less gate anywhere in the graph is handled without skipping intervening stages. The synthesized recovery->terminal edge is now only a no-ticket fallback. - Freestyle wiring: ExecutionPlanCompiler injectRecovery flag (Main passes true) synthesizes a write-capable recovery stage; ticket evidence reaches it via buildRecoveryTicketEntry. - Read-only tools always present: StageConfig.effectiveAllowedTools adds file_read/list_dir to any tool-granting stage (fixes the verifier reaching for `shell ls -R` to inspect the tree). Tests: RecoveryRoutingTest (deterministic gate, no static return edge — proves dynamic return) + GateRetryBudgetExhaustionTest RECOVER case + two compiler tests. All green; detekt clean.
413 lines
21 KiB
Kotlin
413 lines
21 KiB
Kotlin
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.FailureTicketOpenedEvent
|
|
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)")
|
|
}
|
|
|
|
// 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 { _, _, _ ->
|
|
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())
|