fix(kernel,tools,validation): unblock role_pipeline end-to-end + QA audit

Live QA audit of role_pipeline against a sample repo surfaced and fixed a chain of defects that prevented any tool+artifact workflow from running end-to-end. The pipeline now completes cleanly with a real, validated, reviewed file change. - F-008 workflow: propagate stage token_budget -> generationConfig.maxTokens (TomlWorkflowLoader) so large stages stop truncating at the 2048 default. - F-009 tools/kernel: file_read missing-file is recoverable; recoverable tool errors feed back into the loop instead of aborting the stage. - F-010/F-018 kernel: reject premature stage_complete and nudge the model to emit the owed artifact / invoke a write tool (bounded by MAX_TOOL_ROUNDS). - F-011 schemas: list-typed artifact fields string -> array (analysis/design/ impl_plan) to match model output. - F-012 kernel: strip an outer markdown code fence from LLM artifacts. - F-013 validation: payload-validation failures are retryable; structural failures stay terminal (wires the invariant-#7 reject+retry contract). - F-014 tools: file_edit schema matches its real params; edits emit a diff. - F-015 kernel: record tool-execution events + materialise a real file_written artifact from the on-disk write; gate validation so a no-write stage cannot be rubber-stamped (no more false-success runs). - F-016 server: raise stageTimeoutMs 60s -> 180s for slow local models. - F-019 artifacts/kernel: file_written artifact carries the unified diff so the reviewer can verify the change (shared DiffUtil; file_write emits a diff too). - F-020 tools: model-correctable file_edit failures are recoverable + steer toward replace/file_write over the fragile patch op. Full findings register (F-001..F-021, incl. open F-021 resume rehydrate bug) in qa/audit-report-2026-06-08.md. All module tests green.
2026-06-08 21:51:21 +04:00
parent d518400b5f
commit b407b47503
16 changed files with 664 additions and 45 deletions
@@ -12,7 +12,9 @@ import com.correx.core.approvals.model.ApprovalScopeIdentity
 import com.correx.core.approvals.model.DomainApprovalRequest
 import com.correx.core.artifacts.ArtifactState
 import com.correx.core.artifacts.kind.JsonSchema
+import com.correx.core.artifacts.kind.FileWrittenArtifact
 import com.correx.core.artifacts.kind.ProcessResultArtifact
+import com.correx.core.artifacts.kind.TypedArtifactSlot
 import com.correx.core.artifactstore.ArtifactStore
 import com.correx.core.context.builder.ContextPackBuilder
 import com.correx.core.context.model.ContextEntry
@@ -40,9 +42,13 @@ 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.SteeringNoteAddedEvent
+import com.correx.core.events.events.ToolExecutionCompletedEvent
+import com.correx.core.events.events.ToolExecutionFailedEvent
 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.risk.RiskAction
 import com.correx.core.events.risk.RiskSummary
@@ -87,6 +93,7 @@ import com.correx.core.risk.RiskContext
 import com.correx.core.risk.toApprovalTier
 import com.correx.core.sessions.Session
 import com.correx.core.tools.compression.ToolOutputContext
+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
@@ -114,6 +121,8 @@ import com.correx.core.journal.DecisionJournalRenderer
 import com.correx.core.journal.DefaultDecisionJournalRepository
 import com.correx.core.kernel.orchestration.subagent.SubagentRunner
 import org.slf4j.LoggerFactory
+import java.nio.file.Files
+import java.nio.file.Paths
 import java.util.*
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
@@ -121,6 +130,7 @@ import kotlin.coroutines.cancellation.CancellationException

 private const val MAX_TOOL_ROUNDS = 10
 private const val STAGE_COMPLETE_TOOL = "stage_complete"
+private const val OUTPUT_SUMMARY_LIMIT = 500
 private const val REPO_MAP_INJECT_TOP_K = 30

@SuppressWarnings(
@@ -366,16 +376,74 @@ abstract class SessionOrchestrator(
            sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
        )
        var toolRounds = 0
+
+        val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" }
+        fun owesFileWrite(): Boolean = fileWrittenSlots.isNotEmpty() &&
+            fileWrittenSlots.any { artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() }
+
+        // Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS).
+        // Returns the new result rather than mutating inferenceResult, to preserve smart casts.
+        suspend fun pushBack(nudge: String): InferenceResult {
+            accumulatedEntries = accumulatedEntries + ContextEntry(
+                id = ContextEntryId(UUID.randomUUID().toString()),
+                layer = ContextLayer.L2,
+                sourceType = "toolResult",
+                sourceId = UUID.randomUUID().toString(),
+                content = nudge,
+                tokenEstimate = estimateTokens(nudge),
+                role = EntryRole.TOOL,
+            )
+            currentContext = contextPackBuilder.build(
+                id = ContextPackId(UUID.randomUUID().toString()),
+                sessionId = sessionId,
+                stageId = stageId,
+                entries = accumulatedEntries,
+                budget = TokenBudget(limit = stageConfig.tokenBudget),
+            )
+            emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
+            toolRounds++
+            return runInference(
+                sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
+            )
+        }
+
+        val writeNudge = "ERROR: you described the change but did not apply it. You MUST call the " +
+            "file_edit or file_write tool to write the change to the file before finishing — do " +
+            "not output the file contents as a message."
+
        while (
            inferenceResult is InferenceResult.Success &&
-            inferenceResult.response.finishReason is FinishReason.ToolCall &&
-            toolRounds < MAX_TOOL_ROUNDS
+            toolRounds < MAX_TOOL_ROUNDS &&
+            (inferenceResult.response.finishReason is FinishReason.ToolCall || owesFileWrite())
        ) {
+            // Content turn (no tool call) but the stage still owes a file_written artifact: the
+            // model emitted the change as prose instead of invoking the write tool. Nudge it to
+            // actually write (F-018).
+            if (inferenceResult.response.finishReason !is FinishReason.ToolCall) {
+                inferenceResult = pushBack(writeNudge)
+                continue
+            }
            // stage_complete is a meta-tool: the LLM calls it to signal the stage's goal is met.
            // Skip tool dispatch — the loop exits and normal validation/transition follows
            // (emitToolArtifacts + verifyProduces run on the success path after the loop).
            if (inferenceResult.response.toolCalls.any { it.function.name == STAGE_COMPLETE_TOOL }) {
-                break
+                // Premature-completion guard: don't accept stage_complete while the stage still
+                // owes an artifact. For an LLM-emitted slot, nudge to emit the JSON (F-010); for a
+                // file_written slot, nudge to invoke the write tool (F-018). Otherwise complete.
+                val owedLlm = llmEmittedSlots.firstOrNull()?.takeIf {
+                    inferenceResult.response.text.isBlank() &&
+                        artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank()
+                }
+                when {
+                    owedLlm != null -> inferenceResult = pushBack(
+                        "ERROR: stage_complete is not allowed yet — you have not emitted the " +
+                            "required '${owedLlm.name.value}' artifact. Respond with the artifact as " +
+                            "a single JSON message (no tool call) matching the provided schema.",
+                    )
+                    owesFileWrite() -> inferenceResult = pushBack(writeNudge)
+                    else -> break
+                }
+                continue
            }
            val toolEntries = dispatchToolCalls(sessionId, stageId,
                inferenceResult.response.toolCalls, stageConfig, effectives)
@@ -387,14 +455,9 @@ abstract class SessionOrchestrator(
                    retryable = false,
                )
            }
-            val errorEntry = toolEntries.firstOrNull { it.content.startsWith("ERROR:") }
-            if (errorEntry != null) {
-                emitProcessResultEvents(sessionId, stageId, stageConfig)
-                return StageExecutionResult.Failure(
-                    errorEntry.content.removePrefix("ERROR: "),
-                    retryable = true,
-                )
-            }
+            // Recoverable (ERROR:) tool failures are NOT terminal — they flow back into the
+            // loop as tool-result context so the model can see the error and adapt (bounded by
+            // MAX_TOOL_ROUNDS). Only FATAL: failures (handled above) abort the stage.
            accumulatedEntries = accumulatedEntries + toolEntries
            currentContext = contextPackBuilder.build(
                id = ContextPackId(UUID.randomUUID().toString()),
@@ -459,6 +522,21 @@ abstract class SessionOrchestrator(
    ): String = "[SessionOrchestrator] stage=${stageId.value}: " +
            "failed to load prompt '$path': ${throwable.message}"

+    /**
+     * Strip an outer markdown code fence (```` ```json `` / ``` ``` ````) when the whole response
+     * is a single fenced block. Conservative: only acts when the trimmed text both starts and ends
+     * with a fence, so prose responses that merely contain an inline block are left untouched.
+     */
+    private fun stripCodeFence(text: String): String {
+        val trimmed = text.trim()
+        if (!trimmed.startsWith("```") || !trimmed.endsWith("```")) return text
+        val withoutClose = trimmed.removeSuffix("```").trimEnd()
+        val firstNewline = withoutClose.indexOf('\n')
+        if (firstNewline < 0) return text
+        // Drop the opening fence line (``` optionally followed by a language tag).
+        return withoutClose.substring(firstNewline + 1).trim()
+    }
+
    @Suppress("CyclomaticComplexMethod")
    private suspend fun dispatchToolCalls(
        sessionId: SessionId,
@@ -469,6 +547,7 @@ abstract class SessionOrchestrator(
    ): List<ContextEntry> {
        val executor = effectives.executor ?: return emptyList()
        val processResultSlots = stageConfig.produces.filter { it.kind.id == "process_result" }
+        val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" }
        return toolCalls.flatMap { toolCall ->
            val invocationId = ToolInvocationId(UUID.randomUUID().toString())
            val parameters = runCatching {
@@ -730,6 +809,16 @@ abstract class SessionOrchestrator(
                )
            }

+            // Record the tool execution outcome as events (invariant #5: every side effect is
+            // captured; silent execution is not allowed). For file-mutating tools, materialise a
+            // real file_written artifact from the actual on-disk result — so a stage that did NOT
+            // write cannot have its file_written slot rubber-stamped as validated (F-015), and the
+            // artifact carries the diff as evidence of the change.
+            recordToolExecution(
+                sessionId, stageId, toolCall, invocationId, tier, result,
+                tool as? FileAffectingTool, request, fileWrittenSlots,
+            )
+
            val sourceId = toolCall.id ?: invocationId.value
            val assistantEntry = ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
@@ -942,12 +1031,120 @@ abstract class SessionOrchestrator(
        stageConfig.produces
            .filter { !it.kind.llmEmitted }
            .forEach { slot ->
+                // A file_written slot must be backed by real content materialised from an actual
+                // successful write (see recordToolExecution). If none happened, do NOT fabricate a
+                // validated artifact (F-015) — leave the slot unproduced so verifyProduces fails the
+                // stage and it retries, rather than reporting a green run that changed nothing.
+                if (slot.kind.id == "file_written" &&
+                    artifactContentCache["${sessionId.value}:${slot.name.value}"] == null
+                ) {
+                    return@forEach
+                }
                emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1))
                emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId))
                emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId))
            }
    }

