From 5df35879eb99e89dafdac814a2c4b24a7915b7eb Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 21 Jul 2026 17:37:25 +0400 Subject: [PATCH] =?UTF-8?q?feat(acr):=20ACR=20Store=201=20=E2=80=94=20cont?= =?UTF-8?q?ent-hash=20observation=20cache=20for=20repo-file=20descriptors?= =?UTF-8?q?=20(#305)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01HMbPmZZjcXhR2crU82zZ8S --- .../kotlin/com/correx/apps/server/Main.kt | 2 + core/context/build.gradle | 2 + .../context/observation/ObservationStore.kt | 36 +++++++++ .../InMemoryObservationStoreTest.kt | 43 +++++++++++ .../DefaultSessionOrchestrator.kt | 4 +- .../orchestration/SessionOrchestrator.kt | 2 + .../SessionOrchestratorArtifacts.kt | 9 +-- .../SessionOrchestratorObservation.kt | 35 +++++++++ infrastructure/build.gradle | 1 + infrastructure/persistence/build.gradle | 2 + .../observation/SqliteObservationStore.kt | 76 +++++++++++++++++++ .../observation/SqliteObservationStoreTest.kt | 37 +++++++++ .../infrastructure/InfrastructureModule.kt | 8 ++ 13 files changed, 251 insertions(+), 6 deletions(-) create mode 100644 core/context/src/main/kotlin/com/correx/core/context/observation/ObservationStore.kt create mode 100644 core/context/src/test/kotlin/com/correx/core/context/observation/InMemoryObservationStoreTest.kt create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorObservation.kt create mode 100644 infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStore.kt create mode 100644 infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStoreTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 6f04420a..8f433214 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -390,6 +390,7 @@ fun main() { semanticReviewer = SemanticReviewerImpl(inferenceRouter), ) val decisionJournalRepository = InfrastructureModule.createDecisionJournalRepository(eventStore) + val observationStore = InfrastructureModule.createObservationStore() val defaultOrchestrationConfig = OrchestrationConfig( sandboxRoot = sandboxRoot, defaultSystemPromptPath = toolsConfig.defaultSystemPromptPath, @@ -496,6 +497,7 @@ fun main() { compactionService = journalCompactionService, artifactKindRegistry = artifactKindRegistry, repoKnowledgeRetriever = repoKnowledgeRetriever, + observationStore = observationStore, readyTaskCounter = com.correx.apps.server.tasks.ProjectReadyTaskCounter(taskService), taskClaimCoordinator = com.correx.apps.server.tasks.DefaultTaskClaimCoordinator( taskService, diff --git a/core/context/build.gradle b/core/context/build.gradle index 707dc9a5..01308924 100644 --- a/core/context/build.gradle +++ b/core/context/build.gradle @@ -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 } diff --git a/core/context/src/main/kotlin/com/correx/core/context/observation/ObservationStore.kt b/core/context/src/main/kotlin/com/correx/core/context/observation/ObservationStore.kt new file mode 100644 index 00000000..97716141 --- /dev/null +++ b/core/context/src/main/kotlin/com/correx/core/context/observation/ObservationStore.kt @@ -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, 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 + } +} diff --git a/core/context/src/test/kotlin/com/correx/core/context/observation/InMemoryObservationStoreTest.kt b/core/context/src/test/kotlin/com/correx/core/context/observation/InMemoryObservationStoreTest.kt new file mode 100644 index 00000000..960f2089 --- /dev/null +++ b/core/context/src/test/kotlin/com/correx/core/context/observation/InMemoryObservationStoreTest.kt @@ -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")) + } +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 15f82d2d..3d871040 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -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 = ConcurrentHashMap() diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 28c1f6f4..4da4b50d 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -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 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 652d12ba..e212f9f9 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 @@ -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) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorObservation.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorObservation.kt new file mode 100644 index 00000000..800512c3 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestratorObservation.kt @@ -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 +} diff --git a/infrastructure/build.gradle b/infrastructure/build.gradle index 98e87cb9..ca8ad2b1 100644 --- a/infrastructure/build.gradle +++ b/infrastructure/build.gradle @@ -12,6 +12,7 @@ dependencies { implementation project(":core:approvals") implementation project(":core:sessions") implementation project(":core:config") + implementation project(":core:context") implementation project(":infrastructure:inference") implementation project(":infrastructure:inference:commons") implementation project(":infrastructure:inference:llama_cpp") diff --git a/infrastructure/persistence/build.gradle b/infrastructure/persistence/build.gradle index 0e0af0e6..509a33ad 100644 --- a/infrastructure/persistence/build.gradle +++ b/infrastructure/persistence/build.gradle @@ -10,11 +10,13 @@ dependencies { implementation(project(":core:sessions")) implementation(project(":core:artifacts")) implementation(project(":core:artifacts-store")) + implementation(project(":core:context")) implementation "org.xerial:sqlite-jdbc" implementation "org.slf4j:slf4j-api:2.0.16" testImplementation(testFixtures(project(":testing:contracts"))) testImplementation(project(":testing:fixtures")) testImplementation "org.junit.jupiter:junit-jupiter" + testImplementation "org.jetbrains.kotlin:kotlin-test" } tasks.named("koverVerify").configure { enabled = false } diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStore.kt new file mode 100644 index 00000000..5aea3883 --- /dev/null +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStore.kt @@ -0,0 +1,76 @@ +package com.correx.infrastructure.persistence.observation + +import com.correx.core.context.observation.Observation +import com.correx.core.context.observation.ObservationStore +import com.correx.infrastructure.persistence.util.JDBCHelper.transaction +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.sql.Connection + +/** + * Durable backend for ACR Store 1 (docs/plans/2026-07-21-acr-knowledge-accretion.md), so + * observations survive a server restart the way the event log itself does. One row per + * (repo_root, path); a `put` overwrites the prior observation for that key outright — this store + * only ever holds the latest fact, staleness is resolved by the caller comparing content_hash. + */ +class SqliteObservationStore(private val connection: Connection) : ObservationStore { + + init { + connection.createStatement().use { stmt -> + stmt.execute( + """ + CREATE TABLE IF NOT EXISTS observations ( + repo_root TEXT NOT NULL, + path TEXT NOT NULL, + content_hash TEXT NOT NULL, + descriptor_render TEXT, + observed_at_ms INTEGER NOT NULL, + PRIMARY KEY (repo_root, path) + ) + """.trimIndent(), + ) + } + } + + override suspend fun get(repoRoot: String, path: String): Observation? = withContext(Dispatchers.IO) { + connection.prepareStatement( + "SELECT content_hash, descriptor_render, observed_at_ms FROM observations " + + "WHERE repo_root = ? AND path = ?", + ).use { stmt -> + stmt.setString(1, repoRoot) + stmt.setString(2, path) + stmt.executeQuery().use { rs -> + if (!rs.next()) return@withContext null + Observation( + repoRoot = repoRoot, + path = path, + contentHash = rs.getString("content_hash"), + descriptorRender = rs.getString("descriptor_render"), + observedAtMs = rs.getLong("observed_at_ms"), + ) + } + } + } + + @Suppress("MagicNumber") // JDBC positional parameter indices, not data + override suspend fun put(observation: Observation) = withContext(Dispatchers.IO) { + connection.transaction { + connection.prepareStatement( + "INSERT INTO observations (repo_root, path, content_hash, descriptor_render, observed_at_ms) " + + "VALUES (?, ?, ?, ?, ?) " + + "ON CONFLICT(repo_root, path) DO UPDATE SET " + + "content_hash = excluded.content_hash, " + + "descriptor_render = excluded.descriptor_render, " + + "observed_at_ms = excluded.observed_at_ms", + ).use { stmt -> + stmt.setString(1, observation.repoRoot) + stmt.setString(2, observation.path) + stmt.setString(3, observation.contentHash) + stmt.setString(4, observation.descriptorRender) + stmt.setLong(5, observation.observedAtMs) + stmt.executeUpdate() + } + } + Unit + } +} diff --git a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStoreTest.kt b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStoreTest.kt new file mode 100644 index 00000000..3490344d --- /dev/null +++ b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/observation/SqliteObservationStoreTest.kt @@ -0,0 +1,37 @@ +package com.correx.infrastructure.persistence.observation + +import com.correx.core.context.observation.Observation +import kotlinx.coroutines.runBlocking +import java.sql.DriverManager +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class SqliteObservationStoreTest { + + private fun store() = SqliteObservationStore(DriverManager.getConnection("jdbc:sqlite::memory:")) + + @Test + fun `put then get round-trips through sqlite`(): Unit = runBlocking { + val store = store() + store.put(Observation("repoA", "src/Foo.kt", "hash1", "module=foo", 100L)) + + val fetched = store.get("repoA", "src/Foo.kt") + + assertEquals(Observation("repoA", "src/Foo.kt", "hash1", "module=foo", 100L), fetched) + } + + @Test + fun `put overwrites the prior row for the same key`(): Unit = runBlocking { + val store = store() + 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(store().get("repoA", "nope")) + } +} diff --git a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt index 1ee130c0..7271e579 100644 --- a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt +++ b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt @@ -53,8 +53,10 @@ import io.ktor.client.HttpClient import io.ktor.client.engine.cio.CIO import com.correx.infrastructure.router.turbovec.TurboVecL3MemoryStore import com.correx.infrastructure.router.turbovec.TurboVecSidecarConfig +import com.correx.core.context.observation.ObservationStore import com.correx.infrastructure.persistence.SqliteEventStore import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository +import com.correx.infrastructure.persistence.observation.SqliteObservationStore import com.correx.infrastructure.tools.DefaultToolRegistry import com.correx.infrastructure.tools.DispatchingToolExecutor import com.correx.infrastructure.tools.SandboxedToolExecutor @@ -106,6 +108,12 @@ object InfrastructureModule { ) } + fun createObservationStore(dbPath: String = defaultDbPath): ObservationStore { + val dir = java.io.File(dbPath).parentFile + if (!dir.exists()) dir.mkdirs() + return SqliteObservationStore(DriverManager.getConnection("jdbc:sqlite:$dbPath")) + } + fun createArtifactStore(rootDir: Path = Paths.get(defaultArtifactsRoot)): ArtifactStore { Files.createDirectories(rootDir) val index = SqliteArtifactIndex(rootDir.resolve("index.sqlite"))