merge: integrate sonnet-vikunja (#299,#300,#297,#301) into codex handoff HEAD

# Conflicts: # core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestrationTuning.kt # examples/workflows/prompts/analyst_freestyle.md
2026-07-21 11:43:05 +04:00
parent 6e844ef1e1 8cc418a381
commit 28d369ebce
11 changed files with 714 additions and 6 deletions
@@ -43,10 +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
@@ -216,6 +220,84 @@ 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<ModelCapability>): 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 `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")
@@ -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
@@ -719,6 +722,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()