+    private suspend fun recordToolExecution(
+        sessionId: SessionId,
+        stageId: StageId,
+        toolCall: ToolCallRequest,
+        invocationId: ToolInvocationId,
+        tier: Tier,
+        result: ToolResult,
+        fileTool: FileAffectingTool?,
+        request: ToolRequest,
+        fileWrittenSlots: List<TypedArtifactSlot>,
+    ) {
+        when (result) {
+            is ToolResult.Failure -> emit(
+                sessionId,
+                ToolExecutionFailedEvent(invocationId, sessionId, toolCall.function.name, result.reason),
+            )
+
+            is ToolResult.Success -> {
+                val diff = result.metadata["diff"]?.takeIf { it.isNotBlank() }
+                val affected = fileTool?.affectedPaths(request).orEmpty()
+                emit(
+                    sessionId,
+                    ToolExecutionCompletedEvent(
+                        invocationId = invocationId,
+                        sessionId = sessionId,
+                        toolName = toolCall.function.name,
+                        receipt = ToolReceipt(
+                            invocationId = invocationId,
+                            toolName = toolCall.function.name,
+                            exitCode = result.exitCode,
+                            outputSummary = result.output.take(OUTPUT_SUMMARY_LIMIT),
+                            affectedEntities = affected.map { it.toString() },
+                            durationMs = 0,
+                            tier = tier,
+                            timestamp = Clock.System.now(),
+                            diff = diff,
+                        ),
+                    ),
+                )
+                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(),
+                            ),
+                        )
+                    }
+                }
+            }
+        }
+    }
+
+    /**
+     * Materialise a [FileWrittenArtifact] from the actual on-disk file after a successful mutation,
+     * store it as the stage's file_written slot content, and record [ArtifactContentStoredEvent].
+     * This ties the file_written artifact to a real write so validation can no longer be faked.
+     */
+    private suspend fun materializeFileWritten(
+        sessionId: SessionId,
+        stageId: StageId,
+        request: ToolRequest,
+        affected: Set<java.nio.file.Path>,
+        fileWrittenSlots: List<TypedArtifactSlot>,
+        diff: String?,
+    ) {
+        if (fileWrittenSlots.isEmpty()) return
+        val relPath = request.parameters["path"] as? String ?: return
+        val onDisk = affected.firstOrNull() ?: Paths.get(relPath)
+        val bytes = runCatching { Files.readAllBytes(onDisk) }.getOrNull() ?: return
+        val contentHash = artifactStore.put(bytes)
+        val artifact = FileWrittenArtifact(
+            path = relPath,
+            contentHash = contentHash.value,
+            size = bytes.size.toLong(),
+            mode = "0644",
+            diff = diff,
+        )
+        val json = Json.encodeToString(FileWrittenArtifact.serializer(), artifact)
+        val storedHash = artifactStore.put(json.toByteArray())
+        fileWrittenSlots.forEach { slot ->
+            artifactContentCache["${sessionId.value}:${slot.name.value}"] = json
+            emit(
+                sessionId,
+                ArtifactContentStoredEvent(
+                    artifactId = slot.name,
+                    contentHash = storedHash,
+                    sessionId = sessionId,
+                    stageId = stageId,
+                ),
+            )
+        }
+    }
+
    private suspend fun emitLlmArtifacts(
        sessionId: SessionId,
        stageId: StageId,
@@ -1075,7 +1272,12 @@ abstract class SessionOrchestrator(
        if (isCancelled(sessionId)) return InferenceResult.Cancelled

        return try {
-            val response = withTimeout(timeoutMs) { provider.infer(request) }
+            val rawResponse = withTimeout(timeoutMs) { provider.infer(request) }
+            // Models frequently wrap a JSON artifact in a ```json … ``` markdown fence despite
+            // being told not to. The fence makes the artifact unparseable → validation fails.
+            // Strip an outer fence wrapper so the stored artifact, its CAS hash, the content
+            // cache, and validation all see the same clean payload.
+            val response = rawResponse.copy(text = stripCodeFence(rawResponse.text))
            log.debug(
                "[Orchestrator] inference done session={} stage={} tokens={} latencyMs={}",
                sessionId.value, stageId.value, response.tokensUsed, response.latencyMs,