feat(memory): state-keyed repo-map reuse via L3 presence check; embed entries on fresh scan
- 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:<root>:<stateKey>"; 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.
This commit is contained in:
@@ -41,7 +41,7 @@ class ProjectMemoryService(
|
|||||||
private val journalRepository: DefaultDecisionJournalRepository,
|
private val journalRepository: DefaultDecisionJournalRepository,
|
||||||
private val eventStore: EventStore,
|
private val eventStore: EventStore,
|
||||||
private val distiller: ProjectMemoryDistiller = ProjectMemoryDistiller(),
|
private val distiller: ProjectMemoryDistiller = ProjectMemoryDistiller(),
|
||||||
private val indexer: RepoMapIndexer = RepoMapIndexer(),
|
private val indexer: RepoMapIndexerPort = RepoMapIndexer(),
|
||||||
private val workspaceStateProbe: WorkspaceStateProbe = RealWorkspaceStateProbe(),
|
private val workspaceStateProbe: WorkspaceStateProbe = RealWorkspaceStateProbe(),
|
||||||
) {
|
) {
|
||||||
private fun tag(repoRoot: String) = "project:$repoRoot"
|
private fun tag(repoRoot: String) = "project:$repoRoot"
|
||||||
@@ -56,6 +56,13 @@ class ProjectMemoryService(
|
|||||||
*/
|
*/
|
||||||
suspend fun indexAndRecord(sessionId: SessionId, repoRoot: String) {
|
suspend fun indexAndRecord(sessionId: SessionId, repoRoot: String) {
|
||||||
if (!config.enabled) return
|
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 {
|
runCatching {
|
||||||
val entries = indexer.index(java.nio.file.Path.of(repoRoot))
|
val entries = indexer.index(java.nio.file.Path.of(repoRoot))
|
||||||
if (entries.isEmpty()) return
|
if (entries.isEmpty()) return
|
||||||
@@ -73,10 +80,30 @@ class ProjectMemoryService(
|
|||||||
sessionId = sessionId,
|
sessionId = sessionId,
|
||||||
repoRoot = repoRoot,
|
repoRoot = repoRoot,
|
||||||
entries = entries,
|
entries = entries,
|
||||||
|
stateKey = stateKey,
|
||||||
computedAt = Clock.System.now(),
|
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 ->
|
}.exceptionOrNull()?.let { e ->
|
||||||
if (e is CancellationException) throw e
|
if (e is CancellationException) throw e
|
||||||
log.warn("repo-map indexing failed for {}: {}", repoRoot, e.message)
|
log.warn("repo-map indexing failed for {}: {}", repoRoot, e.message)
|
||||||
|
|||||||
@@ -11,6 +11,11 @@ import kotlin.io.path.name
|
|||||||
import kotlin.io.path.readText
|
import kotlin.io.path.readText
|
||||||
import kotlin.io.path.relativeTo
|
import kotlin.io.path.relativeTo
|
||||||
|
|
||||||
|
/** Minimal port for repo-map scanning — allows injection of test doubles. */
|
||||||
|
interface RepoMapIndexerPort {
|
||||||
|
fun index(repoRoot: Path): List<RepoMapEntry>
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Walks a repo and produces a ranked file/symbol index. Reads the filesystem (a
|
* Walks a repo and produces a ranked file/symbol index. Reads the filesystem (a
|
||||||
* nondeterministic environment observation) — the caller records the result as a
|
* nondeterministic environment observation) — the caller records the result as a
|
||||||
@@ -24,8 +29,8 @@ import kotlin.io.path.relativeTo
|
|||||||
class RepoMapIndexer(
|
class RepoMapIndexer(
|
||||||
private val maxDepth: Int = 4,
|
private val maxDepth: Int = 4,
|
||||||
private val ignoreGlobs: List<String> = emptyList(),
|
private val ignoreGlobs: List<String> = emptyList(),
|
||||||
) {
|
) : RepoMapIndexerPort {
|
||||||
fun index(repoRoot: Path): List<RepoMapEntry> {
|
override fun index(repoRoot: Path): List<RepoMapEntry> {
|
||||||
if (!repoRoot.isDirectory()) return emptyList()
|
if (!repoRoot.isDirectory()) return emptyList()
|
||||||
val ignore = ignoreGlobs.toSet()
|
val ignore = ignoreGlobs.toSet()
|
||||||
|
|
||||||
|
|||||||
+124
@@ -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<RepoMapEntry>) : RepoMapIndexerPort {
|
||||||
|
var callCount = 0
|
||||||
|
override fun index(repoRoot: Path): List<RepoMapEntry> {
|
||||||
|
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",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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() {
|
override suspend fun close() {
|
||||||
// No-op for in-memory store
|
// No-op for in-memory store
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,6 +15,11 @@ interface L3MemoryStore {
|
|||||||
*/
|
*/
|
||||||
suspend fun query(query: L3Query): List<L3Hit>
|
suspend fun query(query: L3Query): List<L3Hit>
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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.
|
* Close the store and release resources.
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -51,6 +51,33 @@ class InMemoryL3MemoryStoreTest {
|
|||||||
assertEquals("entry-2", hits[1].entry.id, "Second match should be entry-2")
|
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
|
@Test
|
||||||
fun `respects sessionIdFilter`(): Unit = runBlocking {
|
fun `respects sessionIdFilter`(): Unit = runBlocking {
|
||||||
val store = InMemoryL3MemoryStore()
|
val store = InMemoryL3MemoryStore()
|
||||||
|
|||||||
+3
@@ -43,6 +43,9 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra
|
|||||||
metadata[entry.id] = entry
|
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<L3Hit> {
|
override suspend fun query(query: L3Query): List<L3Hit> {
|
||||||
val request = SidecarRequest(
|
val request = SidecarRequest(
|
||||||
op = "search",
|
op = "search",
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ class L3MetadataRehydratorTest {
|
|||||||
}
|
}
|
||||||
override suspend fun store(entry: L3MemoryEntry) = Unit
|
override suspend fun store(entry: L3MemoryEntry) = Unit
|
||||||
override suspend fun query(query: L3Query): List<L3Hit> = emptyList()
|
override suspend fun query(query: L3Query): List<L3Hit> = emptyList()
|
||||||
|
override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = false
|
||||||
override suspend fun close() = Unit
|
override suspend fun close() = Unit
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -374,6 +374,8 @@ class RouterNarrationTest {
|
|||||||
|
|
||||||
override suspend fun query(query: L3Query): List<com.correx.core.router.l3.L3Hit> = emptyList()
|
override suspend fun query(query: L3Query): List<com.correx.core.router.l3.L3Hit> = emptyList()
|
||||||
|
|
||||||
|
override suspend fun existsByTurnIdPrefix(prefix: String): Boolean = false
|
||||||
|
|
||||||
override suspend fun close() = Unit
|
override suspend fun close() = Unit
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user