From 9d38a89c88c061c3557717b76ac27e70bad9d6e1 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 11 Jun 2026 13:30:57 +0400 Subject: [PATCH] feat(memory): state-keyed repo-map reuse via L3 presence check; embed entries on fresh scan MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add existsByTurnIdPrefix(prefix) to L3MemoryStore interface; implemented in InMemoryL3MemoryStore (mutex-guarded) and TurboVecL3MemoryStore (metadata map). All test fakes (TrackingL3MemoryStore, CapturingStore) updated. - Introduce RepoMapIndexerPort interface; RepoMapIndexer implements it. Allows injection of test doubles without framework overhead. - ProjectMemoryService.indexAndRecord: reads latest WorkspaceStateObservedEvent from event store (invariant #9 — recorded fact, not re-probe); skips scan+embed when L3 already holds entries for "repomap::"; passes stateKey to RepoMapComputedEvent; embeds each scanned entry after append; per-entry failures warn-logged (CancellationException rethrown). - Tests: InMemoryL3MemoryStoreTest +2 cases; new ProjectMemoryServiceReuseTest covers same-key skip, changed-key re-index, no-observation always-reindex, and embedding turnId tag verification. --- .../server/memory/ProjectMemoryService.kt | 29 +++- .../apps/server/memory/RepoMapIndexer.kt | 9 +- .../memory/ProjectMemoryServiceReuseTest.kt | 124 ++++++++++++++++++ .../core/router/l3/InMemoryL3MemoryStore.kt | 3 + .../correx/core/router/l3/L3MemoryStore.kt | 5 + .../router/l3/InMemoryL3MemoryStoreTest.kt | 27 ++++ .../router/turbovec/TurboVecL3MemoryStore.kt | 3 + .../test/kotlin/L3MetadataRehydratorTest.kt | 1 + .../src/test/kotlin/RouterNarrationTest.kt | 2 + 9 files changed, 200 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceReuseTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt b/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt index 04b31372..faaf32a1 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt @@ -41,7 +41,7 @@ class ProjectMemoryService( private val journalRepository: DefaultDecisionJournalRepository, private val eventStore: EventStore, private val distiller: ProjectMemoryDistiller = ProjectMemoryDistiller(), - private val indexer: RepoMapIndexer = RepoMapIndexer(), + private val indexer: RepoMapIndexerPort = RepoMapIndexer(), private val workspaceStateProbe: WorkspaceStateProbe = RealWorkspaceStateProbe(), ) { private fun tag(repoRoot: String) = "project:$repoRoot" @@ -56,6 +56,13 @@ class ProjectMemoryService( */ suspend fun indexAndRecord(sessionId: SessionId, repoRoot: String) { if (!config.enabled) return + val stateKey = eventStore.read(sessionId) + .mapNotNull { it.payload as? WorkspaceStateObservedEvent } + .lastOrNull()?.stateKey + if (stateKey != null && l3MemoryStore.existsByTurnIdPrefix("repomap:$repoRoot:$stateKey")) { + log.debug("repo-map already indexed for {}@{} — skipping", repoRoot, stateKey) + return + } runCatching { val entries = indexer.index(java.nio.file.Path.of(repoRoot)) if (entries.isEmpty()) return @@ -73,10 +80,30 @@ class ProjectMemoryService( sessionId = sessionId, repoRoot = repoRoot, entries = entries, + stateKey = stateKey, computedAt = Clock.System.now(), ), ), ) + val tag = if (stateKey != null) "repomap:$repoRoot:$stateKey" else "repomap:$repoRoot" + entries.forEach { entry -> + runCatching { + val text = entry.path + if (entry.symbols.isEmpty()) "" else ": ${entry.symbols.joinToString(", ")}" + l3MemoryStore.store( + L3MemoryEntry( + id = UUID.randomUUID().toString(), + sessionId = sessionId, + turnId = tag, + text = text, + vector = embedder.embed(text), + timestampMs = Clock.System.now().toEpochMilliseconds(), + ), + ) + }.exceptionOrNull()?.let { e -> + if (e is CancellationException) throw e + log.warn("repo-map embedding failed for {}: {}", entry.path, e.message) + } + } }.exceptionOrNull()?.let { e -> if (e is CancellationException) throw e log.warn("repo-map indexing failed for {}: {}", repoRoot, e.message) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/memory/RepoMapIndexer.kt b/apps/server/src/main/kotlin/com/correx/apps/server/memory/RepoMapIndexer.kt index dcfae1b1..d027ba8f 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/memory/RepoMapIndexer.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/memory/RepoMapIndexer.kt @@ -11,6 +11,11 @@ import kotlin.io.path.name import kotlin.io.path.readText import kotlin.io.path.relativeTo +/** Minimal port for repo-map scanning — allows injection of test doubles. */ +interface RepoMapIndexerPort { + fun index(repoRoot: Path): List +} + /** * Walks a repo and produces a ranked file/symbol index. Reads the filesystem (a * nondeterministic environment observation) — the caller records the result as a @@ -24,8 +29,8 @@ import kotlin.io.path.relativeTo class RepoMapIndexer( private val maxDepth: Int = 4, private val ignoreGlobs: List = emptyList(), -) { - fun index(repoRoot: Path): List { +) : RepoMapIndexerPort { + override fun index(repoRoot: Path): List { if (!repoRoot.isDirectory()) return emptyList() val ignore = ignoreGlobs.toSet() diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceReuseTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceReuseTest.kt new file mode 100644 index 00000000..e5345b91 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceReuseTest.kt @@ -0,0 +1,124 @@ +package com.correx.apps.server.memory + +import com.correx.core.config.ProjectConfig +import com.correx.core.events.events.RepoMapEntry +import com.correx.core.inference.Embedder +import com.correx.core.journal.DecisionJournalProjector +import com.correx.core.journal.DefaultDecisionJournalReducer +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.router.l3.InMemoryL3MemoryStore +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.events.types.SessionId +import com.correx.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.nio.file.Path + +private class CountingIndexer(private val entries: List) : RepoMapIndexerPort { + var callCount = 0 + override fun index(repoRoot: Path): List { + callCount++ + return entries + } +} + +private class ConstantEmbedderReuse(override val dimension: Int = 4) : Embedder { + override suspend fun embed(text: String): FloatArray = FloatArray(dimension) { 0.5f } +} + +class ProjectMemoryServiceReuseTest { + + private val entries = listOf( + RepoMapEntry(path = "src/Foo.kt", score = 1.0, symbols = listOf("FooClass")), + ) + + private fun service( + es: InMemoryEventStore, + l3: InMemoryL3MemoryStore, + indexer: CountingIndexer, + probe: WorkspaceStateProbe, + root: String = "/repo", + ) = ProjectMemoryService( + config = ProjectConfig(enabled = true, root = root), + embedder = ConstantEmbedderReuse(), + l3MemoryStore = l3, + journalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(es, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ), + eventStore = es, + indexer = indexer, + workspaceStateProbe = probe, + ) + + @Test + fun `same stateKey skips re-index on second session`(): Unit = runBlocking { + val es = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val indexer = CountingIndexer(entries) + val probe = FakeWorkspaceStateProbe(WorkspaceState("git:hash1", "git", "main", false)) + + val svc = service(es, l3, indexer, probe) + val sessionA = SessionId("session-reuse-a") + val sessionB = SessionId("session-reuse-b") + + svc.observeAndRecord(sessionA, "/repo") + svc.indexAndRecord(sessionA, "/repo") + assertEquals(1, indexer.callCount, "first session should index") + + svc.observeAndRecord(sessionB, "/repo") + svc.indexAndRecord(sessionB, "/repo") + assertEquals(1, indexer.callCount, "second session with same stateKey should skip index") + } + + @Test + fun `changed stateKey re-indexes`(): Unit = runBlocking { + val es = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val indexer = CountingIndexer(entries) + + val svc1 = service(es, l3, indexer, FakeWorkspaceStateProbe(WorkspaceState("git:hash1", "git", "main", false))) + val sessionA = SessionId("session-change-a") + svc1.observeAndRecord(sessionA, "/repo") + svc1.indexAndRecord(sessionA, "/repo") + assertEquals(1, indexer.callCount) + + val svc2 = service(es, l3, indexer, FakeWorkspaceStateProbe(WorkspaceState("git:hash2", "git", "main", false))) + val sessionB = SessionId("session-change-b") + svc2.observeAndRecord(sessionB, "/repo") + svc2.indexAndRecord(sessionB, "/repo") + assertEquals(2, indexer.callCount, "changed stateKey should trigger re-index") + } + + @Test + fun `no observation event always re-indexes`(): Unit = runBlocking { + val es = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val indexer = CountingIndexer(entries) + val probe = FakeWorkspaceStateProbe(null) + + val svc = service(es, l3, indexer, probe) + svc.indexAndRecord(SessionId("session-noobs-1"), "/repo") + svc.indexAndRecord(SessionId("session-noobs-2"), "/repo") + assertEquals(2, indexer.callCount, "no observation event -> always re-index") + } + + @Test + fun `entries are embedded with stateKey turnId tag after fresh index`(): Unit = runBlocking { + val es = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val indexer = CountingIndexer(entries) + val probe = FakeWorkspaceStateProbe(WorkspaceState("git:hash1", "git", "main", false)) + + val svc = service(es, l3, indexer, probe) + val sessionId = SessionId("session-embed-1") + svc.observeAndRecord(sessionId, "/repo") + svc.indexAndRecord(sessionId, "/repo") + + assertTrue( + l3.existsByTurnIdPrefix("repomap:/repo:git:hash1"), + "entries should be embedded with stateKey tag", + ) + } +} diff --git a/core/router/src/main/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStore.kt b/core/router/src/main/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStore.kt index 4c8431e7..4ed7df66 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStore.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStore.kt @@ -33,6 +33,9 @@ class InMemoryL3MemoryStore : L3MemoryStore { } } + override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = + mutex.withLock { entries.any { it.turnId.startsWith(prefix) } } + override suspend fun close() { // No-op for in-memory store } diff --git a/core/router/src/main/kotlin/com/correx/core/router/l3/L3MemoryStore.kt b/core/router/src/main/kotlin/com/correx/core/router/l3/L3MemoryStore.kt index 38d67c60..b3777302 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/l3/L3MemoryStore.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/l3/L3MemoryStore.kt @@ -15,6 +15,11 @@ interface L3MemoryStore { */ suspend fun query(query: L3Query): List + /** + * Returns true if any stored entry has a [L3MemoryEntry.turnId] that starts with [prefix]. + */ + suspend fun existsByTurnIdPrefix(prefix: String): Boolean + /** * Close the store and release resources. */ diff --git a/core/router/src/test/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStoreTest.kt b/core/router/src/test/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStoreTest.kt index 75c48315..2b074e1f 100644 --- a/core/router/src/test/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStoreTest.kt +++ b/core/router/src/test/kotlin/com/correx/core/router/l3/InMemoryL3MemoryStoreTest.kt @@ -51,6 +51,33 @@ class InMemoryL3MemoryStoreTest { assertEquals("entry-2", hits[1].entry.id, "Second match should be entry-2") } + @Test + fun `existsByTurnIdPrefix returns true when entry turnId starts with prefix`(): Unit = runBlocking { + val store = InMemoryL3MemoryStore() + val sessionId = TypeId("session-1") + store.store( + L3MemoryEntry( + id = "e1", sessionId = sessionId, turnId = "repomap:/repo:git:abc123", + text = "foo", vector = floatArrayOf(1f), timestampMs = 1L + ) + ) + assert(store.existsByTurnIdPrefix("repomap:/repo:git:abc123")) { "should match exact" } + assert(store.existsByTurnIdPrefix("repomap:/repo:git:")) { "should match prefix" } + } + + @Test + fun `existsByTurnIdPrefix returns false when no entry matches`(): Unit = runBlocking { + val store = InMemoryL3MemoryStore() + val sessionId = TypeId("session-1") + store.store( + L3MemoryEntry( + id = "e1", sessionId = sessionId, turnId = "project:/repo", + text = "foo", vector = floatArrayOf(1f), timestampMs = 1L + ) + ) + assert(!store.existsByTurnIdPrefix("repomap:/repo:git:")) { "should not match different prefix" } + } + @Test fun `respects sessionIdFilter`(): Unit = runBlocking { val store = InMemoryL3MemoryStore() diff --git a/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt b/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt index 81ef52c6..fafe978d 100644 --- a/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt +++ b/infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt @@ -43,6 +43,9 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra metadata[entry.id] = entry } + override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = + metadata.values.any { it.turnId.startsWith(prefix) } + override suspend fun query(query: L3Query): List { val request = SidecarRequest( op = "search", diff --git a/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt b/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt index 85d54c8c..4687fea0 100644 --- a/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt +++ b/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt @@ -30,6 +30,7 @@ class L3MetadataRehydratorTest { } override suspend fun store(entry: L3MemoryEntry) = Unit override suspend fun query(query: L3Query): List = emptyList() + override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = false override suspend fun close() = Unit } diff --git a/testing/deterministic/src/test/kotlin/RouterNarrationTest.kt b/testing/deterministic/src/test/kotlin/RouterNarrationTest.kt index 4a0a7b52..0f791eb0 100644 --- a/testing/deterministic/src/test/kotlin/RouterNarrationTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterNarrationTest.kt @@ -374,6 +374,8 @@ class RouterNarrationTest { override suspend fun query(query: L3Query): List = emptyList() + override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = false + override suspend fun close() = Unit } }