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 3f301524..3e283982 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 @@ -58,7 +58,6 @@ import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.ExecutionPlanLockedEvent -import com.correx.core.events.events.FileWrittenEvent import com.correx.core.events.events.StageCheckpointFailedEvent import com.correx.core.events.events.StageCheckpointPassedEvent import com.correx.core.events.events.SteeringNoteAddedEvent @@ -162,6 +161,7 @@ private const val STAGE_COMPLETE_TOOL = "stage_complete" private const val EMIT_ARTIFACT_TOOL = "emit_artifact" private const val SCOPE_PROPOSAL_TOOL = "propose_scope" private const val READ_BEFORE_WRITE_CODE = "READ_BEFORE_WRITE" +private const val REFERENCE_EXISTS_CODE = "REFERENCE_EXISTS" private const val OUTPUT_SUMMARY_LIMIT = 500 private const val REPO_MAP_INJECT_TOP_K = 30 @@ -1165,11 +1165,23 @@ abstract class SessionOrchestrator( if (p.sessionId == sessionId && ToolCapability.FILE_READ in p.capabilities) pendingReadIds += p.invocationId.value is ToolCallAssessedEvent -> - if (p.sessionId == sessionId && p.disposition == RiskAction.BLOCK && - p.issues.any { it.code == READ_BEFORE_WRITE_CODE } - ) { - blocked = true - pendingReadIds.clear() + if (p.sessionId == sessionId && p.disposition == RiskAction.BLOCK) { + when { + p.issues.any { it.code == READ_BEFORE_WRITE_CODE } -> { + blocked = true + pendingReadIds.clear() + } + // A read BLOCKED because its target doesn't exist (REFERENCE_EXISTS) can never + // produce a completion, so it can't be what lifts read-only mode. Treat the + // attempt as satisfying read-before-write: the file is absent, so there is + // nothing to read before creating it. Without this, a stage that writes a NEW + // file deadlocks — writes are filtered out and every read of the not-yet-created + // target is blocked, burning MAX_TOOL_ROUNDS with no artifact. + p.issues.any { it.code == REFERENCE_EXISTS_CODE } -> { + blocked = false + pendingReadIds.clear() + } + } } is ToolExecutionCompletedEvent -> if (p.sessionId == sessionId && p.invocationId.value in pendingReadIds) { @@ -1440,6 +1452,10 @@ abstract class SessionOrchestrator( request: ToolRequest, fileWrittenSlots: List, ) { + // Invariant #5: every tool side effect is captured. This is the single authoritative + // ToolExecutionCompleted/Failed record and the source the read-before-write gate replays. + // FileWrittenEvent (with pre/post-image hashes for reversibility) and ToolExecutionStarted are + // emitted once, by SandboxedToolExecutor — NOT here — so completions aren't duplicated in the log. when (result) { is ToolResult.Failure -> emit( sessionId, @@ -1461,8 +1477,7 @@ abstract class SessionOrchestrator( exitCode = result.exitCode, outputSummary = result.output.take(OUTPUT_SUMMARY_LIMIT), // Carry the tool's structured metadata (e.g. file_read's contentHash) so - // session projections can read it — matching SandboxedToolExecutor, which - // already does this; the two completion paths were inconsistent. + // session projections can read it. structuredOutput = result.metadata, affectedEntities = affected.map { it.toString() }, durationMs = 0, @@ -1474,18 +1489,6 @@ abstract class SessionOrchestrator( ) if (affected.isNotEmpty()) { materializeFileWritten(sessionId, stageId, request, affected, fileWrittenSlots, diff) - affected.forEach { p -> - emit( - sessionId, - FileWrittenEvent( - invocationId = invocationId, - sessionId = sessionId, - path = p.toString(), - preExisted = true, - timestampMs = Clock.System.now().toEpochMilliseconds(), - ), - ) - } } } } @@ -2363,8 +2366,9 @@ private suspend fun computeToolPreview(toolName: String, parameters: Map - emitFailed(sessionId, invocationId, toolName, invalid.reason) return@withContext ToolResult.Failure( invocationId = invocationId, reason = invalid.reason, @@ -91,14 +86,13 @@ class SandboxedToolExecutor( } // 7. delegate - val durationMs = { System.currentTimeMillis() - startTime } - when (val result = delegate.execute(request)) { is ToolResult.Success -> { - val diff = computeDiff(affectedPaths, backupMap) restoreOrClean(backupMap, success = true) cleanWorkingDir(workingDir) - emitCompleted(sessionId, invocationId, toolName, result, tool, affectedPaths, durationMs(), diff) + // ToolExecutionCompleted/Failed are recorded once, by SessionOrchestrator (the + // authoritative execution recorder). Here we only emit what the orchestrator can't + // reconstruct: pre/post-image hashes (reversibility) and research-source markers. emitResearchSourceEvents(sessionId, request.stageId, result) emitFileMutations(sessionId, invocationId, affectedPaths, preImages) result @@ -107,7 +101,6 @@ class SandboxedToolExecutor( is ToolResult.Failure -> { restoreOrClean(backupMap, success = false) cleanWorkingDir(workingDir) - emitFailed(sessionId, invocationId, toolName, result.reason) result } } @@ -168,39 +161,6 @@ class SandboxedToolExecutor( toolName: String, ) = emit(sessionId, ToolExecutionStartedEvent(invocationId, sessionId, toolName)) - private suspend fun emitCompleted( - sessionId: SessionId, - invocationId: ToolInvocationId, - toolName: String, - result: ToolResult.Success, - tool: Tool, - affectedPaths: Set, - durationMs: Long, - diff: String? = null, - ) { - val affectedEntities = affectedPaths.map { it.toString() } - emit( - sessionId, - ToolExecutionCompletedEvent( - invocationId = invocationId, - sessionId = sessionId, - toolName = toolName, - receipt = ToolReceipt( - invocationId = invocationId, - toolName = toolName, - exitCode = result.exitCode, - outputSummary = result.output, - structuredOutput = result.metadata, - affectedEntities = affectedEntities, - durationMs = durationMs, - tier = tool.tier, - timestamp = Clock.System.now(), - diff = diff, - ), - ), - ) - } - /** * Promotes the research fetch's quality + content-hash (BACKLOG §D) from the generic * [ToolExecutionCompletedEvent] metadata into first-class events, additively: the metadata write @@ -232,44 +192,6 @@ class SandboxedToolExecutor( } } - @Suppress("MaxLineLength") - private fun computeDiff(affectedPaths: Set, backupMap: Map): String? { - val diffs = mutableListOf() - - // Files that existed before the tool ran (backed up) - for ((original, backup) in backupMap) { - runCatching { - val oldContent = Files.readString(backup) - val newContent = if (Files.exists(original)) Files.readString(original) else "" - if (oldContent != newContent) { - diffs.add(DiffUtil.unifiedDiff(oldContent, newContent, original.toString())) - } - } - } - - // Newly created files (no backup, but affected and now exist) - for (path in affectedPaths) { - if (!backupMap.containsKey(path) && Files.exists(path)) { - runCatching { - val newContent = Files.readString(path) - if (newContent.isNotEmpty()) { - diffs.add(DiffUtil.unifiedDiff("", newContent, path.toString())) - } - } - } - } - - if (diffs.isEmpty()) return null - return diffs.joinToString("\n") - } - - private suspend fun emitFailed( - sessionId: SessionId, - invocationId: ToolInvocationId, - toolName: String, - reason: String, - ) = emit(sessionId, ToolExecutionFailedEvent(invocationId, sessionId, toolName, reason)) - private suspend fun emit(sessionId: SessionId, payload: EventPayload) { eventDispatcher.emit(payload, sessionId) } diff --git a/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorValidationTest.kt b/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorValidationTest.kt index d470a774..61200ca3 100644 --- a/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorValidationTest.kt +++ b/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorValidationTest.kt @@ -82,6 +82,9 @@ class SandboxedToolExecutorValidationTest { result as ToolResult.Failure assertTrue(result.recoverable, "validation failures must be recoverable so the model can correct + retry") - assertEquals(1, events.payloads.filterIsInstance().size) + // The executor surfaces the failure via the returned result; it does NOT emit + // ToolExecutionFailedEvent — SessionOrchestrator is the single authoritative recorder of + // tool completion/failure (avoids the double-emit that duplicated tool rows in the TUI). + assertEquals(0, events.payloads.filterIsInstance().size) } } diff --git a/testing/integration/src/test/kotlin/ToolCallGateTest.kt b/testing/integration/src/test/kotlin/ToolCallGateTest.kt index 900201e8..0316e493 100644 --- a/testing/integration/src/test/kotlin/ToolCallGateTest.kt +++ b/testing/integration/src/test/kotlin/ToolCallGateTest.kt @@ -19,6 +19,7 @@ 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.toolintent.rules.ReadBeforeWriteRule +import com.correx.core.toolintent.rules.ReferenceExistsRule import com.correx.core.tools.contract.ParamRole import com.correx.core.events.execution.RetryPolicy import com.correx.core.events.risk.RiskAction @@ -575,6 +576,146 @@ class ToolCallGateTest { ) } + @Test + fun `a read blocked by REFERENCE_EXISTS lifts read-only mode so a new-file write is not deadlocked`(): Unit = runBlocking { + val writeTarget = "/work/Existing.kt" // exists → READ_BEFORE_WRITE blocks the unread write + val ghostRead = "/work/Ghost.kt" // absent → REFERENCE_EXISTS blocks the read + + val fileWriteTool = object : Tool { + override val name = "file_write" + override val description = "write" + override val parametersSchema: JsonObject = buildJsonObject {} + override val tier = Tier.T1 + 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 fileReadTool = object : Tool { + override val name = "file_read" + override val description = "read" + override val parametersSchema: JsonObject = buildJsonObject {} + override val tier = Tier.T1 + override val requiredCapabilities: Set = setOf(ToolCapability.FILE_READ) + override val paramRoles: Map = mapOf("path" to ParamRole.PATH) + override fun validateRequest(request: ToolRequest): ValidationResult = ValidationResult.Valid + } + val twoToolRegistry = object : ToolRegistry { + override fun resolve(name: String): Tool? = when (name) { + "file_write" -> fileWriteTool + "file_read" -> fileReadTool + else -> null + } + override fun all(): List = listOf(fileWriteTool, fileReadTool) + } + + // turn 1 → write existing-but-unread (→ READ_BEFORE_WRITE block → read-only), + // turn 2 → read a file that does not exist (→ REFERENCE_EXISTS block), + // turn 3 → write again; file_write must be offered again (read-only lifted, no deadlock). + val offeredToolsPerTurn = mutableListOf>() + var turnCount = 0 + val provider = object : InferenceProvider { + override val id = ProviderId("rbw-ghost") + override val name = "rbw-ghost" + override val tokenizer: Tokenizer = MockTokenizer() + override suspend fun infer(request: InferenceRequest): InferenceResponse { + offeredToolsPerTurn += request.tools.map { it.function.name }.toSet() + turnCount++ + val toolCall = when (turnCount) { + 1 -> ToolCallRequest(id = "tc-w1", function = ToolCallFunction("file_write", """{"path":"$writeTarget"}""")) + 2 -> ToolCallRequest(id = "tc-r1", function = ToolCallFunction("file_read", """{"path":"$ghostRead"}""")) + 3 -> ToolCallRequest(id = "tc-w2", function = ToolCallFunction("file_write", """{"path":"$writeTarget"}""")) + else -> null + } + return if (toolCall != null) { + InferenceResponse(request.requestId, "", FinishReason.ToolCall, TokenUsage(1, 1), 0, listOf(toolCall)) + } 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)) + } + + // Path-discriminating probe: the write target exists, the ghost read does not. + val fakeProbe = object : com.correx.core.toolintent.WorldProbe { + override fun exists(path: java.nio.file.Path): Boolean = path.toString().endsWith("Existing.kt") + override fun resolveReal(path: java.nio.file.Path): java.nio.file.Path = path.toAbsolutePath().normalize() + } + + val eventStore = InMemoryEventStore() + val artifactStore = NoopArtifactStore() + val assessor = ToolCallAssessor(listOf(ReadBeforeWriteRule(), ReferenceExistsRule())) + val policy = WorkspacePolicy(workspace) + + 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 = object : ToolExecutor { + override suspend fun execute(request: ToolRequest): ToolResult = + ToolResult.Success(request.invocationId, "ok", 0) + }, + toolRegistry = twoToolRegistry, + toolCallAssessor = assessor, + workspacePolicy = policy, + worldProbe = fakeProbe, + ) + val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = artifactStore, + decisionJournalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ), + ) + + val sessionId = SessionId("rbw-ghost-deadlock") + val graph = WorkflowGraph( + id = "rbw-ghost-test", + stages = mapOf( + StageId("A") to StageConfig(allowedTools = setOf("file_write", "file_read")), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + orchestrator.run(sessionId, graph, OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))) + + // Turn 2 is read-only (the turn-1 write was READ_BEFORE_WRITE-blocked). + assertTrue(offeredToolsPerTurn.size >= 3, "expected at least 3 inference turns, got ${offeredToolsPerTurn.size}") + assertTrue("file_write" !in offeredToolsPerTurn[1], "turn 2 must be read-only after the write block") + // Turn 3: the ghost read could never complete, so without the fix read-only would never lift + // and file_write would stay filtered forever. The fix lifts it on the REFERENCE_EXISTS block. + assertTrue( + "file_write" in offeredToolsPerTurn[2], + "turn 3 must offer file_write again — a REFERENCE_EXISTS-blocked read must lift read-only, " + + "got: ${offeredToolsPerTurn[2]}", + ) + } + @Test fun `empty allowedTools denies all domain tools and emits ToolExecutionRejectedEvent`(): Unit = runBlocking { val executor = RecordingExecutor()