fix(kernel,talkie,context): three instruction-corruption fixes from the 2026-08-26 context audit

Each of the three lost or rewrote an instruction before the model saw it. 1. Orphan corrective nudges. pushBack() and the final tools-disabled emission built their nudge as a toolResult with a fresh sourceId, so it had no matching assistantToolCall and reconcileToolPairs() deleted it — the orchestrator believed it had corrected the model while the correction never reached the prompt. Affected the invalid-emit_artifact, premature-stage_complete, missing-write, read-loop, rejection-loop and final-JSON nudges. They are now USER turns (sourceType orchestratorCorrection, REQUIRED bucket, STRUCTURED in ContextClassifier so pruning cannot shred them), appended last so the builder's positional ordinal puts them at the end of the transcript. A superseded nudge is dropped rather than stacking stale demands. 2. Steering laundered through the router. The SteeringNoteAddedEvent carried the router model's paraphrase of the operator's message, not the message — negations, filenames, constraints and priority could change before the orchestrator saw them. It now carries the raw input; the router turn is still produced and shown as conversational acknowledgement, it is just not the mandate. Resolves the ponytail: note at TalkieFacade.kt:220. 3. Journal compaction erased its own history. compactIfNeeded() summarized only state.records while the reducer overwrote summaryArtifactId and dropped covered records, so the second compaction lost everything the first had preserved — and a low-salience-only batch replaced it with the "(no high-salience decisions)" fallback. Compaction is cumulative now, and a blank or fallback-only rewrite never replaces real history. Tests: rendered-prompt regression for (1) — verified to fail against the old toolResult shape — updated steering expectations for (2), two cumulative-compaction tests for (3). Full build green, 1831 tests, 0 failures. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-27 01:33:16 +04:00
parent d892587420
commit 519290368f
7 changed files with 180 additions and 31 deletions
@@ -30,16 +30,38 @@ class JournalCompactionService(
        val highRecords = state.records.filter { it.kind.salience() == Salience.HIGH }
        val lowCount = state.records.count { it.kind.salience() == Salience.LOW }

-        val summaryText = if (highRecords.isEmpty()) {
+        // Compaction is CUMULATIVE. The reducer overwrites summaryArtifactId and drops every
+        // covered record, so a summary built from `state.records` alone erases the previous
+        // summary on the second compaction: the intent, approvals and steering it carried are
+        // neither an input here nor retained as a predecessor. Feed the prior summary back in.
+        val priorSummary = state.summaryArtifactId
+            ?.let { artifactStore.get(it) }
+            ?.toString(Charsets.UTF_8)
+            ?.takeIf { it.isNotBlank() }
+
+        val summaryText = if (highRecords.isEmpty() && priorSummary == null) {
            "(no high-salience decisions to summarize)"
+        } else if (highRecords.isEmpty()) {
+            // Nothing new worth keeping — carry the prior summary forward untouched rather than
+            // replacing it with the fallback text.
+            priorSummary!!
        } else {
            val prompt = buildString {
                appendLine("Summarize the following key decisions concisely (≤200 words).")
                appendLine("Preserve all user intent, approvals, steering, and failures.")
                appendLine()
+                priorSummary?.let {
+                    appendLine("Summary of earlier decisions (already compacted — preserve its content):")
+                    appendLine(it)
+                    appendLine()
+                    appendLine("New decisions since then:")
+                }
                highRecords.forEach { appendLine("- [${it.kind}] ${it.summary}") }
            }
-            summarize(prompt)
+            // A blank or fallback-only rewrite must never replace real history.
+            summarize(prompt).takeIf { it.isNotBlank() }
+                ?: priorSummary
+                ?: "(no high-salience decisions to summarize)"
        }

        val summaryArtifactId = artifactStore.put(summaryText.toByteArray(Charsets.UTF_8))
@@ -2,6 +2,7 @@ package com.correx.core.kernel.orchestration

 import com.correx.core.context.builder.RequiredContextOverflowException
 import com.correx.core.context.model.ContextEntry
+import com.correx.core.context.model.ContextBucket
 import com.correx.core.context.model.ContextLayer
 import com.correx.core.context.model.TokenBudget
 import com.correx.core.context.model.EntryRole
@@ -351,18 +352,19 @@ internal suspend fun SessionOrchestrator.executeStage(
        return completedIds.isEmpty()
    }

-    // Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS).
+    // Append a corrective USER turn and re-run inference (bounded by MAX_TOOL_ROUNDS).
    // Returns the new result rather than mutating inferenceResult, to preserve smart casts.
+    //
+    // NOT a toolResult: these nudges are orchestrator-authored, so they have no matching
+    // assistantToolCall, and reconcileToolPairs() drops every tool result whose call ID is absent
+    // (see its doc). The orchestrator believed it had corrected the model while the correction
+    // never reached the prompt. A USER turn is what this actually is — the operator side of the
+    // loop telling the model what to do next — and it survives reconciliation. Appended last so
+    // the builder's positional ordinal stamp puts it at the end of the transcript, and only one
+    // correction is ever live: a superseded nudge is dropped rather than stacking stale demands.
    suspend fun pushBack(nudge: String, forceWriteOnly: Boolean = false): 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,
-        )
+        accumulatedEntries = accumulatedEntries.filterNot { it.sourceType == CORRECTION_SOURCE_TYPE } +
+            correctionEntry(nudge)
        currentContext = contextPackBuilder.build(
            id = ContextPackId(UUID.randomUUID().toString()),
            sessionId = sessionId,
@@ -582,15 +584,8 @@ internal suspend fun SessionOrchestrator.executeStage(
        if (needsCleanEmission && !isCancelled(sessionId)) {
            val nudge = "Stop calling tools. Output the required '${llmEmittedSlots.first().name.value}' " +
                "artifact now as a single JSON object matching the schema — no tool calls, no commentary."
-            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,
-            )
+            accumulatedEntries = accumulatedEntries.filterNot { it.sourceType == CORRECTION_SOURCE_TYPE } +
+                correctionEntry(nudge)
            currentContext = contextPackBuilder.build(
                id = ContextPackId(UUID.randomUUID().toString()),
                sessionId = sessionId,
@@ -687,3 +682,19 @@ internal suspend fun SessionOrchestrator.executeStage(
        }
    }
 }
+
+// Orchestrator-authored corrections. STRUCTURED in ContextClassifier (never token-pruned) and
+// REQUIRED so a correction is never traded away for budget — the whole point of a nudge is that
+// the next inference sees it.
+internal const val CORRECTION_SOURCE_TYPE = "orchestratorCorrection"
+
+private suspend fun SessionOrchestrator.correctionEntry(nudge: String) = ContextEntry(
+    id = ContextEntryId(UUID.randomUUID().toString()),
+    layer = ContextLayer.L2,
+    sourceType = CORRECTION_SOURCE_TYPE,
+    sourceId = UUID.randomUUID().toString(),
+    content = nudge,
+    tokenEstimate = estimateTokens(nudge),
+    role = EntryRole.USER,
+    bucket = ContextBucket.REQUIRED,
+)
@@ -33,6 +33,65 @@ class JournalCompactionServiceTest {
    private fun makeRecord(seq: Long, kind: DecisionKind) =
        DecisionRecord(sequence = seq, kind = kind, summary = "summary of $kind")

+    // A store that really round-trips, so a second compaction can read back the first summary.
+    private fun recordingArtifactStore(): Pair<ArtifactStore, MutableMap<TypeId, ByteArray>> {
+        val blobs = mutableMapOf<TypeId, ByteArray>()
+        val store = object : ArtifactStore {
+            override suspend fun put(bytes: ByteArray): TypeId {
+                val id = TypeId("artifact-${blobs.size + 1}")
+                blobs[id] = bytes
+                return id
+            }
+            override suspend fun get(id: TypeId): ByteArray? = blobs[id]
+            override suspend fun flushBefore(commit: suspend () -> Unit) = commit()
+        }
+        return store to blobs
+    }
+
+    @Test
+    fun `second compaction feeds the prior summary back into the prompt`(): Unit = runBlocking {
+        val (store, blobs) = recordingArtifactStore()
+        val prompts = mutableListOf<String>()
+        val svc = JournalCompactionService(store, { prompts += it; "SUMMARY-${prompts.size}" }, { 100 })
+
+        val first = stateWithRecords(makeRecord(1, DecisionKind.INTENT))
+        val emitted = mutableListOf<EventPayload>()
+        assertTrue(svc.compactIfNeeded(SessionId("s1"), first, 500) { emitted += it })
+        val firstId = (emitted.single() as JournalCompactedEvent).summaryArtifactId
+
+        // Reducer semantics: covered records are gone, summaryArtifactId now points at SUMMARY-1.
+        val second = DecisionJournalState(
+            records = listOf(makeRecord(2, DecisionKind.INTENT)),
+            compactedThroughSequence = 1,
+            summaryArtifactId = firstId,
+        )
+        emitted.clear()
+        assertTrue(svc.compactIfNeeded(SessionId("s1"), second, 500) { emitted += it })
+
+        assertTrue(prompts[1].contains("SUMMARY-1"), "prior summary must be an input: ${prompts[1]}")
+        val kept = blobs[(emitted.single() as JournalCompactedEvent).summaryArtifactId]!!
+        assertEquals("SUMMARY-2", kept.toString(Charsets.UTF_8))
+    }
+
+    @Test
+    fun `a low-salience-only batch carries the prior summary forward instead of erasing it`():
+        Unit = runBlocking {
+        val (store, blobs) = recordingArtifactStore()
+        val priorId = store.put("EARLIER DECISIONS".toByteArray(Charsets.UTF_8))
+        val svc = JournalCompactionService(store, { "should not be called" }, { 100 })
+
+        val state = DecisionJournalState(
+            records = listOf(makeRecord(2, DecisionKind.TRANSITION)),
+            compactedThroughSequence = 1,
+            summaryArtifactId = priorId,
+        )
+        val emitted = mutableListOf<EventPayload>()
+        assertTrue(svc.compactIfNeeded(SessionId("s1"), state, 500) { emitted += it })
+
+        val kept = blobs[(emitted.single() as JournalCompactedEvent).summaryArtifactId]!!
+        assertEquals("EARLIER DECISIONS", kept.toString(Charsets.UTF_8))
+    }
+
    @Test
    fun `returns false when token estimate is below threshold`(): Unit = runBlocking {
        val svc = JournalCompactionService(fakeArtifactStore(), { it }, tokenThreshold = { 2000 })