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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GeyGFXczJb8RUWGBKmkm6G
This commit is contained in:
kami
@@ -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<ToolCapability> = setOf(ToolCapability.FILE_WRITE)
|
||||
override val paramRoles: Map<String, ParamRole> = 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<Tool> = 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<InferenceRequest>()
|
||||
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<CapabilityScore> = 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<InferenceState> {
|
||||
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<ModelCapability>): 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()
|
||||
|
||||
Reference in New Issue
Block a user