feat(context): overlay sibling-stage writes onto repo retrieval
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 <noreply@anthropic.com>
This commit is contained in:
+35
@@ -10,6 +10,7 @@ import com.correx.core.context.model.TokenBudget
|
|||||||
import com.correx.core.context.model.EntryRole
|
import com.correx.core.context.model.EntryRole
|
||||||
import com.correx.core.events.events.ArtifactContentStoredEvent
|
import com.correx.core.events.events.ArtifactContentStoredEvent
|
||||||
import com.correx.core.events.events.FileWrittenEvent
|
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.ArtifactCreatedEvent
|
||||||
import com.correx.core.events.events.ArtifactRepairAttemptedEvent
|
import com.correx.core.events.events.ArtifactRepairAttemptedEvent
|
||||||
import com.correx.core.events.events.ArtifactRepairFailedEvent
|
import com.correx.core.events.events.ArtifactRepairFailedEvent
|
||||||
@@ -227,6 +228,40 @@ internal suspend fun SessionOrchestrator.fileWrittenManifest(sessionId: SessionI
|
|||||||
}.trimEnd()
|
}.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<RepoKnowledgeHit> {
|
||||||
|
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(
|
internal suspend fun SessionOrchestrator.emitStageCheckpoint(
|
||||||
sessionId: SessionId,
|
sessionId: SessionId,
|
||||||
stageId: StageId,
|
stageId: StageId,
|
||||||
|
|||||||
+16
-7
@@ -320,14 +320,23 @@ internal suspend fun SessionOrchestrator.buildContextualRepoEntries(
|
|||||||
return repoEntriesOrMapFloor(sessionId, recorded.hits, stagePrompt) + buildDocsCatalogEntry(sessionId)
|
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
|
val retriever = repoKnowledgeRetriever
|
||||||
?: return buildRepoMapEntries(sessionId, stagePrompt) + buildDocsCatalogEntry(sessionId)
|
val semanticHits = if (retriever == null) {
|
||||||
val hits = runCatching { retriever.retrieve(sessionId, stagePrompt, tuning.repoMapInjectTopK) }
|
emptyList()
|
||||||
.getOrElse { e ->
|
} else {
|
||||||
if (e is CancellationException) throw e
|
runCatching { retriever.retrieve(sessionId, stagePrompt, tuning.repoMapInjectTopK) }
|
||||||
log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message)
|
.getOrElse { e ->
|
||||||
emptyList()
|
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
|
// 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.
|
// 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))
|
if (hits.isNotEmpty()) emit(sessionId, RepoKnowledgeRetrievedEvent(sessionId, stageId, stagePrompt, hits))
|
||||||
|
|||||||
@@ -48,9 +48,18 @@ import com.correx.testing.fixtures.cyclePolicyMissingValidator
|
|||||||
import com.correx.testing.fixtures.inference.MockInferenceProvider
|
import com.correx.testing.fixtures.inference.MockInferenceProvider
|
||||||
import com.correx.testing.fixtures.transitions.TransitionFixtures
|
import com.correx.testing.fixtures.transitions.TransitionFixtures
|
||||||
import com.correx.testing.kernel.MockSessionEventReplayer
|
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.ArtifactCreatedEvent
|
||||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
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.coroutines.runBlocking
|
||||||
|
import kotlinx.datetime.Clock
|
||||||
import org.junit.jupiter.api.Assertions.assertFalse
|
import org.junit.jupiter.api.Assertions.assertFalse
|
||||||
import org.junit.jupiter.api.Assertions.assertTrue
|
import org.junit.jupiter.api.Assertions.assertTrue
|
||||||
import org.junit.jupiter.api.Test
|
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
|
@Test
|
||||||
fun `stage with empty needs adds no neededArtifact entries`(): Unit = runBlocking {
|
fun `stage with empty needs adds no neededArtifact entries`(): Unit = runBlocking {
|
||||||
val sessionId = SessionId("needs-test-2")
|
val sessionId = SessionId("needs-test-2")
|
||||||
|
|||||||
Reference in New Issue
Block a user