From 82a4395f34ec1767108394b04d10cadfc24ee0ff Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 20 Jul 2026 14:38:34 +0400 Subject: [PATCH] feat(context): overlay sibling-stage writes onto repo retrieval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Files written by earlier stages this session aren't in the session-start repo map and were never embedded, so semantic retrieval is structurally blind to them. Overlay each stage's FileWrittenEvent post-images as deterministic hits (score 1.0) leading the semantic hits, deduped by path — no reindex, no embed. Descriptors derive via the comment-free sourcedesc describe() so agent-written content can't inject prose. Co-Authored-By: Claude Opus 4.8 --- .../SessionOrchestratorArtifacts.kt | 35 ++++++++++ .../SessionOrchestratorContext.kt | 23 ++++-- .../test/kotlin/NeedsArtifactInjectionTest.kt | 70 +++++++++++++++++++ 3 files changed, 121 insertions(+), 7 deletions(-) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorArtifacts.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorArtifacts.kt index 777120f7..652d12ba 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorArtifacts.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorArtifacts.kt @@ -10,6 +10,7 @@ import com.correx.core.context.model.TokenBudget import com.correx.core.context.model.EntryRole import com.correx.core.events.events.ArtifactContentStoredEvent import com.correx.core.events.events.FileWrittenEvent +import com.correx.core.events.events.RepoKnowledgeHit import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.ArtifactRepairAttemptedEvent import com.correx.core.events.events.ArtifactRepairFailedEvent @@ -227,6 +228,40 @@ internal suspend fun SessionOrchestrator.fileWrittenManifest(sessionId: SessionI }.trimEnd() } +/** + * Files written earlier THIS session by OTHER stages, projected as deterministic retrieval hits so + * a successor stage discovers sibling output that the session-start repo map (a snapshot) can never + * contain and the embedder can't rank (it was never indexed). Latest post-image per path, capped, + * most-recent first; each carries the comment-free [describe] descriptor over its recorded CAS bytes + * (untrusted navigation; CAS hash authoritative). Score 1.0 — this is deterministic grounding, not + * cosine similarity — so it survives the retriever's floor and the useful-hit filter. The current + * stage's own writes are excluded (retry-repair state and the file-written manifest cover those). + * Pure projection over recorded events + CAS, replay-safe. + */ +internal suspend fun SessionOrchestrator.sessionWrittenHits( + sessionId: SessionId, + currentStageId: StageId, +): List { + val events = eventStore.read(sessionId) + val invToStage = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent } + .associate { it.invocationId to it.stageId } + val latestWrites = events.mapNotNull { it.payload as? FileWrittenEvent } + .filter { it.postImageHash != null && invToStage[it.invocationId].let { s -> s != null && s != currentStageId } } + .groupBy { it.path } + .mapValues { (_, writes) -> writes.last() } + if (latestWrites.isEmpty()) return emptyList() + return latestWrites.values + .sortedByDescending { it.timestampMs } + .take(tuning.repoMapInjectTopK) + .mapNotNull { write -> + val hash = write.postImageHash ?: return@mapNotNull null + val descriptor = artifactStore.get(ArtifactId(hash))?.let { describe(write.path, it).render() } + ?.takeIf { it.isNotBlank() } + val text = if (descriptor != null) "${write.path}: $descriptor" else write.path + RepoKnowledgeHit(path = write.path, text = text, score = 1.0f) + } +} + internal suspend fun SessionOrchestrator.emitStageCheckpoint( sessionId: SessionId, stageId: StageId, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt index 2452038d..2be06a82 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorContext.kt @@ -320,14 +320,23 @@ internal suspend fun SessionOrchestrator.buildContextualRepoEntries( return repoEntriesOrMapFloor(sessionId, recorded.hits, stagePrompt) + buildDocsCatalogEntry(sessionId) } + // Sibling-stage output written earlier this session isn't in the session-start repo map and was + // never embedded, so semantic retrieval is structurally blind to it. Overlay it as deterministic + // hits (score 1.0) that lead the semantic hits, deduped by path — no reindex, no embed. + val writtenHits = sessionWrittenHits(sessionId, stageId) val retriever = repoKnowledgeRetriever - ?: return buildRepoMapEntries(sessionId, stagePrompt) + buildDocsCatalogEntry(sessionId) - val hits = runCatching { retriever.retrieve(sessionId, stagePrompt, tuning.repoMapInjectTopK) } - .getOrElse { e -> - if (e is CancellationException) throw e - log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message) - emptyList() - } + val semanticHits = if (retriever == null) { + emptyList() + } else { + runCatching { retriever.retrieve(sessionId, stagePrompt, tuning.repoMapInjectTopK) } + .getOrElse { e -> + if (e is CancellationException) throw e + log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message) + emptyList() + } + } + val writtenPaths = writtenHits.mapTo(mutableSetOf()) { it.path } + val hits = writtenHits + semanticHits.filterNot { it.path in writtenPaths } // Record the raw retrieval for audit (a degenerate all-zero result is itself the signal that // the embedder is noop / the index is empty), but only useful hits ever reach the agent. if (hits.isNotEmpty()) emit(sessionId, RepoKnowledgeRetrievedEvent(sessionId, stageId, stagePrompt, hits)) diff --git a/testing/integration/src/test/kotlin/NeedsArtifactInjectionTest.kt b/testing/integration/src/test/kotlin/NeedsArtifactInjectionTest.kt index 3183c1da..bafa82ef 100644 --- a/testing/integration/src/test/kotlin/NeedsArtifactInjectionTest.kt +++ b/testing/integration/src/test/kotlin/NeedsArtifactInjectionTest.kt @@ -48,9 +48,18 @@ import com.correx.testing.fixtures.cyclePolicyMissingValidator import com.correx.testing.fixtures.inference.MockInferenceProvider import com.correx.testing.fixtures.transitions.TransitionFixtures import com.correx.testing.kernel.MockSessionEventReplayer +import com.correx.core.approvals.Tier import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.FileWrittenEvent +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.ToolRequest +import com.correx.core.events.types.EventId +import com.correx.core.events.types.ToolInvocationId import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.Test @@ -197,6 +206,67 @@ class NeedsArtifactInjectionTest { ) } + /** + * Slice 6: a file written by an EARLIER stage this session must surface to a later stage as a + * deterministic retrieval hit (a "relevantFiles" entry naming the path), even with no embedder — + * the session-start repo map can't contain it and semantic retrieval never indexed it. + */ + @Test + fun `file written by an earlier stage surfaces to a later stage as a relevant file`(): Unit = runBlocking { + val sessionId = SessionId("write-overlay-1") + val stageA = StageId("A") + val stageB = StageId("B") + val inv = ToolInvocationId("inv-A-1") + val siblingPath = "src/Sibling.kt" + + // Seed stage A's write into the log before the run; sessionWrittenHits projects it for stage B. + listOf( + ToolInvocationRequestedEvent( + invocationId = inv, sessionId = sessionId, stageId = stageA, + toolName = "file_write", tier = Tier.T2, + request = ToolRequest( + invocationId = inv, sessionId = sessionId, stageId = stageA, + toolName = "file_write", parameters = mapOf("path" to siblingPath), + ), + ), + FileWrittenEvent( + invocationId = inv, sessionId = sessionId, path = siblingPath, + postImageHash = "deadbeef", preExisted = false, timestampMs = 1, + ), + ).forEach { payload -> + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(java.util.UUID.randomUUID().toString()), + sessionId = sessionId, timestamp = Clock.System.now(), + schemaVersion = 1, causationId = null, correlationId = null, + ), + payload = payload, + ), + ) + } + + // stageC makes stageB non-terminal so stageB actually gets a context pack built (terminal + // stages are entered as end-markers without assembling context). + val stageC = StageId("C") + val graph = WorkflowGraph( + id = "write-overlay-graph", + stages = mapOf(stageA to StageConfig(), stageB to StageConfig(), stageC to StageConfig()), + transitions = setOf( + TransitionEdge(id = TransitionId("t1"), from = stageA, to = stageB, condition = { true }), + TransitionEdge(id = TransitionId("t2"), from = stageB, to = stageC, condition = { true }), + ), + start = stageA, + ) + buildOrchestrator(InferenceFixtures.fixedRouter()).run(sessionId, graph, defaultConfig) + + val relevant = capturedEntries.filter { it.sourceType == "relevantFiles" } + assertTrue( + relevant.any { it.content.contains(siblingPath) }, + "Expected a relevantFiles entry naming $siblingPath, got: ${relevant.map { it.content }}", + ) + } + @Test fun `stage with empty needs adds no neededArtifact entries`(): Unit = runBlocking { val sessionId = SessionId("needs-test-2")