refactor(acr): ACR Store 1 as a disposable memo, not a durable side-store (#305)

The shipped Store 1 persisted repo-file descriptors in a new SqliteObservationStore
— a second, unsynchronized SQLite writer on the event-log DB (no WAL/busy_timeout,
one shared Connection across coroutines) holding a fact the log already carries. That
violates "the event log is the only source of truth" (inv #1/#8): an Observation is
just FileWrittenEvent.path + postImageHash + describe(bytes).render(), and render is a
pure function of content-addressed CAS bytes.

Replace the whole durable apparatus with an in-process memo (descriptorMemo, keyed on
repoRoot+path+contentHash) on the orchestrator, next to the existing artifactContentCache.
Rebuilt from the log on restart, disposable, no external store. Preserves the perf win
(skip CAS read + regex on repeat describes) with none of the concurrency/lock risk.

Deletes: ObservationStore/InMemoryObservationStore/SqliteObservationStore (+ tests),
InfrastructureModule.createObservationStore, the FileReadTool priming path (which never
hit anyway — it keyed on a workspace-relative path while every consumer looks up the
absolute FileWrittenEvent.path), the FileReadConfig/SessionOrchestrator plumbing, and
five orphaned build.gradle deps.

Store 2 (soft-confidence hints) unchanged — it was correct.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HMbPmZZjcXhR2crU82zZ8S
This commit is contained in:
2026-07-21 18:42:47 +04:00
parent 8806de1628
commit 12775d56da
16 changed files with 9 additions and 267 deletions
-1
View File
@@ -12,7 +12,6 @@ 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")
-1
View File
@@ -10,7 +10,6 @@ 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")))
@@ -1,76 +0,0 @@
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
}
}
@@ -1,37 +0,0 @@
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"))
}
}
@@ -53,10 +53,8 @@ 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
@@ -108,12 +106,6 @@ 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"))
-1
View File
@@ -17,7 +17,6 @@ dependencies {
implementation project(":infrastructure:tools:filesystem")
implementation project(":core:artifacts")
implementation project(":core:artifacts-store")
implementation project(":core:context")
// Web research tools (web_search, web_fetch) + deterministic HTML→markdown extraction.
implementation 'org.jsoup:jsoup:1.20.1'
@@ -10,8 +10,6 @@ dependencies {
implementation project(':core:approvals')
implementation project(':core:artifacts')
implementation project(':core:artifacts-store')
implementation project(':core:context')
implementation project(':core:sourcedesc')
implementation 'org.slf4j:slf4j-api:2.0.16'
}
@@ -1,10 +1,7 @@
package com.correx.infrastructure.tools.filesystem
import com.correx.core.approvals.Tier
import com.correx.core.context.observation.Observation
import com.correx.core.context.observation.ObservationStore
import com.correx.core.events.events.ToolRequest
import com.correx.core.sourcedesc.describe
import com.correx.core.tools.compression.CompressionRule.StripBlankLines
import com.correx.core.tools.compression.CompressionRule.StripLeadingWhitespace
import com.correx.core.tools.compression.DeclarativeCompressor
@@ -35,12 +32,6 @@ import kotlin.io.path.name
class FileReadTool(
private val allowedPaths: Set<Path> = emptySet(),
private val workingDir: Path? = null,
// ACR Store 1 (docs/plans/2026-07-21-acr-knowledge-accretion.md): primes the cross-session
// observation cache on a whole-file read, not just on write, so a later describeCached lookup
// (SessionOrchestratorArtifacts.kt) for this exact content hash is already warm. Both null by
// default — no-op unless a workspace wiring supplies them.
private val observationStore: ObservationStore? = null,
private val repoRoot: String? = null,
) : Tool, ToolExecutor {
// Resolve relative paths against the bound workspace's working dir, NOT the JVM
@@ -149,7 +140,7 @@ class FileReadTool(
}
}
private suspend fun readFile(path: Path, startLine: Int?, endLine: Int?, request: ToolRequest): ToolResult = runCatching {
private fun readFile(path: Path, startLine: Int?, endLine: Int?, request: ToolRequest): ToolResult = runCatching {
// Read the file once. Lines and the (whole-read) content hash both derive from these bytes,
// so a full read no longer hits the disk twice (readAllLines + readAllBytes for the hash).
val bytes = Files.readAllBytes(path)
@@ -178,9 +169,7 @@ class FileReadTool(
// against what the agent actually saw. A partial or truncated view establishes no baseline —
// the agent must read the relevant range before editing.
val sawWholeFile = startLine == null && endLine == null && !lineCapped && !charCapped
val hash = if (sawWholeFile) sha256(bytes) else null
val metadata = hash?.let { mapOf("contentHash" to it) } ?: emptyMap()
hash?.let { primeObservation(path, bytes, it) }
val metadata = if (sawWholeFile) mapOf("contentHash" to sha256(bytes)) else emptyMap()
ToolResult.Success(
invocationId = request.invocationId,
output = content + note,
@@ -194,17 +183,6 @@ class FileReadTool(
)
}
/** No-op unless both [observationStore] and [repoRoot] are wired; skips if already cached for this hash. */
private suspend fun primeObservation(path: Path, bytes: ByteArray, hash: String) {
val store = observationStore ?: return
val root = repoRoot ?: return
val relPath = workingDir?.let { runCatching { it.relativize(path).toString() }.getOrNull() } ?: path.toString()
val cached = store.get(root, relPath)
if (cached?.contentHash == hash) return
val rendered = describe(relPath, bytes).render().takeIf { it.isNotBlank() }
store.put(Observation(root, relPath, hash, rendered, System.currentTimeMillis()))
}
private fun sha256(bytes: ByteArray): String =
java.security.MessageDigest.getInstance("SHA-256").digest(bytes)
.joinToString("") { "%02x".format(it) }
@@ -1,6 +1,5 @@
package com.correx.infrastructure.tools.filesystem
import com.correx.core.context.observation.InMemoryObservationStore
import com.correx.core.events.events.ToolRequest
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
@@ -194,24 +193,6 @@ class FileReadToolTest {
assertTrue((result as ToolResult.Success).metadata["contentHash"] != null)
}
@Test
fun `execute primes the observation store for a whole-file read`(): Unit = runBlocking {
val tempFile = Files.createTempFile("test_prime", ".kt")
Files.writeString(tempFile, "fun foo() {}")
val store = InMemoryObservationStore()
val tool = FileReadTool(
allowedPaths = setOf(tempFile.parent),
workingDir = tempFile.parent,
observationStore = store,
repoRoot = "repoA",
)
tool.execute(createRequest(tempFile.toString()))
val observed = store.get("repoA", tempFile.fileName.toString())
assertTrue(observed != null)
}
@Test
fun `outputCompressor strips blank lines and leading whitespace`() {
val tool = FileReadTool()
@@ -1,6 +1,5 @@
package com.correx.infrastructure.tools
import com.correx.core.context.observation.ObservationStore
import com.correx.core.tools.contract.Tool
import com.correx.infrastructure.tools.filesystem.FileDeleteTool
import com.correx.infrastructure.tools.filesystem.FileEditTool
@@ -40,9 +39,6 @@ data class FileReadConfig(
val enabled: Boolean = false,
val allowedPaths: Set<Path> = emptySet(),
val workingDir: Path? = null,
// ACR Store 1 hot-path priming (docs/plans/2026-07-21-acr-knowledge-accretion.md); both null is a no-op.
val observationStore: ObservationStore? = null,
val repoRoot: String? = null,
)
data class FileWriteConfig(
val enabled: Boolean = false,
@@ -79,8 +75,6 @@ fun ToolConfig.buildTools(): List<Tool> = buildList {
FileReadTool(
allowedPaths = fileRead.allowedPaths,
workingDir = fileRead.workingDir,
observationStore = fileRead.observationStore,
repoRoot = fileRead.repoRoot,
),
)
// list_dir is read-only enumeration; shares the reader's jail + anchor and the same toggle.