fix(kernel): de-dup tool events, restore approval diff, break read-before-write deadlock

Three tool-path correctness fixes: - Tool-execution events were emitted twice — SandboxedToolExecutor and SessionOrchestrator.recordToolExecution both emitted Completed/Failed/ FileWritten per call (double TUI rows, double-counted metrics). Split ownership: orchestrator is the sole recorder of Completed/Failed (truncates output, drives the read-before-write gate); executor owns Started + FileWritten (pre/post-image hashes it alone can capture). - computeToolPreview still gated on operation == "write", but file_write lost its operation param when delete was split out, so it bailed to raw-args for every write (approval card showed raw JSON). Drop the dead guard. - A file_read blocked by REFERENCE_EXISTS can never complete, so it could never lift read-only mode: a stage writing a NEW file deadlocked (writes filtered, every read of the not-yet-created target blocked) until MAX_TOOL_ROUNDS. Lift read-only on a REFERENCE_EXISTS block, and reword the block message to tell the model to file_write directly. Regression test added.
2026-07-03 01:01:19 +04:00
parent ca3fd7971e
commit 2698971082
5 changed files with 178 additions and 106 deletions
@@ -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<TypedArtifactSlot>,
    ) {
+        // 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<String,
    if (toolName == "task_decompose") return renderDecomposePreview(parameters)
    if (toolName != "file_write") return null
    val path = parameters["path"] as? String ?: return null
-    val operation = parameters["operation"] as? String
-    if (operation != "write") return null
+    // file_write no longer carries an `operation` param (delete was split into file_delete), so the
+    // preview is unconditional: path + content is always a write. The old `operation == "write"` gate
+    // silently bailed to the raw-JSON args fallback for every file_write.
    val proposedContent = parameters["content"] as? String ?: return null

    val existingContent = withContext(Dispatchers.IO) {
@@ -45,8 +45,10 @@ class ReferenceExistsRule : ToolCallRule {
            if (inWorkspace && !exists) {
                issues += ValidationIssue(
                    code = RULE_CODE,
-                    message = "Tool '${input.request.toolName}' references '$raw', which does not " +
-                        "exist — list the directory or fix the path; do not assume its contents.",
+                    message = "Tool '${input.request.toolName}' references '$raw', which does not exist. " +
+                        "Do NOT keep retrying the read. If you intend to create this file, call file_write " +
+                        "directly — you cannot read a file that does not exist yet. Otherwise list the " +
+                        "directory or fix the path; do not assume its contents.",
                    severity = ValidationSeverity.ERROR,
                )
                disposition = maxAction(disposition, RiskAction.BLOCK)
@@ -7,15 +7,12 @@ import com.correx.core.events.events.FileWrittenEvent
 import com.correx.core.events.events.LowQualityExtractionEvent
 import com.correx.core.events.events.SourceFetchedEvent
 import com.correx.core.events.events.ToolExecutionCompletedEvent
-import com.correx.core.events.events.ToolExecutionFailedEvent
 import com.correx.core.events.events.ToolExecutionStartedEvent
-import com.correx.core.events.events.ToolReceipt
 import com.correx.core.events.events.ToolRequest
 import com.correx.core.events.types.SessionId
 import com.correx.core.events.types.StageId
 import com.correx.core.events.types.ToolInvocationId
 import com.correx.core.tools.contract.FileAffectingTool
-import com.correx.core.tools.contract.Tool
 import com.correx.core.tools.contract.ToolExecutor
 import com.correx.core.tools.contract.ToolResult
 import com.correx.core.tools.contract.ValidationResult
@@ -43,7 +40,6 @@ class SandboxedToolExecutor(
        val sessionId = request.sessionId
        val invocationId = request.invocationId
        val toolName = request.toolName
-        val startTime = System.currentTimeMillis()

        // 1. resolve tool
        val tool = registry.resolve(toolName)
@@ -61,7 +57,6 @@ class SandboxedToolExecutor(
        // tool's arg schema and the model can correct + retry (bounded by MAX_TOOL_ROUNDS), instead
        // of stranding the stage with no artifact ("no transition condition matched").
        (tool.validateRequest(request) as? ValidationResult.Invalid)?.let { invalid ->
-            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<Path>,
-        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<Path>, backupMap: Map<Path, Path>): String? {
-        val diffs = mutableListOf<String>()
-
-        // 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)
    }
@@ -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<ToolExecutionFailedEvent>().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<ToolExecutionFailedEvent>().size)
    }
 }
@@ -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<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 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<ToolCapability> = setOf(ToolCapability.FILE_READ)
+            override val paramRoles: Map<String, ParamRole> = 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<Tool> = 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<Set<String>>()
+        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<CapabilityScore> = 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<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 = 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()