diff --git a/core/context/src/main/kotlin/com/correx/core/context/compression/ContextClassifier.kt b/core/context/src/main/kotlin/com/correx/core/context/compression/ContextClassifier.kt index c41d547c..a1113306 100644 --- a/core/context/src/main/kotlin/com/correx/core/context/compression/ContextClassifier.kt +++ b/core/context/src/main/kotlin/com/correx/core/context/compression/ContextClassifier.kt @@ -35,8 +35,13 @@ class ContextClassifier { // shredded the JSON to a bare id + protected path, giving the model amnesia about what it // had already done → it re-issued the same call and looped. Structured = format-compress // ok, never pruned. + // "orchestratorCorrection" is an in-loop corrective USER turn authored by the + // orchestrator (invalid emit_artifact, premature stage_complete, read loop, missing write). + // Pruning it as freeform prose can shred the tool name or the negation that makes it + // actionable, which leaves the model with a vague complaint instead of an instruction. val STRUCTURED_SOURCES = setOf( "toolLog", "artifact", "config", "structured", "steeringNote", "assistantToolCall", + "orchestratorCorrection", ) } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt index ae6210c0..d11b8f57 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/JournalCompactionService.kt @@ -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)) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt index 0b587dda..ab238a9d 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorExecution.kt @@ -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, +) diff --git a/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt index a48615e4..3d1f4fdf 100644 --- a/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt +++ b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/JournalCompactionServiceTest.kt @@ -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> { + val blobs = mutableMapOf() + 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() + val svc = JournalCompactionService(store, { prompts += it; "SUMMARY-${prompts.size}" }, { 100 }) + + val first = stateWithRecords(makeRecord(1, DecisionKind.INTENT)) + val emitted = mutableListOf() + 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() + 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 }) diff --git a/core/talkie/src/main/kotlin/com/correx/core/talkie/TalkieFacade.kt b/core/talkie/src/main/kotlin/com/correx/core/talkie/TalkieFacade.kt index a7a1dcbb..b785eade 100644 --- a/core/talkie/src/main/kotlin/com/correx/core/talkie/TalkieFacade.kt +++ b/core/talkie/src/main/kotlin/com/correx/core/talkie/TalkieFacade.kt @@ -217,16 +217,16 @@ class DefaultTalkieFacade( emitIdeasCaptured(sessionId, ideas.ideas) } - // ponytail: STEERING launders the user's text through the router LLM (the `content` above) - // before injecting it as a note. For a clear instruction that's an extra inference that can - // distort intent; the reformulation only earns its cost when the input is terse/context- - // dependent. Upgrade path: inject the raw (validated) input directly and skip the rewrite - // unless a heuristic flags the message as too short/ambiguous to stand alone. - val steeringEmitted = mode == ChatMode.STEERING && rawContent.isNotBlank() + // The steering note carries the operator's OWN text, not the router's paraphrase of it. + // Routing it through inference first let a lossy rewrite acquire the authority of a user + // directive — negations, filenames, constraints and priority could all change before the + // orchestrator saw them. The router turn above is still produced and shown to the operator + // as conversational acknowledgement; it is just not the mandate. + val steeringEmitted = mode == ChatMode.STEERING && input.isNotBlank() if (steeringEmitted) { - val validationError = validateSteering?.invoke(content) + val validationError = validateSteering?.invoke(input) if (validationError == null) { - emitSteeringNote(sessionId, content, effectiveStageId) + emitSteeringNote(sessionId, input, effectiveStageId) } } diff --git a/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt b/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt index de3d4449..06c654ba 100644 --- a/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt +++ b/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt @@ -5,8 +5,10 @@ import com.correx.core.context.builder.RequiredContextOverflowException import com.correx.core.context.model.ContextBucket import com.correx.core.context.model.ContextEntry import com.correx.core.context.model.ContextLayer +import com.correx.core.context.model.EntryRole import com.correx.core.context.model.TokenBudget import com.correx.core.events.types.ContextEntryId +import com.correx.core.inference.PromptRenderer import com.correx.core.events.types.ContextPackId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId @@ -42,6 +44,55 @@ class DefaultContextPackBuilderTest { tokenEstimate = tokens ) + @Test + fun `an orchestrator correction reaches the rendered prompt as the last user turn`() = + kotlinx.coroutines.runBlocking { + // The nudge used to be built as a toolResult with a fresh sourceId, so it had no + // matching assistantToolCall and reconcileToolPairs() deleted it: the orchestrator + // "corrected" the model and the correction never reached the provider. Assert on the + // RENDERED prompt, not the entry list — that is where the defect was invisible. + val entries = listOf( + typedEntry("sys", ContextLayer.L0, 100, "systemPrompt"), + typedEntry("task", ContextLayer.L1, 50, "agentPrompt"), + ContextEntry( + id = ContextEntryId("call1"), + layer = ContextLayer.L2, + content = "{\"tool\":\"file_read\"}", + sourceType = "assistantToolCall", + sourceId = "inv-1", + tokenEstimate = 40, + role = EntryRole.ASSISTANT, + ), + ContextEntry( + id = ContextEntryId("res1"), + layer = ContextLayer.L2, + content = "file contents", + sourceType = "toolResult", + sourceId = "inv-1", + tokenEstimate = 40, + role = EntryRole.TOOL, + ), + ContextEntry( + id = ContextEntryId("nudge"), + layer = ContextLayer.L2, + content = "STOP reading. You MUST now call file_write.", + sourceType = "orchestratorCorrection", + sourceId = "corr-1", + tokenEstimate = 20, + role = EntryRole.USER, + bucket = ContextBucket.REQUIRED, + ), + ) + val pack = builder.build(packId, sessionId, stageId, entries, TokenBudget(limit = 4096)) + val messages = PromptRenderer.render(pack) + val last = messages.last() + assertEquals("user", last.role) + assertTrue( + last.content.contains("STOP reading"), + "correction must survive to the prompt; got: " + messages.map { it.role to it.content }, + ) + } + @Test fun `oversized tool results from a stage tool loop are trimmed to budget`() = kotlinx.coroutines.runBlocking { // Live repro (2026-06-11): analyst stage with three file_read results of ~5.6k/11.9k/10.7k diff --git a/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt b/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt index 8b71a8a4..77ccdfe7 100644 --- a/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt +++ b/testing/deterministic/src/test/kotlin/TalkieFacadeTest.kt @@ -129,7 +129,8 @@ class TalkieFacadeTest { assertEquals(com.correx.core.events.events.ChatTurnRole.USER, chatEvents[0].role) assertEquals("inference response", chatEvents[1].content) assertEquals(com.correx.core.events.events.ChatTurnRole.ROUTER, chatEvents[1].role) - assertEquals("inference response", steeringEvents[0].content) + // The steering note is the operator's raw text, never the router's paraphrase of it. + assertEquals("steer this way", steeringEvents[0].content) } @Test @@ -158,7 +159,7 @@ class TalkieFacadeTest { val payloads = mockStore.appendedEvents.map { it.payload } val steeringEvents = payloads.filterIsInstance() assertEquals(1, steeringEvents.size) - assertEquals("steering response", steeringEvents[0].content) + assertEquals("Hello!", steeringEvents[0].content) } // --------------------------------------------------------------------------