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 81b38078..b93a0434 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 @@ -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, diff --git a/core/router/src/main/kotlin/com/correx/core/router/l3/L3MetadataRehydrator.kt b/core/router/src/main/kotlin/com/correx/core/router/l3/L3MetadataRehydrator.kt new file mode 100644 index 00000000..b9e63939 --- /dev/null +++ b/core/router/src/main/kotlin/com/correx/core/router/l3/L3MetadataRehydrator.kt @@ -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 + } +} diff --git a/core/router/src/main/kotlin/com/correx/core/router/l3/RehydratableL3MemoryStore.kt b/core/router/src/main/kotlin/com/correx/core/router/l3/RehydratableL3MemoryStore.kt new file mode 100644 index 00000000..9597e5e3 --- /dev/null +++ b/core/router/src/main/kotlin/com/correx/core/router/l3/RehydratableL3MemoryStore.kt @@ -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) +} 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 30ec366d..81ef52c6 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 @@ -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() - // SCAFFOLDING: metadata persistence to be added with ChatTurnEvent + override suspend fun rehydrateMetadata(entries: List) { + entries.forEach { metadata.putIfAbsent(it.id, it) } + } override suspend fun store(entry: L3MemoryEntry) { val request = SidecarRequest( diff --git a/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt b/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt new file mode 100644 index 00000000..85d54c8c --- /dev/null +++ b/testing/deterministic/src/test/kotlin/L3MetadataRehydratorTest.kt @@ -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() + override suspend fun rehydrateMetadata(entries: List) { + loaded += entries + } + override suspend fun store(entry: L3MemoryEntry) = Unit + override suspend fun query(query: L3Query): List = emptyList() + override suspend fun close() = Unit + } + + private class FakeEventStore(private val stored: List) : EventStore { + override fun allEvents(): Sequence = stored.asSequence() + override fun read(sessionId: SessionId): List = + stored.filter { it.metadata.sessionId == sessionId } + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = read(sessionId) + override fun lastSequence(sessionId: SessionId): Long? = null + override fun allSessionIds(): Set = stored.map { it.metadata.sessionId }.toSet() + override suspend fun append(event: NewEvent): StoredEvent = throw UnsupportedOperationException() + override suspend fun appendAll(events: List): List = throw UnsupportedOperationException() + override fun subscribe(sessionId: SessionId): Flow = throw UnsupportedOperationException() + override fun subscribeAll(): Flow = 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) + } +}