fix(l3): persistent sidecar reader, kill-on-desync, wire persistPath save/load

infrastructure/router/turbovec/src/main/kotlin/com/correx/infrastructure/router/turbovec/TurboVecL3MemoryStore.kt

@@ -8,8 +8,10 @@ import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.json.Json
+import java.io.BufferedReader
 import java.nio.file.Paths
 import kotlin.io.path.createDirectories
+import kotlin.io.path.exists
 
 /**
  * Adapter to TurboVec sidecar process for cross-session vector-based memory.
@@ -24,6 +26,9 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra
     private val json = Json { ignoreUnknownKeys = true }
     private val mutex = Mutex()
     private var process: Process? = null
+    // One reader owned by the store for the process's lifetime. A fresh bufferedReader() per request
+    // would strand read-ahead bytes in the discarded reader's buffer, desyncing the stream.
+    private var reader: BufferedReader? = null
     private val metadata = java.util.concurrent.ConcurrentHashMap<String, L3MemoryEntry>()
 
     override suspend fun rehydrateMetadata(entries: List<L3MemoryEntry>) {
@@ -72,10 +77,12 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra
     override suspend fun close() {
         mutex.withLock {
             process?.let { p ->
-                val shutdownRequest = SidecarRequest(op = "shutdown")
+                // Persist vectors before exit so the next process can `load` them (only save() puts
-                runCatching { sendRequest(shutdownRequest) }
+                // the quantized index on disk — init alone loses everything between runs).
+                config.persistPath?.let { runCatching { sendRequestLocked(SidecarRequest(op = "save", path = it.toString())) } }
+                runCatching { sendRequestLocked(SidecarRequest(op = "shutdown")) }
                 try {
-                    withTimeout(2000) {
+                    withTimeout(SHUTDOWN_TIMEOUT_MS) {
                         p.waitFor()
                     }
                 } catch (e: Exception) {
@@ -85,27 +92,41 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra
                         throw e
                     }
                 }
+                reader = null
+                process = null
             }
         }
     }
 
-    private suspend fun sendRequest(request: SidecarRequest): SidecarResponse {
+    private suspend fun sendRequest(request: SidecarRequest): SidecarResponse =
-        mutex.withLock {
+        mutex.withLock { sendRequestLocked(request) }
+
+    /** Caller must hold [mutex]. */
+    private fun sendRequestLocked(request: SidecarRequest): SidecarResponse {
         ensureProcessStarted()
         val process = process ?: throw RuntimeException("Process failed to start")
+        val reader = reader ?: throw RuntimeException("Sidecar reader not initialized")
 
         val requestLine = json.encodeToString(SidecarRequest.serializer(), request) + "\n"
         process.outputStream.write(requestLine.toByteArray())
         process.outputStream.flush()
 
-            val responseLine = runCatching {
+        // A blocking readLine can't be cancelled by a coroutine timeout, so a slow/late response
-                withTimeout(config.requestTimeoutMs) {
+        // would land in the NEXT request's read and desync the stream permanently. We don't wrap it
-                    process.inputStream.bufferedReader().readLine()
+        // in withTimeout at all: instead the null/failure path below tears the process down so the
+        // next request restarts from a clean, empty stream rather than reading a stale line.
+        val responseLine = runCatching { reader.readLine() }.getOrNull()
+        if (responseLine == null) {
+            killProcess()
+            throw RuntimeException("No response from sidecar (process reset)")
         }
-            }.getOrNull() ?: throw RuntimeException("No response from sidecar")
-
         return json.decodeFromString(SidecarResponse.serializer(), responseLine)
     }
+
+    private fun killProcess() {
+        runCatching { process?.destroyForcibly() }
+        reader = null
+        process = null
     }
 
     private fun ensureProcessStarted() {
@@ -122,19 +143,24 @@ class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : Rehydra
 
         val started = pb.start()
         process = started
+        reader = started.inputStream.bufferedReader()
 
-        // The sidecar rejects add/search until the index is created. Send init once on startup;
+        // Reload a persisted index if one exists, else create a fresh one. Either way the sidecar
-        // without this every store/query fails with "Index not initialized".
+        // rejects add/search until initialized; without this every store/query fails "not initialized".
-        val initLine = json.encodeToString(
+        val bootstrap = config.persistPath
-            SidecarRequest.serializer(),
+            ?.takeIf { it.exists() }
-            SidecarRequest(op = "init", dim = config.dim, bitWidth = config.bitWidth)
+            ?.let { SidecarRequest(op = "load", path = it.toString()) }
-        ) + "\n"
+            ?: SidecarRequest(op = "init", dim = config.dim, bitWidth = config.bitWidth)
-        started.outputStream.write(initLine.toByteArray())
+
+        started.outputStream.write((json.encodeToString(SidecarRequest.serializer(), bootstrap) + "\n").toByteArray())
         started.outputStream.flush()
-        val response = started.inputStream.bufferedReader().readLine()
+        val response = reader!!.readLine()?.let { json.decodeFromString(SidecarResponse.serializer(), it) }
-            ?.let { json.decodeFromString(SidecarResponse.serializer(), it) }
         if (response?.ok != true) {
-            throw RuntimeException("TurboVec init failed: ${response?.error ?: "no response"}")
+            throw RuntimeException("TurboVec ${bootstrap.op} failed: ${response?.error ?: "no response"}")
         }
     }
+
+    private companion object {
+        const val SHUTDOWN_TIMEOUT_MS = 2000L
+    }
 }