feat(recovery): failure-ticket routing + review-gate RECOVER + return-to-sender
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.
This commit is contained in:
@@ -14,6 +14,7 @@ import com.correx.core.events.events.ReviewFinding
|
||||
import com.correx.core.events.events.ReviewFindingsRaisedEvent
|
||||
import com.correx.core.events.events.ReviewSeverity
|
||||
import com.correx.core.events.events.ReviewVerdict
|
||||
import com.correx.core.events.events.FailureTicketOpenedEvent
|
||||
import com.correx.core.events.events.RetrySalvageDecidedEvent
|
||||
import com.correx.core.events.events.SalvageDecision
|
||||
import com.correx.core.events.events.WorkflowFailedEvent
|
||||
@@ -311,6 +312,61 @@ class GateRetryBudgetExhaustionTest {
|
||||
assertTrue(blockedReviews >= 2, "expected at least 2 blocked review attempts (pre- and post-salvage)")
|
||||
}
|
||||
|
||||
// Stage A opts into the (always-failing) review gate; a sibling recovery stage holds file_write.
|
||||
// Recovery has NO outgoing edge — the kernel returns it to the ticket origin (A) dynamically.
|
||||
private fun recoveryReviewGraph(): WorkflowGraph = WorkflowGraph(
|
||||
id = "gate-review-recover-test",
|
||||
stages = mapOf(
|
||||
StageId("A") to StageConfig(allowedTools = setOf("file_write"), semanticReview = true),
|
||||
StageId("recovery") to StageConfig(
|
||||
allowedTools = setOf("file_write"),
|
||||
metadata = mapOf("role" to "recovery"),
|
||||
),
|
||||
),
|
||||
transitions = setOf(
|
||||
TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }),
|
||||
),
|
||||
start = StageId("A"),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `review gate exhaustion with RECOVER salvage routes to the recovery stage and exhausts the route budget`(): Unit =
|
||||
runBlocking {
|
||||
// Slice 2: the salvage judge chooses RECOVER instead of CONTINUE/FAIL, so a review-gate
|
||||
// exhaustion hands off to the recovery stage (same destination as the deterministic agency
|
||||
// guard) rather than retrying A in place or failing outright. Bounded by RECOVERY_ROUTE_BUDGET.
|
||||
val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, _ ->
|
||||
SalvageJudgment(SalvageDecision.RECOVER, "hand this to recovery")
|
||||
}
|
||||
val (orchestrator, eventStore) = buildOrchestrator(alwaysFailingReviewer, salvageJudge)
|
||||
val sessionId = SessionId("gate-review-recover")
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0, perGateMaxAttempts = mapOf("review" to 1)),
|
||||
)
|
||||
|
||||
val result = orchestrator.run(sessionId, recoveryReviewGraph(), config)
|
||||
|
||||
assertTrue(result is WorkflowResult.Failed, "route budget must exhaust terminally")
|
||||
assertTrue((result as WorkflowResult.Failed).retryExhausted)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val tickets = events.mapNotNull { it.payload as? FailureTicketOpenedEvent }
|
||||
assertEquals(2, tickets.size, "one ticket per route, bounded by RECOVERY_ROUTE_BUDGET")
|
||||
tickets.forEach {
|
||||
assertEquals("A", it.stageId.value)
|
||||
assertEquals("review", it.gate)
|
||||
assertEquals("implementation", it.category)
|
||||
assertEquals("file_write", it.requiredCapability)
|
||||
assertEquals("recovery", it.routeTo.value)
|
||||
}
|
||||
assertEquals(listOf(1, 2), tickets.map { it.routeAttempt })
|
||||
// Every salvage decision on this path is RECOVER (judge consulted afresh each re-entry
|
||||
// because TransitionExecuted resets gateSalvageUsed).
|
||||
val salvageDecisions = events.mapNotNull { it.payload as? RetrySalvageDecidedEvent }
|
||||
assertTrue(salvageDecisions.size >= 2, "judge consulted on each review exhaustion")
|
||||
assertTrue(salvageDecisions.all { it.decision == SalvageDecision.RECOVER })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `review gate exhaustion with FAIL salvage is terminal immediately`(): Unit = runBlocking {
|
||||
val salvageJudge = com.correx.core.kernel.orchestration.SalvageJudge { _, _, _ ->
|
||||
|
||||
@@ -0,0 +1,257 @@
|
||||
import com.correx.core.approvals.ApprovalOutcome
|
||||
import com.correx.core.approvals.ApprovalProjector
|
||||
import com.correx.core.approvals.ApprovalStatus
|
||||
import com.correx.core.approvals.DefaultApprovalReducer
|
||||
import com.correx.core.approvals.DefaultApprovalRepository
|
||||
import com.correx.core.approvals.domain.ApprovalEngine
|
||||
import com.correx.core.approvals.model.ApprovalContext
|
||||
import com.correx.core.approvals.model.ApprovalDecision
|
||||
import com.correx.core.approvals.model.ApprovalGrant
|
||||
import com.correx.core.approvals.model.DomainApprovalRequest
|
||||
import com.correx.core.artifacts.DefaultArtifactReducer
|
||||
import com.correx.core.events.EventDispatcher
|
||||
import com.correx.core.events.events.FailureTicketOpenedEvent
|
||||
import com.correx.core.events.events.WorkflowFailedEvent
|
||||
import com.correx.core.events.execution.RetryPolicy
|
||||
import com.correx.core.events.types.ProviderId
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.StageId
|
||||
import com.correx.core.events.types.TransitionId
|
||||
import com.correx.core.inference.CapabilityScore
|
||||
import com.correx.core.inference.FinishReason
|
||||
import com.correx.core.inference.InferenceProvider
|
||||
import com.correx.core.inference.InferenceRepository
|
||||
import com.correx.core.inference.InferenceRequest
|
||||
import com.correx.core.inference.InferenceResponse
|
||||
import com.correx.core.inference.InferenceRouter
|
||||
import com.correx.core.inference.InferenceState
|
||||
import com.correx.core.inference.ModelCapability
|
||||
import com.correx.core.inference.ProviderHealth
|
||||
import com.correx.core.inference.Tokenizer
|
||||
import com.correx.core.inference.ToolCallFunction
|
||||
import com.correx.core.inference.ToolCallRequest
|
||||
import com.correx.core.inference.TokenUsage
|
||||
import com.correx.core.journal.DecisionJournalProjector
|
||||
import com.correx.core.journal.DefaultDecisionJournalReducer
|
||||
import com.correx.core.journal.DefaultDecisionJournalRepository
|
||||
import com.correx.core.kernel.execution.WorkflowResult
|
||||
import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
|
||||
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
||||
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
||||
import com.correx.core.kernel.orchestration.OrchestrationProjector
|
||||
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
||||
import com.correx.core.kernel.orchestration.OrchestratorEngines
|
||||
import com.correx.core.kernel.orchestration.OrchestratorRepositories
|
||||
import com.correx.core.kernel.orchestration.StaticAnalysisRunResult
|
||||
import com.correx.core.kernel.orchestration.StaticAnalysisRunner
|
||||
import com.correx.core.kernel.retry.DefaultRetryCoordinator
|
||||
import com.correx.core.risk.DefaultRiskAssessor
|
||||
import com.correx.core.sessions.DefaultSessionReducer
|
||||
import com.correx.core.sessions.DefaultSessionRepository
|
||||
import com.correx.core.sessions.SessionProjector
|
||||
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
|
||||
import com.correx.core.sessions.projections.replay.EventReplayer
|
||||
import com.correx.core.toolintent.WorkspacePolicy
|
||||
import com.correx.core.tools.contract.Tool
|
||||
import com.correx.core.tools.registry.ToolRegistry
|
||||
import com.correx.core.transitions.graph.StageConfig
|
||||
import com.correx.core.transitions.graph.TransitionEdge
|
||||
import com.correx.core.transitions.graph.WorkflowGraph
|
||||
import com.correx.core.transitions.resolution.DefaultTransitionResolver
|
||||
import com.correx.core.validation.pipeline.ValidationPipeline
|
||||
import com.correx.infrastructure.persistence.InMemoryEventStore
|
||||
import com.correx.infrastructure.tools.SandboxedToolExecutor
|
||||
import com.correx.infrastructure.tools.shell.ShellTool
|
||||
import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore
|
||||
import com.correx.testing.fixtures.context.ContextFixtures
|
||||
import com.correx.testing.fixtures.cyclePolicyMissingValidator
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.datetime.Instant
|
||||
import java.nio.file.Files
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
/**
|
||||
* Retry-agency / recovery-routing coverage. A write-less `verify` stage runs `shell true` (so it
|
||||
* passes the producer gate, exactly as #41's `final_verification` ran shell) then fails a
|
||||
* `static_analysis` gate (fake runner always non-clean). Because `verify`'s allowedTools lack the
|
||||
* `file_write` capability that gate requires, the orchestrator must NOT retry it in place (futile) —
|
||||
* it opens a [FailureTicketOpenedEvent] and routes to the `recovery` stage (marked
|
||||
* `metadata["role"]="recovery"`), which loops back to `verify`. After RECOVERY_ROUTE_BUDGET routes
|
||||
* the run fails terminally.
|
||||
*/
|
||||
class RecoveryRoutingTest {
|
||||
|
||||
private val workspace = Files.createTempDirectory("recovery-routing-test")
|
||||
|
||||
/** Odd calls issue a `shell true` tool call; even calls end the round — repeats across re-entries. */
|
||||
private class ShellThenStopProvider : InferenceProvider {
|
||||
override val id = ProviderId("shell-caller")
|
||||
override val name = "shell-caller"
|
||||
override val tokenizer: Tokenizer = com.correx.testing.fixtures.inference.MockTokenizer()
|
||||
private var callCount = 0
|
||||
override suspend fun infer(request: InferenceRequest): InferenceResponse {
|
||||
callCount++
|
||||
return if (callCount % 2 == 1) {
|
||||
InferenceResponse(
|
||||
requestId = request.requestId,
|
||||
text = "",
|
||||
finishReason = FinishReason.ToolCall,
|
||||
tokensUsed = TokenUsage(10, 5),
|
||||
latencyMs = 0,
|
||||
toolCalls = listOf(
|
||||
ToolCallRequest(
|
||||
id = "tc-$callCount",
|
||||
function = ToolCallFunction(name = "shell", arguments = """{"argv":["true"]}"""),
|
||||
),
|
||||
),
|
||||
)
|
||||
} else {
|
||||
InferenceResponse(
|
||||
requestId = request.requestId,
|
||||
text = "done",
|
||||
finishReason = FinishReason.Stop,
|
||||
tokensUsed = TokenUsage(10, 5),
|
||||
latencyMs = 0,
|
||||
)
|
||||
}
|
||||
}
|
||||
override suspend fun healthCheck(): ProviderHealth = ProviderHealth.Healthy
|
||||
override fun capabilities(): Set<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)
|
||||
}
|
||||
|
||||
private fun buildOrchestrator(): Pair<DefaultSessionOrchestrator, InMemoryEventStore> {
|
||||
val eventStore = InMemoryEventStore()
|
||||
val shell = ShellTool(allowedExecutables = setOf("true"), workingDir = workspace)
|
||||
val toolRegistry = SingleToolRegistry(shell)
|
||||
val executor = SandboxedToolExecutor(
|
||||
delegate = shell,
|
||||
registry = toolRegistry,
|
||||
eventDispatcher = EventDispatcher(eventStore),
|
||||
workDir = workspace,
|
||||
artifactStore = NoopArtifactStore(),
|
||||
)
|
||||
val provider = ShellThenStopProvider()
|
||||
val inferenceRouter = object : InferenceRouter {
|
||||
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>) = provider
|
||||
}
|
||||
val failingStaticAnalysis =
|
||||
StaticAnalysisRunner { _, _ -> StaticAnalysisRunResult(exitCode = 1, output = "TS5023: bad option") }
|
||||
|
||||
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 = com.correx.infrastructure.persistence.artifact.LiveArtifactRepository(
|
||||
eventStore, DefaultArtifactReducer(),
|
||||
),
|
||||
approvalRepository = DefaultApprovalRepository(
|
||||
DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())),
|
||||
),
|
||||
)
|
||||
|
||||
val engines = OrchestratorEngines(
|
||||
transitionResolver = DefaultTransitionResolver { _, _ -> true },
|
||||
contextPackBuilder = ContextFixtures.simpleBuilder(),
|
||||
inferenceRouter = inferenceRouter,
|
||||
validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())),
|
||||
riskAssessor = DefaultRiskAssessor(),
|
||||
toolExecutor = executor,
|
||||
toolRegistry = toolRegistry,
|
||||
workspacePolicy = WorkspacePolicy(workspace),
|
||||
staticAnalysisRunner = failingStaticAnalysis,
|
||||
approvalEngine = object : ApprovalEngine {
|
||||
override fun evaluate(
|
||||
request: DomainApprovalRequest,
|
||||
context: ApprovalContext,
|
||||
grants: List<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",
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
val decisionJournalRepository = DefaultDecisionJournalRepository(
|
||||
DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())),
|
||||
)
|
||||
val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines,
|
||||
retryCoordinator = DefaultRetryCoordinator(eventStore),
|
||||
artifactStore = NoopArtifactStore(),
|
||||
decisionJournalRepository = decisionJournalRepository,
|
||||
)
|
||||
return orchestrator to eventStore
|
||||
}
|
||||
|
||||
// verify (write-less, failing static gate) --t-verify--> done
|
||||
// recovery (role=recovery, holds file_write) has NO outgoing edge on purpose: the kernel returns
|
||||
// it to the ticket's origin stage (recoveryReturnMove), so the recovery->verify loop is driven
|
||||
// dynamically. If the return-to-sender path regresses, recovery dead-ends and the route count is
|
||||
// wrong — this graph would fail rather than silently pass on a stale static edge.
|
||||
private fun recoveryGraph(): WorkflowGraph = WorkflowGraph(
|
||||
id = "recovery-routing-test",
|
||||
stages = mapOf(
|
||||
StageId("verify") to StageConfig(
|
||||
allowedTools = setOf("shell"),
|
||||
staticAnalysis = listOf("typecheck"),
|
||||
),
|
||||
StageId("recovery") to StageConfig(
|
||||
allowedTools = setOf("shell", "file_write"),
|
||||
metadata = mapOf("role" to "recovery"),
|
||||
),
|
||||
),
|
||||
transitions = setOf(
|
||||
TransitionEdge(TransitionId("t-verify"), StageId("verify"), StageId("done"), condition = { true }),
|
||||
),
|
||||
start = StageId("verify"),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `write-less gate failure routes to recovery and exhausts its own route budget`(): Unit = runBlocking {
|
||||
val (orchestrator, eventStore) = buildOrchestrator()
|
||||
val sessionId = SessionId("recovery-route")
|
||||
// Generous in-place per-gate budget proves routing pre-empts it: without the agency guard,
|
||||
// static_analysis would retry verify in place (never able to change the outcome) instead.
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 10, backoffMs = 0))
|
||||
|
||||
val result = orchestrator.run(sessionId, recoveryGraph(), config)
|
||||
|
||||
assertTrue(result is WorkflowResult.Failed, "route budget must exhaust terminally")
|
||||
assertTrue((result as WorkflowResult.Failed).retryExhausted)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val tickets = events.mapNotNull { it.payload as? FailureTicketOpenedEvent }
|
||||
assertEquals(2, tickets.size, "one ticket per route, bounded by RECOVERY_ROUTE_BUDGET")
|
||||
tickets.forEach {
|
||||
assertEquals("verify", it.stageId.value)
|
||||
assertEquals("static_analysis", it.gate)
|
||||
assertEquals("implementation", it.category)
|
||||
assertEquals("file_write", it.requiredCapability)
|
||||
assertEquals("recovery", it.routeTo.value)
|
||||
}
|
||||
assertEquals(listOf(1, 2), tickets.map { it.routeAttempt })
|
||||
assertTrue(events.any { it.payload is WorkflowFailedEvent })
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user