feat(recovery,context): tier-2 intent-holder arbiter + remaining-delta pinning + surfaced signal events
- recovery: two-tier repair ladder — owner then arbiter (recovery stage recast
as intent-holder, holds initial intent, reconciles cross-file contract disputes)
with independent INTENT_ROUTE_BUDGET; RecoveryRoutingTest coverage
- context: remaining-delta checklist pinning (RemainingDelta{Pinning,Entry}Test)
- surface write-only events (session name, quality signals) into stats/browse
- parseToolArguments lenient-Json fallback for bare-key drift
- tools: ChildProcess seam
This commit is contained in:
@@ -53,7 +53,11 @@ 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.contract.ToolExecutor
|
||||
import com.correx.core.tools.contract.ToolResult
|
||||
import com.correx.core.events.events.ToolRequest
|
||||
import com.correx.core.tools.registry.ToolRegistry
|
||||
import com.correx.infrastructure.tools.filesystem.FileWriteTool
|
||||
import com.correx.core.transitions.graph.StageConfig
|
||||
import com.correx.core.transitions.graph.TransitionEdge
|
||||
import com.correx.core.transitions.graph.WorkflowGraph
|
||||
@@ -282,7 +286,282 @@ class RecoveryRoutingTest {
|
||||
assertTrue(tickets.map { it.fingerprint }.toSet().size == tickets.size, "each route recorded a distinct fingerprint")
|
||||
}
|
||||
|
||||
/** Emits a tool call on odd infers, ends the round on even — but the tool depends on the stage:
|
||||
* `impl` writes a file (so it becomes the file's author), any other stage runs `shell true`. */
|
||||
private class StageRoutingProvider : InferenceProvider {
|
||||
override val id = ProviderId("stage-router")
|
||||
override val name = "stage-router"
|
||||
override val tokenizer: Tokenizer = com.correx.testing.fixtures.inference.MockTokenizer()
|
||||
private val counts = mutableMapOf<String, Int>()
|
||||
override suspend fun infer(request: InferenceRequest): InferenceResponse {
|
||||
val stage = request.stageId.value
|
||||
val n = (counts[stage] ?: 0) + 1
|
||||
counts[stage] = n
|
||||
if (n % 2 == 0) {
|
||||
return InferenceResponse(
|
||||
requestId = request.requestId, text = "done", finishReason = FinishReason.Stop,
|
||||
tokensUsed = TokenUsage(10, 5), latencyMs = 0,
|
||||
)
|
||||
}
|
||||
val (tool, args) = if (stage == "impl") {
|
||||
"file_write" to """{"path":"App.tsx","content":"export const x = 1;\n"}"""
|
||||
} else {
|
||||
"shell" to """{"argv":["true"]}"""
|
||||
}
|
||||
return InferenceResponse(
|
||||
requestId = request.requestId, text = "", finishReason = FinishReason.ToolCall,
|
||||
tokensUsed = TokenUsage(10, 5), latencyMs = 0,
|
||||
toolCalls = listOf(
|
||||
ToolCallRequest(id = "tc-$stage-$n", function = ToolCallFunction(name = tool, arguments = args)),
|
||||
),
|
||||
)
|
||||
}
|
||||
override suspend fun healthCheck(): ProviderHealth = ProviderHealth.Healthy
|
||||
override fun capabilities(): Set<CapabilityScore> = setOf(CapabilityScore(ModelCapability.General, 1.0))
|
||||
}
|
||||
|
||||
/** Like [StageRoutingProvider], but records the context-entry sourceTypes it saw on each `impl`
|
||||
* inference — so a test can assert whether impl's own prompt (agentPrompt) was present. */
|
||||
private class RecordingRoutingProvider : InferenceProvider {
|
||||
override val id = ProviderId("recording-router")
|
||||
override val name = "recording-router"
|
||||
override val tokenizer: Tokenizer = com.correx.testing.fixtures.inference.MockTokenizer()
|
||||
val implSourceTypes = mutableListOf<Set<String>>()
|
||||
private val counts = mutableMapOf<String, Int>()
|
||||
override suspend fun infer(request: InferenceRequest): InferenceResponse {
|
||||
val stage = request.stageId.value
|
||||
if (stage == "impl") {
|
||||
implSourceTypes.add(request.contextPack.layers.values.flatten().map { it.sourceType }.toSet())
|
||||
}
|
||||
val n = (counts[stage] ?: 0) + 1
|
||||
counts[stage] = n
|
||||
if (n % 2 == 0) {
|
||||
return InferenceResponse(
|
||||
requestId = request.requestId, text = "done", finishReason = FinishReason.Stop,
|
||||
tokensUsed = TokenUsage(10, 5), latencyMs = 0,
|
||||
)
|
||||
}
|
||||
val (tool, args) = if (stage == "impl") {
|
||||
"file_write" to """{"path":"App.tsx","content":"export const x = 1;\n"}"""
|
||||
} else {
|
||||
"shell" to """{"argv":["true"]}"""
|
||||
}
|
||||
return InferenceResponse(
|
||||
requestId = request.requestId, text = "", finishReason = FinishReason.ToolCall,
|
||||
tokensUsed = TokenUsage(10, 5), latencyMs = 0,
|
||||
toolCalls = listOf(
|
||||
ToolCallRequest(id = "tc-$stage-$n", function = ToolCallFunction(name = tool, arguments = args)),
|
||||
),
|
||||
)
|
||||
}
|
||||
override suspend fun healthCheck(): ProviderHealth = ProviderHealth.Healthy
|
||||
override fun capabilities(): Set<CapabilityScore> = setOf(CapabilityScore(ModelCapability.General, 1.0))
|
||||
}
|
||||
|
||||
private class MultiToolRegistry(private val tools: List<Tool>) : ToolRegistry {
|
||||
override fun resolve(name: String): Tool? = tools.firstOrNull { it.name == name }
|
||||
override fun all(): List<Tool> = tools
|
||||
}
|
||||
|
||||
// impl (file_write) --t-impl--> verify (write-less, failing static gate naming impl's file) --> done
|
||||
// No `recovery` stage exists: the ticket must route to the file's AUTHOR (impl) via route-to-owner,
|
||||
// not to a generic recovery stage. impl repairs and returns to verify (ticketReturnMove).
|
||||
private fun ownerGraph(): WorkflowGraph = WorkflowGraph(
|
||||
id = "route-to-owner-test",
|
||||
stages = mapOf(
|
||||
StageId("impl") to StageConfig(allowedTools = setOf("file_write")),
|
||||
StageId("verify") to StageConfig(allowedTools = setOf("shell"), staticAnalysis = listOf("typecheck")),
|
||||
),
|
||||
transitions = setOf(
|
||||
TransitionEdge(TransitionId("t-impl"), StageId("impl"), StageId("verify"), condition = { true }),
|
||||
TransitionEdge(TransitionId("t-verify"), StageId("verify"), StageId("done"), condition = { true }),
|
||||
),
|
||||
start = StageId("impl"),
|
||||
)
|
||||
|
||||
// impl (file_write) + verify + a recovery stage (role=recovery, holds file_write). Tier 1 routes to
|
||||
// the author (impl); when its budget exhausts, tier 2 escalates to the arbiter (recovery).
|
||||
private fun ownerAndArbiterGraph(): WorkflowGraph = WorkflowGraph(
|
||||
id = "route-to-owner-escalation-test",
|
||||
stages = mapOf(
|
||||
StageId("impl") to StageConfig(allowedTools = setOf("file_write")),
|
||||
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-impl"), StageId("impl"), StageId("verify"), condition = { true }),
|
||||
TransitionEdge(TransitionId("t-verify"), StageId("verify"), StageId("done"), condition = { true }),
|
||||
),
|
||||
start = StageId("impl"),
|
||||
)
|
||||
|
||||
/** Builds an orchestrator whose only writing stage is `impl` (via file_write) and whose `verify`
|
||||
* static gate fails with [staticOutput] — so route-to-owner can resolve `impl` as the file author. */
|
||||
private fun buildMultiToolOrchestrator(
|
||||
staticOutput: String = "App.tsx: TS2322 type mismatch",
|
||||
provider: InferenceProvider = StageRoutingProvider(),
|
||||
): Pair<DefaultSessionOrchestrator, InMemoryEventStore> {
|
||||
val eventStore = InMemoryEventStore()
|
||||
val shell = ShellTool(allowedExecutables = setOf("true"), workingDir = workspace)
|
||||
val fileWrite = FileWriteTool(allowedPaths = setOf(workspace), workingDir = workspace)
|
||||
val toolRegistry = MultiToolRegistry(listOf(shell, fileWrite))
|
||||
// Dispatch delegate: SandboxedToolExecutor resolves the tool from the registry, then hands the
|
||||
// request to this delegate — which routes to the matching tool by name.
|
||||
val dispatch = object : ToolExecutor {
|
||||
override suspend fun execute(request: ToolRequest): ToolResult =
|
||||
(toolRegistry.resolve(request.toolName) as ToolExecutor).execute(request)
|
||||
}
|
||||
val executor = SandboxedToolExecutor(
|
||||
delegate = dispatch, registry = toolRegistry, eventDispatcher = EventDispatcher(eventStore),
|
||||
workDir = workspace, artifactStore = NoopArtifactStore(),
|
||||
)
|
||||
val inferenceRouter = object : InferenceRouter {
|
||||
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>) = provider
|
||||
}
|
||||
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),
|
||||
// Gate names the file impl wrote, so route-to-owner can resolve the author from the manifest.
|
||||
staticAnalysisRunner = StaticAnalysisRunner { _, _ ->
|
||||
StaticAnalysisRunResult(exitCode = 1, output = staticOutput)
|
||||
},
|
||||
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 orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines,
|
||||
retryCoordinator = DefaultRetryCoordinator(eventStore),
|
||||
artifactStore = NoopArtifactStore(),
|
||||
decisionJournalRepository = DefaultDecisionJournalRepository(
|
||||
DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())),
|
||||
),
|
||||
)
|
||||
return orchestrator to eventStore
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `write-less gate failure routes the ticket to the file's author (route-to-owner)`(): Unit = runBlocking {
|
||||
val (orchestrator, eventStore) = buildMultiToolOrchestrator()
|
||||
val sessionId = SessionId("route-to-owner")
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 10, backoffMs = 0))
|
||||
|
||||
val result = orchestrator.run(sessionId, ownerGraph(), config)
|
||||
|
||||
assertTrue(result is WorkflowResult.Failed, "route budget still exhausts terminally")
|
||||
val tickets = eventStore.read(sessionId).mapNotNull { it.payload as? FailureTicketOpenedEvent }
|
||||
assertTrue(tickets.isNotEmpty(), "verify's write-less gate failure must open a ticket")
|
||||
tickets.forEach {
|
||||
assertEquals("verify", it.stageId.value, "the gate that failed is verify")
|
||||
assertEquals("impl", it.routeTo.value, "ticket routes to the file's author, not a recovery stage")
|
||||
assertEquals("file_write", it.requiredCapability)
|
||||
assertTrue(!it.escalated, "no recovery stage exists, so all routes stay tier-1 (owner)")
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `owner budget exhaustion escalates the ticket to the arbiter (intent-holder)`(): Unit = runBlocking {
|
||||
// Same constant failure every round, so neither tier makes progress: the owner (impl) burns its
|
||||
// RECOVERY_ROUTE_BUDGET, THEN the ticket escalates to the arbiter (recovery) for INTENT_ROUTE_BUDGET
|
||||
// more routes, THEN terminal. Proves the two-tier ladder: author first, intent-holder as escalation.
|
||||
val (orchestrator, eventStore) = buildMultiToolOrchestrator()
|
||||
val sessionId = SessionId("owner-escalation")
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 20, backoffMs = 0))
|
||||
|
||||
val result = orchestrator.run(sessionId, ownerAndArbiterGraph(), config)
|
||||
|
||||
assertTrue(result is WorkflowResult.Failed, "ladder still exhausts terminally")
|
||||
val tickets = eventStore.read(sessionId).mapNotNull { it.payload as? FailureTicketOpenedEvent }
|
||||
val tier1 = tickets.filter { !it.escalated }
|
||||
val tier2 = tickets.filter { it.escalated }
|
||||
assertEquals(RECOVERY_ROUTE_BUDGET_TEST, tier1.size, "tier-1 spends the owner budget first")
|
||||
tier1.forEach { assertEquals("impl", it.routeTo.value, "tier-1 routes to the file's author") }
|
||||
assertEquals(INTENT_ROUTE_BUDGET_TEST, tier2.size, "then tier-2 spends the arbiter budget")
|
||||
tier2.forEach { assertEquals("recovery", it.routeTo.value, "tier-2 escalates to the arbiter") }
|
||||
// ordering: every tier-1 ticket precedes every tier-2 ticket (escalation only after owner exhaust)
|
||||
val firstEscalatedIdx = tickets.indexOfFirst { it.escalated }
|
||||
assertTrue(tickets.take(firstEscalatedIdx).none { it.escalated }, "owner routes come before escalation")
|
||||
}
|
||||
|
||||
// impl carries a generative "scaffold" mandate as its own prompt. On the FORWARD visit it applies;
|
||||
// when the ticket routes impl back to repair its own file, that mandate must be suppressed (else it
|
||||
// re-scaffolds and stubs the real file). Same topology as ownerGraph otherwise.
|
||||
private fun ownerGraphWithMandate(): WorkflowGraph = WorkflowGraph(
|
||||
id = "owner-mandate-suppression-test",
|
||||
stages = mapOf(
|
||||
StageId("impl") to StageConfig(
|
||||
allowedTools = setOf("file_write"),
|
||||
metadata = mapOf("promptInline" to "Scaffold the entire project by creating placeholder files."),
|
||||
),
|
||||
StageId("verify") to StageConfig(allowedTools = setOf("shell"), staticAnalysis = listOf("typecheck")),
|
||||
),
|
||||
transitions = setOf(
|
||||
TransitionEdge(TransitionId("t-impl"), StageId("impl"), StageId("verify"), condition = { true }),
|
||||
TransitionEdge(TransitionId("t-verify"), StageId("verify"), StageId("done"), condition = { true }),
|
||||
),
|
||||
start = StageId("impl"),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `ticket re-entry suppresses the owner stage's own generative mandate`(): Unit = runBlocking {
|
||||
val provider = RecordingRoutingProvider()
|
||||
val (orchestrator, _) = buildMultiToolOrchestrator(provider = provider)
|
||||
val sessionId = SessionId("mandate-suppression")
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 10, backoffMs = 0))
|
||||
|
||||
orchestrator.run(sessionId, ownerGraphWithMandate(), config)
|
||||
|
||||
val impl = provider.implSourceTypes
|
||||
assertTrue(impl.isNotEmpty(), "impl must have run")
|
||||
// Forward visit: impl's own prompt is present, no ticket yet.
|
||||
assertTrue(impl.first().contains("agentPrompt"), "forward visit keeps the stage's own prompt")
|
||||
assertTrue("recoveryTicket" !in impl.first(), "no recovery ticket before any failure")
|
||||
// Ticket re-entry: the generative mandate is gone, the recovery ticket is the sole mandate.
|
||||
assertTrue(
|
||||
impl.any { "recoveryTicket" in it && "agentPrompt" !in it },
|
||||
"a repair re-entry dropped the generative mandate and carried only the recovery ticket; saw $impl",
|
||||
)
|
||||
}
|
||||
|
||||
private companion object {
|
||||
const val INTENT_ROUTE_BUDGET_TEST = 2
|
||||
// Mirror of DefaultSessionOrchestrator.RECOVERY_ROUTE_BUDGET (private there); kept in sync by the
|
||||
// `[1, 2]` assertion in the exhaustion test above.
|
||||
const val RECOVERY_ROUTE_BUDGET_TEST = 2
|
||||
|
||||
Reference in New Issue
Block a user