feat(acr): ACR Store 1 — content-hash observation cache for repo-file descriptors (#305)
Adds ObservationStore (core:context) keyed on (repoRoot, path), an in-memory default and a durable SqliteObservationStore (infrastructure:persistence), and wires it into the two SessionOrchestratorArtifacts call sites that re-derive a SourceDescriptor from CAS bytes on every call — a hit on matching content hash skips both the CAS read and the regex extraction. First slice of docs/plans/2026-07-21-acr-knowledge-accretion.md (build order: observations before fixes/plan-shapes). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HMbPmZZjcXhR2crU82zZ8S
This commit is contained in:
@@ -8,6 +8,8 @@ dependencies {
|
||||
implementation(project(":core:events"))
|
||||
implementation(project(":core:artifacts"))
|
||||
implementation(project(":core:sessions"))
|
||||
testImplementation "org.jetbrains.kotlin:kotlin-test"
|
||||
testImplementation "org.junit.jupiter:junit-jupiter"
|
||||
}
|
||||
|
||||
tasks.named("koverVerify").configure { enabled = false }
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
package com.correx.core.context.observation
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* ACR Store 1 (docs/plans/2026-07-21-acr-knowledge-accretion.md): a fact about a repo file,
|
||||
* keyed on (repoRoot, path) — task-agnostic, any future run benefits. [contentHash] is the
|
||||
* staleness check: a caller re-validates by hash before trusting [descriptorRender], per
|
||||
* invariant #9 (never re-derive from a re-observed environment when the prior observation still
|
||||
* holds). [descriptorRender] is [com.correx.core.sourcedesc.SourceDescriptor.render] output — the
|
||||
* only thing every current call site actually consumes, so that's what's stored, not the object.
|
||||
*/
|
||||
data class Observation(
|
||||
val repoRoot: String,
|
||||
val path: String,
|
||||
val contentHash: String,
|
||||
val descriptorRender: String?,
|
||||
val observedAtMs: Long,
|
||||
)
|
||||
|
||||
interface ObservationStore {
|
||||
suspend fun get(repoRoot: String, path: String): Observation?
|
||||
suspend fun put(observation: Observation)
|
||||
}
|
||||
|
||||
/** Default/test backend — process-lifetime only. Production wiring swaps in a durable store. */
|
||||
class InMemoryObservationStore : ObservationStore {
|
||||
private val entries = ConcurrentHashMap<Pair<String, String>, Observation>()
|
||||
|
||||
override suspend fun get(repoRoot: String, path: String): Observation? =
|
||||
entries[repoRoot to path]
|
||||
|
||||
override suspend fun put(observation: Observation) {
|
||||
entries[observation.repoRoot to observation.path] = observation
|
||||
}
|
||||
}
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
package com.correx.core.context.observation
|
||||
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
|
||||
class InMemoryObservationStoreTest {
|
||||
|
||||
@Test
|
||||
fun `put then get returns the stored observation`(): Unit = runBlocking {
|
||||
val store = InMemoryObservationStore()
|
||||
val obs = Observation("repoA", "src/Foo.kt", "hash1", "module=foo", 100L)
|
||||
|
||||
store.put(obs)
|
||||
|
||||
assertEquals(obs, store.get("repoA", "src/Foo.kt"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `same path in different repos does not collide`(): Unit = runBlocking {
|
||||
val store = InMemoryObservationStore()
|
||||
store.put(Observation("repoA", "src/Foo.kt", "hashA", "a", 1L))
|
||||
store.put(Observation("repoB", "src/Foo.kt", "hashB", "b", 2L))
|
||||
|
||||
assertEquals("hashA", store.get("repoA", "src/Foo.kt")?.contentHash)
|
||||
assertEquals("hashB", store.get("repoB", "src/Foo.kt")?.contentHash)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `put overwrites the prior observation for the same key`(): Unit = runBlocking {
|
||||
val store = InMemoryObservationStore()
|
||||
store.put(Observation("repoA", "src/Foo.kt", "hash1", "old", 1L))
|
||||
store.put(Observation("repoA", "src/Foo.kt", "hash2", "new", 2L))
|
||||
|
||||
assertEquals("hash2", store.get("repoA", "src/Foo.kt")?.contentHash)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `unknown key returns null`(): Unit = runBlocking {
|
||||
assertNull(InMemoryObservationStore().get("repoA", "nope"))
|
||||
}
|
||||
}
|
||||
+3
-1
@@ -7,6 +7,7 @@ import com.correx.core.inference.Tokenizer
|
||||
import com.correx.core.journal.DefaultDecisionJournalRepository
|
||||
import com.correx.core.artifacts.kind.ArtifactKindRegistry
|
||||
import com.correx.core.artifactstore.ArtifactStore
|
||||
import com.correx.core.context.observation.ObservationStore
|
||||
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
||||
import com.correx.core.events.events.ArtifactContentStoredEvent
|
||||
import com.correx.core.events.events.ClarificationAnswer
|
||||
@@ -111,7 +112,8 @@ class DefaultSessionOrchestrator(
|
||||
readyTaskCounter: ReadyTaskCounter? = null,
|
||||
taskClaimCoordinator: TaskClaimCoordinator? = null,
|
||||
tuning: OrchestrationTuning = OrchestrationTuning(),
|
||||
) : SessionOrchestrator(repositories, engines, artifactStore, decisionJournalRepository, artifactKindRegistry = artifactKindRegistry, repoKnowledgeRetriever = repoKnowledgeRetriever, readyTaskCounter = readyTaskCounter, taskClaimCoordinator = taskClaimCoordinator, tuning = tuning), ApprovalGateway {
|
||||
observationStore: ObservationStore? = null,
|
||||
) : SessionOrchestrator(repositories, engines, artifactStore, decisionJournalRepository, artifactKindRegistry = artifactKindRegistry, repoKnowledgeRetriever = repoKnowledgeRetriever, readyTaskCounter = readyTaskCounter, taskClaimCoordinator = taskClaimCoordinator, tuning = tuning, observationStore = observationStore), ApprovalGateway {
|
||||
override val tokenizer: Tokenizer? = tokenizer
|
||||
override val cancellations: ConcurrentHashMap<SessionId, AtomicBoolean> =
|
||||
ConcurrentHashMap<SessionId, AtomicBoolean>()
|
||||
|
||||
+2
@@ -8,6 +8,7 @@ import com.correx.core.artifacts.kind.KindInference
|
||||
import com.correx.core.artifactstore.ArtifactStore
|
||||
import com.correx.core.context.builder.ContextPackBuilder
|
||||
import com.correx.core.context.model.ContextPack
|
||||
import com.correx.core.context.observation.ObservationStore
|
||||
import com.correx.core.events.events.ClarificationAnswer
|
||||
import com.correx.core.events.events.ClarificationAnsweredEvent
|
||||
import com.correx.core.events.events.ClarificationRequestedEvent
|
||||
@@ -202,6 +203,7 @@ abstract class SessionOrchestrator(
|
||||
internal val readyTaskCounter: ReadyTaskCounter? = null,
|
||||
internal val taskClaimCoordinator: TaskClaimCoordinator? = null,
|
||||
internal val tuning: OrchestrationTuning = OrchestrationTuning(),
|
||||
internal val observationStore: ObservationStore? = null,
|
||||
) {
|
||||
internal val log = LoggerFactory.getLogger(this::class.java)
|
||||
internal val eventStore: EventStore = repositories.eventStore
|
||||
|
||||
+4
-5
@@ -207,6 +207,7 @@ internal suspend fun SessionOrchestrator.fileWrittenManifest(sessionId: SessionI
|
||||
.groupBy { it.path }
|
||||
.mapValues { (_, writes) -> writes.last() }
|
||||
if (latestWrites.isEmpty()) return null
|
||||
val repoRoot = workspacePolicy?.workspaceRoot?.toString().orEmpty()
|
||||
return buildString {
|
||||
appendLine(
|
||||
"Files written by the producing stage. Each line gives the authoritative CAS image and a " +
|
||||
@@ -215,9 +216,7 @@ internal suspend fun SessionOrchestrator.fileWrittenManifest(sessionId: SessionI
|
||||
)
|
||||
latestWrites.toSortedMap().forEach { (path, write) ->
|
||||
val hash = write.postImageHash ?: return@forEach
|
||||
val descriptor = artifactStore.get(ArtifactId(hash))
|
||||
?.let { describe(path, it).render() }
|
||||
?.takeIf { it.isNotBlank() }
|
||||
val descriptor = describeCached(repoRoot, path, hash)
|
||||
val stage = invToStage[write.invocationId]?.value
|
||||
append("- $path")
|
||||
stage?.let { append(" [by $it]") }
|
||||
@@ -250,13 +249,13 @@ internal suspend fun SessionOrchestrator.sessionWrittenHits(
|
||||
.groupBy { it.path }
|
||||
.mapValues { (_, writes) -> writes.last() }
|
||||
if (latestWrites.isEmpty()) return emptyList()
|
||||
val repoRoot = workspacePolicy?.workspaceRoot?.toString().orEmpty()
|
||||
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 descriptor = describeCached(repoRoot, write.path, hash)
|
||||
val text = if (descriptor != null) "${write.path}: $descriptor" else write.path
|
||||
RepoKnowledgeHit(path = write.path, text = text, score = 1.0f)
|
||||
}
|
||||
|
||||
+35
@@ -0,0 +1,35 @@
|
||||
package com.correx.core.kernel.orchestration
|
||||
|
||||
import com.correx.core.context.observation.Observation
|
||||
import com.correx.core.events.types.ArtifactId
|
||||
import com.correx.core.sourcedesc.describe
|
||||
|
||||
/**
|
||||
* [describe]-and-render a written file's CAS bytes, short-circuited by ACR Store 1
|
||||
* (docs/plans/2026-07-21-acr-knowledge-accretion.md) when a prior observation for this
|
||||
* (repoRoot, path) already carries this exact content hash — skips both the CAS read and the
|
||||
* regex extraction. Falls back to a fresh [describe] + records the result for next time. No-op
|
||||
* (always fresh, never recorded) when no [SessionOrchestrator.observationStore] is wired.
|
||||
*/
|
||||
internal suspend fun SessionOrchestrator.describeCached(
|
||||
repoRoot: String,
|
||||
path: String,
|
||||
hash: String,
|
||||
): String? {
|
||||
val store = observationStore
|
||||
val cached = store?.get(repoRoot, path)
|
||||
if (cached != null && cached.contentHash == hash) return cached.descriptorRender
|
||||
val rendered = artifactStore.get(ArtifactId(hash))
|
||||
?.let { describe(path, it).render() }
|
||||
?.takeIf { it.isNotBlank() }
|
||||
store?.put(
|
||||
Observation(
|
||||
repoRoot = repoRoot,
|
||||
path = path,
|
||||
contentHash = hash,
|
||||
descriptorRender = rendered,
|
||||
observedAtMs = System.currentTimeMillis(),
|
||||
),
|
||||
)
|
||||
return rendered
|
||||
}
|
||||
Reference in New Issue
Block a user