feat: rehydrate L3 metadata from event log on startup

TurboVec persists vectors via the sidecar but kept entry metadata (sessionId/turnId/text) only in an in-memory map, so restarting with backend=turbovec silently dropped every query hit (the id→metadata lookup missed). Rebuild that map from the ChatTurnEvent log at startup. - RehydratableL3MemoryStore: capability interface for stores whose vectors persist independently of the JVM. TurboVec implements it; InMemory deliberately does not (its vectors are volatile too). - L3MetadataRehydrator replays ChatTurnEvents into the metadata map. No embedder: the query path never reads entry.vector, so vectors stay in the sidecar and metadata comes from the event log (the source of truth, invariant #1). Short-circuits for non-rehydratable backends before scanning the log. - Wired into Main before the server accepts queries; no sidecar spawn. Backfill of never-embedded history (needs the embedder) is a separate follow-up. The in_memory non-durability warning already exists.
2026-05-31 22:08:52 +04:00
parent d03479cea7
commit 5beb866036
5 changed files with 181 additions and 3 deletions
@@ -26,6 +26,7 @@ import com.correx.core.kernel.orchestration.OrchestratorEngines
 import com.correx.core.kernel.orchestration.OrchestratorRepositories
 import com.correx.core.kernel.retry.DefaultRetryCoordinator
 import com.correx.core.risk.DefaultRiskAssessor
+import com.correx.core.router.l3.L3MetadataRehydrator
 import com.correx.core.router.model.RouterConfig
 import com.correx.core.sessions.DefaultSessionReducer
 import com.correx.core.sessions.DefaultSessionRepository
@@ -54,6 +55,7 @@ import com.correx.infrastructure.tools.ShellConfig
 import com.correx.infrastructure.tools.ToolConfig
 import io.ktor.server.engine.embeddedServer
 import io.ktor.server.netty.Netty
+import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory
 import java.nio.file.Path
 import java.nio.file.Paths
@@ -163,6 +165,10 @@ fun main() {
    val routerConfig = correxConfig.router
    val embedder = InfrastructureModule.createEmbedderFromConfig(routerConfig.embedder)
    val l3MemoryStore = InfrastructureModule.createL3MemoryStoreFromConfig(routerConfig.l3)
+    // Rebuild the L3 metadata map from the ChatTurnEvent log before serving queries:
+    // persisted vectors survive restart but the metadata map does not (no-op for
+    // non-rehydratable backends like in_memory).
+    runBlocking { L3MetadataRehydrator(eventStore).rehydrate(l3MemoryStore) }
    val routerFacade = InfrastructureModule.createRouterFacade(
        eventStore = eventStore,
        inferenceRouter = inferenceRouter,
@@ -0,0 +1,40 @@
+package com.correx.core.router.l3
+
+import com.correx.core.events.events.ChatTurnEvent
+import com.correx.core.events.stores.EventStore
+import org.slf4j.LoggerFactory
+
+/**
+ * Rebuilds a [RehydratableL3MemoryStore]'s in-memory metadata map from the
+ * `ChatTurnEvent` log at startup.
+ *
+ * Persisted vectors (e.g. the TurboVec sidecar index) survive a restart, but the
+ * metadata map does not; without this every L3 query silently drops its hits
+ * because the `id → metadata` lookup misses. Rehydration reads only the event log
+ * (no embedder), so it is cheap and replay-safe. Stores without independently
+ * persisted vectors are not rehydratable and are skipped.
+ */
+class L3MetadataRehydrator(private val eventStore: EventStore) {
+    private val log = LoggerFactory.getLogger(L3MetadataRehydrator::class.java)
+
+    /** @return the number of entries rehydrated (0 if [store] is not rehydratable). */
+    suspend fun rehydrate(store: L3MemoryStore): Int {
+        if (store !is RehydratableL3MemoryStore) return 0
+        val entries = eventStore.allEvents()
+            .mapNotNull { it.payload as? ChatTurnEvent }
+            .map { turn ->
+                L3MemoryEntry(
+                    id = turn.turnId,
+                    sessionId = turn.sessionId,
+                    turnId = turn.turnId,
+                    text = turn.content,
+                    vector = FloatArray(0),
+                    timestampMs = turn.timestampMs,
+                )
+            }
+            .toList()
+        store.rehydrateMetadata(entries)
+        log.info("L3 metadata rehydrated: {} chat turns from event log", entries.size)
+        return entries.size
+    }
+}
@@ -0,0 +1,23 @@
+package com.correx.core.router.l3
+
+/**
+ * An [L3MemoryStore] whose vectors persist independently of the JVM (e.g. a
+ * sidecar index on disk) but whose entry metadata (`sessionId`/`turnId`/`text`)
+ * lives in memory and is therefore lost on restart.
+ *
+ * Such a store can rebuild that metadata from the `ChatTurnEvent` log at startup:
+ * the event log is the source of truth (invariant #1) and the metadata map is a
+ * disposable projection. Only the metadata map is loaded — the persisted vectors
+ * are never re-sent, so rehydration needs no embedder.
+ *
+ * Stores whose vectors are themselves volatile (e.g. [InMemoryL3MemoryStore])
+ * deliberately do NOT implement this: rehydrating metadata without matching
+ * vectors would be meaningless.
+ */
+interface RehydratableL3MemoryStore : L3MemoryStore {
+    /**
+     * Load entry metadata into the in-memory map without re-sending vectors.
+     * Idempotent: re-running must not clobber entries already present.
+     */
+    suspend fun rehydrateMetadata(entries: List<L3MemoryEntry>)
+}
@@ -2,8 +2,8 @@ package com.correx.infrastructure.router.turbovec

 import com.correx.core.router.l3.L3Hit
 import com.correx.core.router.l3.L3MemoryEntry
-import com.correx.core.router.l3.L3MemoryStore
 import com.correx.core.router.l3.L3Query
+import com.correx.core.router.l3.RehydratableL3MemoryStore
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeout
@@ -15,14 +15,20 @@ import kotlin.io.path.createDirectories
 * Adapter to TurboVec sidecar process for cross-session vector-based memory.
 * Maintains in-memory metadata (sessionId, turnId, text, timestamp) since turbovec
 * stores only vectors and IDs.
+ *
+ * The metadata map is volatile, but the sidecar vectors persist (via `persistPath`);
+ * on restart [rehydrateMetadata] rebuilds the map from the `ChatTurnEvent` log so
+ * queries resolve their hits again (invariant #1: the event log is authoritative).
 */
-class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : L3MemoryStore {
+class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : RehydratableL3MemoryStore {
    private val json = Json { ignoreUnknownKeys = true }
    private val mutex = Mutex()
    private var process: Process? = null
    private val metadata = java.util.concurrent.ConcurrentHashMap<String, L3MemoryEntry>()

-    // SCAFFOLDING: metadata persistence to be added with ChatTurnEvent
+    override suspend fun rehydrateMetadata(entries: List<L3MemoryEntry>) {
+        entries.forEach { metadata.putIfAbsent(it.id, it) }
+    }

    override suspend fun store(entry: L3MemoryEntry) {
        val request = SidecarRequest(
@@ -0,0 +1,103 @@
+import com.correx.core.events.events.ChatTurnEvent
+import com.correx.core.events.events.ChatTurnRole
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.L3MemoryRetrievedEvent
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.StoredEvent
+import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.EventId
+import com.correx.core.events.types.SessionId
+import com.correx.core.router.l3.InMemoryL3MemoryStore
+import com.correx.core.router.l3.L3Hit
+import com.correx.core.router.l3.L3MemoryEntry
+import com.correx.core.router.l3.L3MetadataRehydrator
+import com.correx.core.router.l3.L3Query
+import com.correx.core.router.l3.RehydratableL3MemoryStore
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Instant
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class L3MetadataRehydratorTest {
+
+    private class CapturingStore : RehydratableL3MemoryStore {
+        val loaded = mutableListOf<L3MemoryEntry>()
+        override suspend fun rehydrateMetadata(entries: List<L3MemoryEntry>) {
+            loaded += entries
+        }
+        override suspend fun store(entry: L3MemoryEntry) = Unit
+        override suspend fun query(query: L3Query): List<L3Hit> = emptyList()
+        override suspend fun close() = Unit
+    }
+
+    private class FakeEventStore(private val stored: List<StoredEvent>) : EventStore {
+        override fun allEvents(): Sequence<StoredEvent> = stored.asSequence()
+        override fun read(sessionId: SessionId): List<StoredEvent> =
+            stored.filter { it.metadata.sessionId == sessionId }
+        override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = read(sessionId)
+        override fun lastSequence(sessionId: SessionId): Long? = null
+        override fun allSessionIds(): Set<SessionId> = stored.map { it.metadata.sessionId }.toSet()
+        override suspend fun append(event: NewEvent): StoredEvent = throw UnsupportedOperationException()
+        override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = throw UnsupportedOperationException()
+        override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = throw UnsupportedOperationException()
+        override fun subscribeAll(): Flow<StoredEvent> = throw UnsupportedOperationException()
+        override suspend fun lastGlobalSequence(): Long = 0
+    }
+
+    private var seq = 0L
+
+    private fun event(sessionId: String, payload: EventPayload): StoredEvent {
+        seq += 1
+        return StoredEvent(
+            metadata = EventMetadata(
+                eventId = EventId("e$seq"),
+                sessionId = SessionId(sessionId),
+                timestamp = Instant.fromEpochMilliseconds(seq),
+                schemaVersion = 1,
+                causationId = null,
+                correlationId = null,
+            ),
+            sequence = seq,
+            sessionSequence = seq,
+            payload = payload,
+        )
+    }
+
+    private fun chatTurn(sessionId: String, turnId: String, text: String) =
+        event(sessionId, ChatTurnEvent(SessionId(sessionId), turnId, ChatTurnRole.USER, text, timestampMs = 42L))
+
+    @Test
+    fun `rehydrates chat turns across sessions and ignores non-chat events`(): Unit = runBlocking {
+        val store = FakeEventStore(
+            listOf(
+                chatTurn("s1", "t1", "hello"),
+                event("s1", L3MemoryRetrievedEvent(SessionId("s1"), "t1", emptyList(), timestampMs = 1L)),
+                chatTurn("s2", "t2", "world"),
+            ),
+        )
+        val target = CapturingStore()
+
+        val count = L3MetadataRehydrator(store).rehydrate(target)
+
+        assertEquals(2, count)
+        assertEquals(2, target.loaded.size)
+        val byTurn = target.loaded.associateBy { it.turnId }
+        assertEquals(SessionId("s1"), byTurn.getValue("t1").sessionId)
+        assertEquals("hello", byTurn.getValue("t1").text)
+        assertEquals("t1", byTurn.getValue("t1").id)
+        assertEquals(SessionId("s2"), byTurn.getValue("t2").sessionId)
+        assertTrue(byTurn.getValue("t2").vector.isEmpty(), "rehydrated entry holds no vector")
+    }
+
+    @Test
+    fun `non-rehydratable store is skipped`(): Unit = runBlocking {
+        val store = FakeEventStore(listOf(chatTurn("s1", "t1", "hello")))
+
+        val count = L3MetadataRehydrator(store).rehydrate(InMemoryL3MemoryStore())
+
+        assertEquals(0, count)
+    }
+}