From c5289420a1b9bf785c31ccbbc0fc4a42d7f94114 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 01:57:04 +0400 Subject: [PATCH 1/4] fix(inference): tolerate a briefly-absent provider instead of killing the session (#299) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A transient provider connection drop (e.g. OOM-killed llama.cpp mid-request) collapsed into a hard NoEligibleProviderException that escaped runInference's try/catch (the route() call sat outside it), propagated past the stage retry loop, and landed in ServerModule's session-level catch — failing the WHOLE session even though the failure was retryable and the provider recovered seconds later. - SessionOrchestrator.runInference: wrap router.route() in try/catch so routing failures become InferenceResult.Failed and flow through the normal retryable/backoff/exhaustion machinery instead of escaping. - DefaultInferenceRouter.route: distinguish "capability never configured on any provider" (fail fast, waiting can't help) from "configured but currently unhealthy" (bounded wait/backoff — default 3 attempts x 2s — re-checking health before declaring NoEligibleProvider terminal). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G --- .../core/inference/DefaultInferenceRouter.kt | 36 +++++++++++- .../orchestration/SessionOrchestrator.kt | 10 +++- .../inference/DefaultInferenceRouterTest.kt | 56 +++++++++++++++++++ .../SessionOrchestratorIntegrationTest.kt | 37 ++++++++++++ 4 files changed, 137 insertions(+), 2 deletions(-) diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt index 3fa819d1..b48dd369 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt @@ -2,6 +2,7 @@ package com.correx.core.inference import com.correx.core.events.types.ProviderId import com.correx.core.events.types.StageId +import kotlinx.coroutines.delay import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlin.time.Duration @@ -21,6 +22,12 @@ class DefaultInferenceRouter( private val strategy: RoutingStrategy, private val cacheTtl: Duration = 5.seconds, private val timeSource: TimeSource = TimeSource.Monotonic, + // A provider that briefly drops (crash + qa-stack restart) shouldn't collapse into a hard + // NoEligibleProvider abort — give it bounded time to come back before giving up. This only + // applies when the capability IS configured on some provider but that provider is currently + // unhealthy; a capability nobody ever declared fails immediately (see routeCapabilityCandidates). + private val unavailableRetryAttempts: Int = 3, + private val unavailableRetryDelay: Duration = 2.seconds, ) : InferenceRouter { private val cache = mutableMapOf() @@ -45,12 +52,39 @@ class DefaultInferenceRouter( } } + private suspend fun refreshedHealth(provider: InferenceProvider): ProviderHealth = + lockFor(provider.id).withLock { + val fresh = provider.healthCheck() + mapMutex.withLock { cache[provider.id] = HealthEntry(fresh, timeSource.markNow()) } + fresh + } + override suspend fun route(stageId: StageId, requiredCapabilities: Set): InferenceProvider { val candidates = requiredCapabilities .flatMap { registry.resolve(it) } .distinctBy { it.id } .ifEmpty { registry.listAll() } - val healthy = candidates.filter { cachedHealth(it) !is ProviderHealth.Unavailable } + + // Nobody was ever configured with this capability set — no amount of waiting fixes that, + // fail fast instead of burning the bounded-wait budget below. + if (requiredCapabilities.isNotEmpty() && + candidates.none { it.capabilities().map { c -> c.capability }.toSet().containsAll(requiredCapabilities) } + ) { + throw NoEligibleProviderException(stageId, requiredCapabilities) + } + + var healthy = candidates.filter { cachedHealth(it) !is ProviderHealth.Unavailable } + var attempt = 0 + while (healthy.isEmpty() && attempt < unavailableRetryAttempts) { + attempt++ + log.warn( + "route: capability {} configured but all candidates unhealthy for stage={}" + + " — waiting {} (attempt {}/{}) before declaring NoEligibleProvider", + requiredCapabilities, stageId.value, unavailableRetryDelay, attempt, unavailableRetryAttempts, + ) + delay(unavailableRetryDelay) + healthy = candidates.filter { refreshedHealth(it) !is ProviderHealth.Unavailable } + } val selected = strategy.select(healthy, requiredCapabilities) // Post-selection re-check closes the TOCTOU window between initial filter and dispatch. when (val postHealth = selected.healthCheck()) { 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 a928bf21..b98c6ac7 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 @@ -300,7 +300,15 @@ abstract class SessionOrchestrator( // ToolCalling score and routes the stage to the best tool-caller. val requiredCapabilities = stageConfig.requiredCapabilities + if (withTools && stageConfig.allowedTools.isNotEmpty()) setOf(ModelCapability.ToolCalling) else emptySet() - val provider = inferenceRouter.route(stageId, requiredCapabilities, stageConfig.modelId) + // Routing itself can fail transiently (provider mid-crash-recovery) — it must be retryable + // like any other inference failure, not escape and kill the whole session (see #299). + val provider = try { + inferenceRouter.route(stageId, requiredCapabilities, stageConfig.modelId) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + return InferenceResult.Failed(e.message ?: "routing failed") + } log.debug( "[Orchestrator] inference session={} stage={} provider={} timeoutMs={}", sessionId.value, stageId.value, provider.id.value, timeoutMs, diff --git a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt index 5e405efb..d6284870 100644 --- a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt +++ b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt @@ -13,8 +13,10 @@ import com.correx.core.inference.RoutingStrategy import com.correx.testing.fixtures.inference.MockInferenceProvider import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertSame +import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows +import kotlin.time.Duration.Companion.milliseconds class DefaultInferenceRouterTest { @@ -157,4 +159,58 @@ class DefaultInferenceRouterTest { val result = router.route(stage, setOf(ModelCapability.General), "llama-cpp:sick") assertSame(healthy, result) } + + // ── bounded wait for a briefly-absent provider (#299) ───────────────────── + + @Test + fun `waits for the sole capable provider to recover instead of failing immediately`(): Unit = runBlocking { + var healthChecks = 0 + val recovering = object : InferenceProvider by provider("a", ModelCapability.ToolCalling) { + override suspend fun healthCheck(): ProviderHealth { + healthChecks++ + return if (healthChecks < 3) ProviderHealth.Unavailable("connection dropped") else ProviderHealth.Healthy + } + } + val router = DefaultInferenceRouter( + registryOf(recovering), + firstStrategy(), + unavailableRetryAttempts = 5, + unavailableRetryDelay = 5.milliseconds, + ) + val result = router.route(stage, setOf(ModelCapability.ToolCalling)) + assertSame(recovering, result) + assertTrue(healthChecks >= 3) { "expected at least 3 health checks, got $healthChecks" } + } + + @Test + fun `still throws NoEligibleProviderException if the sole provider never recovers within the bound`() { + val neverRecovers = MockInferenceProvider( + id = ProviderId("a"), + declaredCapabilities = setOf(CapabilityScore(ModelCapability.ToolCalling, 1.0)), + health = ProviderHealth.Unavailable("still down"), + ) + val router = DefaultInferenceRouter( + registryOf(neverRecovers), + throwingStrategy(), + unavailableRetryAttempts = 2, + unavailableRetryDelay = 5.milliseconds, + ) + assertThrows { + runBlocking { router.route(stage, setOf(ModelCapability.ToolCalling)) } + } + } + + @Test + fun `fails fast without waiting when the capability was never configured on any provider`(): Unit = runBlocking { + val p = provider("a", ModelCapability.General) // does not declare ToolCalling + val router = DefaultInferenceRouter( + registryOf(p), + throwingStrategy(), + unavailableRetryAttempts = 5, + unavailableRetryDelay = 10_000.milliseconds, // would time the test out if the wait loop ran + ) + assertThrows { + router.route(stage, setOf(ModelCapability.ToolCalling)) + } + } } diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index 39f2454d..cd833972 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -46,6 +46,7 @@ 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.NoEligibleProviderException import com.correx.core.inference.ProviderHealth import com.correx.core.inference.Token import com.correx.core.inference.TokenUsage @@ -216,6 +217,42 @@ class SessionOrchestratorIntegrationTest { assertTrue(failed.retryExhausted) } + @Test + fun `transient NoEligibleProviderException on a retry is tolerated, not a session-killing crash (#299)`(): Unit = + runBlocking { + val sessionId = SessionId("s2b") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0), + ) + val graph = threeStageGraph() + var routeCalls = 0 + + val recoveringOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines.copy( + inferenceRouter = object : InferenceRouter { + override suspend fun route(stageId: StageId, requiredCapabilities: Set): InferenceProvider { + routeCalls++ + // First attempt hits the provider mid-crash — subsequent retries find it back. + if (routeCalls == 1) throw NoEligibleProviderException(stageId, requiredCapabilities) + return MockInferenceProvider() + } + }, + ), + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + decisionJournalRepository = decisionJournalRepository, + ) + recoveringOrchestrator.run(sessionId, graph, config) + + val events = eventStore.read(sessionId) + // A crash would have skipped straight to an unhandled failure with no WorkflowCompletedEvent + // and no orderly retry — assert the session survived and made it through via the retry path. + assertTrue(routeCalls > 1) { "expected routing to be retried, only called $routeCalls time(s)" } + assertNotNull(events.find { it.payload is WorkflowCompletedEvent }) + assertTrue(events.none { it.payload is WorkflowFailedEvent }) + } + @Test fun `stage that matches no transition retries through the budget before failing`(): Unit = runBlocking { val sessionId = SessionId("s-no-transition-retry") From bf36252736ab66f0671afc1d926fd8681450fa13 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 02:01:35 +0400 Subject: [PATCH 2/4] feat(health): gate routing event-driven on connection drop, not just periodic poll (#300) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit HealthMonitor's periodic poll took ~18s to notice a dead provider — long after a retry had already re-selected it and died. Health needs to GATE routing, not just log reactively. - InferenceRouter.reportFailure(providerId, reason): new default-no-op method; DefaultInferenceRouter implements it by writing Unavailable straight into its health cache, bypassing healthCheck()/TTL entirely. - SessionOrchestrator.runInference: on a connection-level exception (ConnectException/SocketException/IOException or a message matching "prematurely closed"/"connection refused"/"connection reset"), call reportFailure immediately so the very next route() call — the retry driven by #299 — sees the provider as down right away instead of re-selecting it and dying again. Pairs with #299: routing now (1) reacts to a connection drop instantly and (2) waits/backs off for the capability to recover before declaring terminal. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G --- .../core/inference/DefaultInferenceRouter.kt | 10 +++++ .../correx/core/inference/InferenceRouter.kt | 8 ++++ .../orchestration/SessionOrchestrator.kt | 20 +++++++++ .../inference/DefaultInferenceRouterTest.kt | 24 ++++++++++ .../SessionOrchestratorIntegrationTest.kt | 45 +++++++++++++++++++ 5 files changed, 107 insertions(+) diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt index b48dd369..c43b114a 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/DefaultInferenceRouter.kt @@ -59,6 +59,16 @@ class DefaultInferenceRouter( fresh } + // Bypasses healthCheck()/TTL entirely — writes Unavailable straight into the cache so the very + // next route() call gates on it, closing the ~18s reactive-poll lag (#300). A later cache-TTL + // expiry or the bounded-wait re-check in route() will naturally pick the provider back up once + // its own healthCheck() reports healthy again. + override suspend fun reportFailure(providerId: ProviderId, reason: String) { + lockFor(providerId).withLock { + mapMutex.withLock { cache[providerId] = HealthEntry(ProviderHealth.Unavailable(reason), timeSource.markNow()) } + } + } + override suspend fun route(stageId: StageId, requiredCapabilities: Set): InferenceProvider { val candidates = requiredCapabilities .flatMap { registry.resolve(it) } diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt index b2436f95..b0924269 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceRouter.kt @@ -39,6 +39,14 @@ interface InferenceRouter { requiredCapabilities: Set, modelId: String?, ): InferenceProvider = route(stageId, requiredCapabilities) + + /** + * Event-driven health gate: called the moment a connection-level failure is observed on + * [providerId] (e.g. mid-request connection drop), so the NEXT route() call sees it as + * unavailable immediately instead of waiting for the next periodic health poll/cache TTL to + * catch up. Default no-op for routers that don't cache health. + */ + suspend fun reportFailure(providerId: com.correx.core.events.types.ProviderId, reason: String) = Unit } class NoEligibleProviderException( 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 b98c6ac7..2c01dd7e 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 @@ -409,6 +409,13 @@ abstract class SessionOrchestrator( } catch (e: CancellationException) { throw e // never swallow } catch (e: Exception) { + if (isConnectionLevelFailure(e)) { + // Mark it down NOW instead of waiting for the next periodic health poll (~18s lag, + // see #300) — the retry's route() call must see this provider as unavailable + // immediately so it gates/waits (#299) rather than instantly re-selecting the dead + // provider again. + inferenceRouter.reportFailure(provider.id, e.message ?: "connection failure") + } emit( sessionId, InferenceFailedEvent( @@ -423,6 +430,19 @@ abstract class SessionOrchestrator( } } + // Connection-level failures (provider crashed/restarting mid-request) should gate routing + // immediately; other failures (bad request, model error, HTTP 4xx) should not mark the + // provider down since the provider itself is still reachable. + private fun isConnectionLevelFailure(e: Exception): Boolean { + val message = e.message.orEmpty() + return e is java.net.ConnectException || + e is java.net.SocketException || + e is java.io.IOException || // covers ktor/CIO's IOException, which extends java.io.IOException on the JVM + message.contains("prematurely closed", ignoreCase = true) || + message.contains("connection refused", ignoreCase = true) || + message.contains("connection reset", ignoreCase = true) + } + // --- token estimation --- internal open suspend fun estimateTokens(content: String): Int { diff --git a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt index d6284870..244d014c 100644 --- a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt +++ b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/inference/DefaultInferenceRouterTest.kt @@ -213,4 +213,28 @@ class DefaultInferenceRouterTest { router.route(stage, setOf(ModelCapability.ToolCalling)) } } + + // ── event-driven health gating (#300) ───────────────────────────────────── + + @Test + fun `reportFailure gates the very next route call without waiting for a health poll`(): Unit = runBlocking { + var healthCheckCount = 0 + val alwaysClaimsHealthy = object : InferenceProvider by provider("a", ModelCapability.General) { + override suspend fun healthCheck(): ProviderHealth { + healthCheckCount++ + return ProviderHealth.Healthy // the provider's own health probe hasn't caught up yet + } + } + val backup = provider("b", ModelCapability.General) + val router = DefaultInferenceRouter(registryOf(alwaysClaimsHealthy, backup), firstStrategy()) + + // Sanity: before reportFailure, routes to the first (still "healthy") provider. + assertSame(alwaysClaimsHealthy, router.route(stage, setOf(ModelCapability.General))) + + router.reportFailure(alwaysClaimsHealthy.id, "connection reset") + + // Immediately after — no delay, no waiting for the next poll — routing must avoid it. + val result = router.route(stage, setOf(ModelCapability.General)) + assertSame(backup, result) + } } diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index cd833972..dec7cb8c 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -43,11 +43,14 @@ 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.DefaultInferenceRouter import com.correx.core.inference.InferenceRouter import com.correx.core.inference.InferenceState import com.correx.core.inference.ModelCapability import com.correx.core.inference.NoEligibleProviderException import com.correx.core.inference.ProviderHealth +import com.correx.core.inference.ProviderRegistry +import com.correx.core.inference.RoutingStrategy import com.correx.core.inference.Token import com.correx.core.inference.TokenUsage import com.correx.core.inference.Tokenizer @@ -253,6 +256,48 @@ class SessionOrchestratorIntegrationTest { assertTrue(events.none { it.payload is WorkflowFailedEvent }) } + @Test + fun `connection-level inference failure gates routing away from the dead provider immediately (#300)`(): Unit = + runBlocking { + val sessionId = SessionId("s2c") + val config = OrchestrationConfig( + retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0), + ) + val graph = threeStageGraph() + + val dead = MockInferenceProvider( + id = ProviderId("dead"), + forcedFailure = "Failed to parse HTTP response: the server prematurely closed the connection", + ) + val backup = MockInferenceProvider(id = ProviderId("backup")) + // Router's own health probe (dead.healthCheck()) still reports Healthy — it hasn't + // polled yet — so only the event-driven reportFailure() from the orchestrator's catch + // block (not the periodic poll) can gate the retry away from `dead`. + val registry = object : ProviderRegistry { + override fun register(provider: com.correx.core.inference.InferenceProvider) = Unit + override fun resolve(capability: ModelCapability) = listOf(dead, backup) + override fun listAll() = listOf(dead, backup) + override suspend fun healthCheckAll() = mapOf(dead.id to ProviderHealth.Healthy, backup.id to ProviderHealth.Healthy) + } + val router = DefaultInferenceRouter( + registry = registry, + strategy = RoutingStrategy { candidates, _ -> candidates.first() }, + ) + + val recoveringOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines.copy(inferenceRouter = router), + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + decisionJournalRepository = decisionJournalRepository, + ) + recoveringOrchestrator.run(sessionId, graph, config) + + val events = eventStore.read(sessionId) + assertNotNull(events.find { it.payload is WorkflowCompletedEvent }) + assertEquals(1, dead.inferCallCount) { "dead provider should only be attempted once, then gated out" } + } + @Test fun `stage that matches no transition retries through the budget before failing`(): Unit = runBlocking { val sessionId = SessionId("s-no-transition-retry") From 0af2f200d812688829ac1bb15e8cbc87e42cf14e Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 02:05:20 +0400 Subject: [PATCH 3/4] fix(prompts): resolve analyst epic-vs-child ambiguity that burned the full reasoning budget (#297) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live incident: "Write a web-ui for Correx" resolved to a pre-existing parent epic (webui-8, with children webui-9..14). The old prompt only said "name a task that covers this work" and, separately, "after decomposing, name the single ready child" — it never said what to do when an EXISTING task found via task_search/task_context is itself a parent. The model oscillated between naming the epic, re-decomposing, or working a child until it burned all 16384 reasoning tokens and emitted nothing. Reworded analyst_freestyle.md's task-framing section so every branch (existing leaf, existing parent, no task yet + single unit, no task yet + needs decomposition) converges on one rule: never name a parent/epic, always name the single ready child — removing the decision entirely rather than asking the model to make it under ambiguity. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G --- .../workflows/prompts/analyst_freestyle.md | 31 +++++++++++++------ 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/examples/workflows/prompts/analyst_freestyle.md b/examples/workflows/prompts/analyst_freestyle.md index 603eb5df..9da09876 100644 --- a/examples/workflows/prompts/analyst_freestyle.md +++ b/examples/workflows/prompts/analyst_freestyle.md @@ -6,16 +6,27 @@ Before deriving requirements, check for existing work: `task_search` for related blocking tasks and `task_context` to load any the goal names. Fold what you find into the analysis rather than re-deriving it; flag a duplicate instead of restating it. -Then frame the work as a task (per the task policy): -- If a task already covers this work, name its id (e.g. `auth-142`) in the analysis. -- If the goal is a single coherent unit one run can carry to review, `task_create` one and name its - id. -- If the goal has **dependency seams** (a thing that must land before another) or **independent - review/handoff points** (a piece worth shipping or reviewing on its own), `task_decompose` it into - a parent epic + `DEPENDS_ON`-linked children — one approval for the whole graph. A session works - one task at a time, so the children are claimed by *later* runs as they unblock; don't over-split. -- After decomposing, **name in the analysis the single task this run will work** — the one already - ready (no unmet dependency, e.g. the scaffold). Leave the blocked siblings for future runs. +Then frame the work as a task (per the task policy). A run always names exactly ONE task — never +a parent/epic — as the thing it will work: +- If an existing **leaf** task already covers this work (no children of its own), name its id + (e.g. `auth-142`) in the analysis. +- If an existing task covering this goal is itself a **parent/epic** (has `DEPENDS_ON`-linked + children, whether from a past run's `task_decompose` or found via `task_search`/`task_context`), + do **not** name the epic and do not decompose it again. Name the single ready child instead — the + one with no unmet dependency. If every child is already blocked/claimed, name the closest-to-ready + one and note in the analysis that this run is unblocking it, not completing the epic. +- If no task covers this work yet and the goal is a single coherent unit one run can carry to + review, `task_create` one and name its id. +- If no task covers this work yet and the goal has **dependency seams** (a thing that must land + before another) or **independent review/handoff points** (a piece worth shipping or reviewing on + its own), `task_decompose` it into a parent epic + `DEPENDS_ON`-linked children — one approval for + the whole graph. A session works one task at a time, so the children are claimed by *later* runs + as they unblock; don't over-split. Then name the single ready child (the one already unblocked, + e.g. the scaffold) — never the epic itself. + +There is always exactly one task id to name by the time you call `emit_artifact` — if you find +yourself unsure whether to name a parent or a child, the answer is always the child. Do not loop on +this decision. Either way later stages thread the named task through the plan; the rest wait to be claimed. From 8cc418a3812c57687d3b49849189006994896be1 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 02:22:09 +0400 Subject: [PATCH 4/4] feat(orchestration): escalate repeated scope/manifest write-block to user approval (#301) Small models frequently fail to comply with the WRITE_SCOPE/PATH_OUTSIDE_MANIFEST remediation ("add its path via task_update affected_paths") and instead thrash the same out-of-scope write call turn after turn, burning the session without ever completing. After `tuning.escalateScopeAfterN` (default 3) consecutive rejections of the SAME path for a scope/manifest reason, the orchestrator now reaches back to the FIRST rejected invocation of that path (its pristine, pre-degraded arguments), and routes it through the existing approval/pause flow instead of rejecting again. On approval, a new WriteScopeGrantedEvent widens the effective write manifest for that path for the rest of the session (mirroring how OutsidePathAccessGrantedEvent already widens out-of-workspace reads) and the first attempt's write is executed. On denial, the call is rejected as before. escalateScopeAfterN = 0 preserves the old hard-block-forever behavior. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G --- .../core/events/events/OrchestrationEvents.kt | 15 ++ .../orchestration/OrchestrationTuning.kt | 8 + .../SessionOrchestratorToolExec.kt | 201 +++++++++++++++++- .../SessionOrchestratorWorkspace.kt | 51 +++++ .../src/test/kotlin/ToolCallGateTest.kt | 175 +++++++++++++++ 5 files changed, 448 insertions(+), 2 deletions(-) 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 9f9e924c..ca447c70 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 @@ -53,6 +53,21 @@ data class OutsidePathAccessGrantedEvent( val path: String, ) : EventPayload +/** + * Records that the operator approved widening the write scope/manifest to admit [path], after + * the same path was rejected `escalate_scope_after_n` times in a row (small models frequently + * fail to comply with the WRITE_SCOPE/PATH_OUTSIDE_MANIFEST remediation and thrash instead — + * see #301). Folded the same way [OutsidePathAccessGrantedEvent] widens out-of-workspace reads: + * subsequent writes to this path this session are admitted without re-prompting. + */ +@Serializable +@SerialName("WriteScopeGranted") +data class WriteScopeGrantedEvent( + val sessionId: SessionId, + val stageId: StageId, + val path: String, +) : EventPayload + @Serializable @SerialName("OrchestrationPaused") data class OrchestrationPausedEvent( diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestrationTuning.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestrationTuning.kt index 891dab48..9aa08e0f 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestrationTuning.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestrationTuning.kt @@ -41,4 +41,12 @@ data class OrchestrationTuning( val recoveryRouteBudget: Int = 2, /** Budget for tier-2 intent-holder arbiter re-routing. */ val intentRouteBudget: Int = 2, + /** + * After this many consecutive WRITE_SCOPE/PATH_OUTSIDE_MANIFEST rejections of the SAME path in + * a session, stop hard-rejecting and escalate to user approval instead (#301) — small models + * frequently fail to comply with the remediation and thrash rather than widen scope themselves. + * 0 disables escalation (falls back to the hard-block-forever behavior). A headless session + * (no approver connected) auto-rejects the escalated prompt rather than hanging. + */ + val escalateScopeAfterN: Int = 3, ) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorToolExec.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorToolExec.kt index a69f68ee..2496e810 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorToolExec.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorToolExec.kt @@ -31,6 +31,7 @@ import com.correx.core.events.events.ToolExecutionRejectedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent import com.correx.core.events.events.ToolReceipt import com.correx.core.events.events.ToolRequest +import com.correx.core.events.events.WriteScopeGrantedEvent import com.correx.core.events.risk.RiskAction import com.correx.core.events.risk.RiskSummary import com.correx.core.toolintent.ToolCallAssessmentInput @@ -218,14 +219,36 @@ internal suspend fun SessionOrchestrator.dispatchToolCalls( // Per-task write scope: while a task is claimed, narrow the manifest to its affected // paths (recorded on the task) so the implementer can't write outside its unit of work. // Falls back to the stage's static manifest when nothing is claimed. - val effectiveManifest = taskClaimCoordinator?.activeScope(sessionId)?.takeIf { it.isNotEmpty() } - ?: stageConfig.writeManifest + val escalatedScope = escalatedWriteScopePaths(sessionId) + val effectiveManifest = ( + taskClaimCoordinator?.activeScope(sessionId)?.takeIf { it.isNotEmpty() } + ?: stageConfig.writeManifest + ) + escalatedScope val plane2Risk: RiskSummary? = runPlane2Assessment( sessionId, stageId, invocationId, toolCall.function.name, request, tool, effectives, effectiveManifest, )?.let { assessment -> if (assessment.recommendedAction == RiskAction.BLOCK) { val rationale = assessment.rationale.joinToString("; ") + // #301: a write repeatedly rejected for being outside the claimed task's scope or + // the stage's manifest — never for anything else — escalates to user approval + // after N same-path rejections instead of hard-blocking forever. Small models + // routinely fail to comply with the "add its path via task_update" remediation and + // thrash the same call rather than widen scope themselves. + val escalationPath = (parameters["path"] as? String) + ?.takeIf { isScopeBlockRationale(rationale) } + val escalateAfterN = tuning.escalateScopeAfterN + if (escalationPath != null && escalateAfterN > 0) { + val priorRejections = priorScopeRejections(sessionId, escalationPath) + if (priorRejections.size >= escalateAfterN) { + val firstAttempt = priorRejections.first().first + return@flatMap escalateWriteScopeBlock( + sessionId, stageId, invocationId, toolCall, tier, escalationPath, + firstAttempt, effectives, toolCallReasoning, approvalMode, assessment, + fileWrittenSlots, + ) + } + } // On a bad-path block, point the model at the closest real file so it fixes the // path instead of retrying the same wrong guess (deep package paths are easy to // misremember). Only fires when we can name a concrete match from the repo map. @@ -509,6 +532,180 @@ internal suspend fun SessionOrchestrator.dispatchToolCalls( } } +private fun parametersToArgumentsJson(parameters: Map): String = + kotlinx.serialization.json.buildJsonObject { + parameters.forEach { (k, v) -> + when (v) { + is List<*> -> put( + k, + kotlinx.serialization.json.buildJsonArray { + v.forEach { add(kotlinx.serialization.json.JsonPrimitive(it.toString())) } + }, + ) + else -> put(k, kotlinx.serialization.json.JsonPrimitive(v.toString())) + } + } + }.toString() + +/** + * #301: the Nth same-path WRITE_SCOPE/PATH_OUTSIDE_MANIFEST rejection routes into the existing + * approval/pause flow instead of hard-blocking again. [firstAttempt] is the FIRST invocation that + * was rejected for this path/reason (reached back into via [priorScopeRejections]) — its pristine + * arguments are what gets previewed and, on approval, executed; later attempts this session may + * have degraded (e.g. the model giving up and calling `task_update(action="block")` instead), so + * replaying those would not fulfil the original intent. + */ +internal suspend fun SessionOrchestrator.escalateWriteScopeBlock( + sessionId: SessionId, + stageId: StageId, + invocationId: ToolInvocationId, + toolCall: ToolCallRequest, + tier: Tier, + path: String, + firstAttempt: ToolInvocationRequestedEvent, + effectives: RunEffectives, + toolCallReasoning: String?, + approvalMode: ApprovalMode, + plane2Risk: RiskSummary?, + fileWrittenSlots: List, +): List { + val sourceId = toolCall.id ?: invocationId.value + val assistantEntry = ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L2, + sourceType = "assistantToolCall", + sourceId = sourceId, + content = Json.encodeToString(ToolCallRequest.serializer(), toolCall), + tokenEstimate = estimateTokens(toolCall.function.arguments), + role = EntryRole.ASSISTANT, + reasoning = toolCallReasoning, + ) + suspend fun rejected(reason: String): List { + blockTaskOnScopeRejection(sessionId, toolCall.function.name, reason) + emit( + sessionId, + ToolExecutionRejectedEvent( + invocationId = invocationId, + sessionId = sessionId, + toolName = toolCall.function.name, + tier = tier, + reason = reason, + ), + ) + return listOf( + assistantEntry, + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L2, + sourceType = "toolResult", + sourceId = sourceId, + content = "BLOCKED: $reason", + tokenEstimate = estimateTokens(reason), + role = EntryRole.TOOL, + ), + ) + } + + val projectId = effectives.policy?.workspaceRoot?.let { ProjectIdentity.of(it.toString()) } + val approvalCtx = ApprovalContext( + identity = ApprovalScopeIdentity(sessionId, stageId, projectId = projectId), + mode = approvalMode, + ) + val requestId = ApprovalRequestId(UUID.randomUUID().toString()) + val toolPreview = computeToolPreview( + firstAttempt.toolName, firstAttempt.request.parameters, effectives.policy?.workspaceRoot, + ) + val previewArguments = parametersToArgumentsJson(firstAttempt.request.parameters) + val domainRequest = DomainApprovalRequest( + id = requestId, + tier = firstAttempt.tier, + validationReportId = ValidationReportId(UUID.randomUUID().toString()), + riskSummaryId = null, + timestamp = Clock.System.now(), + toolName = firstAttempt.toolName, + preview = toolPreview ?: previewArguments.take(200), + ) + val sessionGrants = approvalRepository.getApprovalState(sessionId).grants.values + val ledgerGrants = approvalRepository.getApprovalState(GRANT_LEDGER_SESSION_ID).grants.values + val activeGrants = (sessionGrants + ledgerGrants).toList() + val engineDecision = approvalEngine.evaluate(domainRequest, approvalCtx, activeGrants, Clock.System.now()) + val approved: Boolean + val denyReason: String? + if (engineDecision.state == ApprovalStatus.COMPLETED) { + // Headless/no-approver sessions resolve here (e.g. approvalMode DENY auto-completes to a + // denial) — the escalation never hangs waiting on a human who isn't connected. + emitDecisionResolved(sessionId, domainRequest, engineDecision) + approved = engineDecision.isApproved + denyReason = engineDecision.reason + } else { + val deferred = CompletableDeferred() + pendingApprovals[requestId] = deferred + emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")) + emit( + sessionId, + ApprovalRequestedEvent( + requestId = requestId, + tier = firstAttempt.tier, + validationReportId = domainRequest.validationReportId, + riskSummaryId = null, + riskSummary = plane2Risk, + sessionId = sessionId, + stageId = stageId, + projectId = null, + toolName = firstAttempt.toolName, + preview = toolPreview ?: previewArguments.take(200), + ), + ) + val userDecision = try { + deferred.await() + } finally { + pendingApprovals.remove(requestId) + } + emitDecisionResolved(sessionId, domainRequest, userDecision) + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) + approved = userDecision.isApproved + denyReason = userDecision.reason + } + if (!approved) { + return rejected(denyReason ?: "scope-widening request denied") + } + + // Approved: widen the write scope for this path for the rest of the session (future writes to + // it skip straight past the risk plane, mirroring OutsidePathAccessGrantedEvent), then execute + // the FIRST rejected attempt's pristine write rather than the model's current — possibly + // degraded — call. + emit(sessionId, WriteScopeGrantedEvent(sessionId, stageId, path)) + val executor = effectives.executor ?: return rejected("no executor available to apply the approved write") + val tool = effectives.registry?.resolve(firstAttempt.toolName) + val result = executor.execute(firstAttempt.request) + val rendered = renderToolResult(firstAttempt.toolName, tool, result) + recordToolExecution( + sessionId, + stageId, + toolCall.copy(function = toolCall.function.copy(name = firstAttempt.toolName)), + invocationId, + tier, + result, + tool as? FileAffectingTool, + firstAttempt.request, + fileWrittenSlots, + rendered.fullOutputHash, + ) + return listOf( + assistantEntry, + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L2, + sourceType = "toolResult", + sourceId = sourceId, + content = "APPROVED: the user widened write scope to admit '$path' after repeated rejection; " + + "the original write is applied. ${rendered.content}", + tokenEstimate = estimateTokens(rendered.content), + role = EntryRole.TOOL, + ), + ) +} + /** * A rejected [SCOPE_PROPOSAL_TOOL] call means the operator denied widening the claimed task's * scope — block the task so the loop advances instead of the implementer retrying the same diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorWorkspace.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorWorkspace.kt index 262dc8b9..1e14e253 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorWorkspace.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorWorkspace.kt @@ -4,7 +4,9 @@ import com.correx.core.events.events.ToolCallAssessedEvent import com.correx.core.events.events.RepoMapComputedEvent import com.correx.core.events.events.OutsidePathAccessGrantedEvent import com.correx.core.events.events.ToolExecutionCompletedEvent +import com.correx.core.events.events.ToolExecutionRejectedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.WriteScopeGrantedEvent import com.correx.core.events.risk.RiskAction import com.correx.core.toolintent.SessionContextProjection import com.correx.core.events.events.TransitionExecutedEvent @@ -68,6 +70,55 @@ internal fun SessionOrchestrator.grantedOutsidePaths(sessionId: SessionId): Set< .map { it.path } .toSet() +/** + * Write-scope/manifest paths the operator has approved widening into this session (folded from + * [WriteScopeGrantedEvent] — see #301). Replay-safe: reads the log, never re-prompts a path once + * granted. + */ + +internal fun SessionOrchestrator.escalatedWriteScopePaths(sessionId: SessionId): Set = + eventStore.read(sessionId) + .mapNotNull { it.payload as? WriteScopeGrantedEvent } + .filter { it.sessionId == sessionId } + .map { it.path } + .toSet() + +/** The rule codes whose BLOCK rationale (`"[CODE] message"`, see [toRiskSummary]) makes a + * rejection eligible for the #301 same-path escalation — a write rejected purely for being + * outside the claimed task's scope or the stage's manifest, not for any other reason (path + * traversal, privileged location, etc. are never escalated). */ +private val SCOPE_BLOCK_CODES = setOf("WRITE_SCOPE", "PATH_OUTSIDE_MANIFEST") + +internal fun SessionOrchestrator.isScopeBlockRationale(rationale: String): Boolean = + SCOPE_BLOCK_CODES.any { rationale.contains("[$it]") } + +/** + * Every (first-attempt-invocation, rejection) pair this session where [path] was rejected for a + * WRITE_SCOPE/PATH_OUTSIDE_MANIFEST reason, oldest first. Reached back into purely by folding the + * event log — no branching/replay-engine change needed (#301). The invocation carries the + * pristine [ToolRequest] parameters from that attempt, which may differ from the model's current + * (possibly degraded) call. + */ + +internal fun SessionOrchestrator.priorScopeRejections( + sessionId: SessionId, + path: String, +): List> { + val events = eventStore.read(sessionId) + val invocationsById = events + .mapNotNull { it.payload as? ToolInvocationRequestedEvent } + .filter { it.sessionId == sessionId } + .associateBy { it.invocationId } + return events + .mapNotNull { it.payload as? ToolExecutionRejectedEvent } + .filter { it.sessionId == sessionId && isScopeBlockRationale(it.reason) } + .mapNotNull { rejected -> + val invocation = invocationsById[rejected.invocationId] ?: return@mapNotNull null + val invocationPath = invocation.request.parameters["path"] as? String + if (invocationPath == path) invocation to rejected else null + } +} + /** * Returns true when the agent must be offered only read-only tools on the next inference turn. * Active from the moment a READ_BEFORE_WRITE block lands until a FILE_READ completion follows it. diff --git a/testing/integration/src/test/kotlin/ToolCallGateTest.kt b/testing/integration/src/test/kotlin/ToolCallGateTest.kt index 1299e954..108c42fb 100644 --- a/testing/integration/src/test/kotlin/ToolCallGateTest.kt +++ b/testing/integration/src/test/kotlin/ToolCallGateTest.kt @@ -18,6 +18,8 @@ import com.correx.core.events.events.ToolExecutionCompletedEvent import com.correx.core.events.events.ToolExecutionRejectedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent import com.correx.core.events.events.ToolRequest +import com.correx.core.events.events.WriteScopeGrantedEvent +import com.correx.core.toolintent.rules.ManifestContainmentRule import com.correx.core.toolintent.rules.ReadBeforeWriteRule import com.correx.core.toolintent.rules.ReferenceExistsRule import com.correx.core.tools.contract.ParamRole @@ -50,6 +52,7 @@ 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.OrchestrationTuning import com.correx.core.kernel.orchestration.OrchestratorRepositories import com.correx.core.kernel.retry.DefaultRetryCoordinator import com.correx.core.risk.DefaultRiskAssessor @@ -717,6 +720,178 @@ class ToolCallGateTest { ) } + /** + * #301: a write repeatedly rejected for being outside the stage's declared manifest escalates + * to user approval after N same-path rejections instead of hard-blocking forever. On approval + * the FIRST attempt's pristine write is applied (widening the manifest for the session), and + * later attempts to the same path skip the prompt entirely. + */ + @Test + fun `Nth same-path PATH_OUTSIDE_MANIFEST rejection escalates to approval, which then executes the first attempt (#301)`(): Unit = + runBlocking { + val blockedPath = "/work/scratch/blocked.kt" + // T2 so the approval engine cannot auto-approve under PROMPT mode (T1 would auto-approve + // and the escalation would never actually pause) — mirrors the other PROMPT_USER tests. + val fileWriteTool = object : Tool { + override val name = "file_write" + override val description = "write" + override val parametersSchema: JsonObject = buildJsonObject {} + override val tier = Tier.T2 + override val requiredCapabilities: Set = setOf(ToolCapability.FILE_WRITE) + override val paramRoles: Map = mapOf("path" to ParamRole.PATH) + override fun validateRequest(request: ToolRequest): ValidationResult = ValidationResult.Valid + } + val toolRegistry = object : ToolRegistry { + override fun resolve(name: String): Tool? = if (name == "file_write") fileWriteTool else null + override fun all(): List = listOf(fileWriteTool) + } + + var turnCount = 0 + val provider = object : InferenceProvider { + override val id = ProviderId("scope-escalate") + override val name = "scope-escalate" + override val tokenizer: Tokenizer = MockTokenizer() + val requests = mutableListOf() + override suspend fun infer(request: InferenceRequest): InferenceResponse { + requests += request + turnCount++ + // Turns 1-3 keep retrying the exact same out-of-manifest write; turn 4 (after + // the escalated approval resolves) stops so the run completes. + return if (turnCount <= 3) { + InferenceResponse( + request.requestId, "", FinishReason.ToolCall, TokenUsage(1, 1), 0, + listOf( + ToolCallRequest( + id = "tc-$turnCount", + function = ToolCallFunction("file_write", """{"path":"$blockedPath"}"""), + ), + ), + ) + } else { + InferenceResponse(request.requestId, "done", FinishReason.Stop, TokenUsage(1, 1), 0) + } + } + override suspend fun healthCheck(): ProviderHealth = ProviderHealth.Healthy + override fun capabilities(): Set = setOf(CapabilityScore(ModelCapability.General, 1.0)) + } + + val fakeProbe = object : com.correx.core.toolintent.WorldProbe { + override fun exists(path: Path): Boolean = true + override fun resolveReal(path: Path): Path = path.toAbsolutePath().normalize() + } + + val eventStore = InMemoryEventStore() + val artifactStore = NoopArtifactStore() + val assessor = ToolCallAssessor(listOf(ManifestContainmentRule())) + val policy = WorkspacePolicy(workspace) + val executor = RecordingExecutor() + + 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 = LiveArtifactRepository(eventStore, DefaultArtifactReducer()), + approvalRepository = DefaultApprovalRepository( + DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())), + ), + ) + val engines = OrchestratorEngines( + transitionResolver = DefaultTransitionResolver { _, _ -> true }, + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = object : InferenceRouter { + override suspend fun route(stageId: StageId, requiredCapabilities: Set): InferenceProvider = provider + }, + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + toolExecutor = executor, + toolRegistry = toolRegistry, + toolCallAssessor = assessor, + workspacePolicy = policy, + worldProbe = fakeProbe, + ) + val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = artifactStore, + decisionJournalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ), + // Escalate after 2 same-path rejections instead of the default 3, so the test's + // 3rd attempt is the one that triggers the approval prompt. + tuning = OrchestrationTuning(escalateScopeAfterN = 2), + ) + + val sessionId = SessionId("scope-escalate") + val graph = WorkflowGraph( + id = "scope-escalate-test", + stages = mapOf( + StageId("A") to StageConfig( + allowedTools = setOf("file_write"), + writeManifest = listOf("allowed/**"), + ), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + val job = launch { orchestrator.run(sessionId, graph, config) } + + // Wait for the 3rd attempt's escalation to raise an ApprovalRequestedEvent. + val approval = withTimeout(5_000) { + var req: ApprovalRequestedEvent? = null + while (req == null) { + req = eventStore.read(sessionId).firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent } + if (req == null) yield() + } + req + } + assertEquals("file_write", approval.toolName) + + // Exactly 2 hard rejections happened before the escalation (turns 1 and 2); the 3rd + // attempt paused for approval instead of rejecting again. + val rejectionsBeforeApproval = eventStore.read(sessionId).count { it.payload is ToolExecutionRejectedEvent } + assertEquals(2, rejectionsBeforeApproval, "expected exactly 2 hard rejections before escalation") + assertTrue(!executor.executeCalled.get(), "executor must not run before the escalated approval resolves") + + orchestrator.submitApprovalDecision( + approval.requestId, + ApprovalDecision( + id = null, + requestId = approval.requestId, + outcome = ApprovalOutcome.APPROVED, + state = ApprovalStatus.COMPLETED, + tier = Tier.T2, + contextSnapshot = ApprovalContext( + identity = ApprovalScopeIdentity(sessionId, StageId("A"), projectId = null), + mode = ApprovalMode.PROMPT, + ), + resolutionTimestamp = Clock.System.now(), + reason = "approved widening", + ), + ) + + withTimeout(5_000) { + while (eventStore.read(sessionId).none { it.payload is WriteScopeGrantedEvent }) yield() + } + job.join() + + assertTrue(executor.executeCalled.get(), "the approved write must execute") + val grant = eventStore.read(sessionId).firstNotNullOfOrNull { it.payload as? WriteScopeGrantedEvent } + assertEquals(blockedPath, grant?.path) + } + @Test fun `empty allowedTools denies all domain tools and emits ToolExecutionRejectedEvent`(): Unit = runBlocking { val executor = RecordingExecutor()