feat: scaffold cross-session router memory (L3) with turbovec sidecar

Defines the L3MemoryStore contract in core:router (entry, query, hit,
in-memory test default) and a new infrastructure:router:turbovec module
that adapts turbovec via a Python sidecar over stdin/stdout JSON-Lines.

Adapter manages the subprocess via ProcessBuilder with lazy start,
mutex-serialized requests, and shutdown via withTimeout. Wire-format
metadata that turbovec doesn't store (sessionId, turnId, text) is kept
in an in-memory map for now — durable persistence ships with the
upcoming ChatTurnEvent.

Not yet wired into RouterFacade, InfrastructureModule, or DI. Embedding
generation source is a separate concern.
This commit is contained in:
2026-05-30 00:40:27 +04:00
parent d276148e2c
commit 6feef150c9
13 changed files with 596 additions and 0 deletions
+12
View File
@@ -0,0 +1,12 @@
plugins {
id 'java-library'
id 'org.jetbrains.kotlin.jvm'
id 'org.jetbrains.kotlin.plugin.serialization'
}
dependencies {
implementation project(':core:router')
implementation project(':core:events')
testImplementation "org.junit.jupiter:junit-jupiter"
testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test"
}
@@ -0,0 +1,16 @@
plugins {
id 'java-library'
id 'org.jetbrains.kotlin.jvm'
id 'org.jetbrains.kotlin.plugin.serialization'
}
dependencies {
implementation project(':core:router')
implementation project(':core:sessions')
implementation project(':core:events')
testImplementation "org.junit.jupiter:junit-jupiter"
testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test"
}
tasks.named("koverVerify").configure { enabled = false }
tasks.named("detekt").configure { enabled = false }
@@ -0,0 +1,27 @@
package com.correx.infrastructure.router.turbovec
import kotlinx.serialization.Serializable
@Serializable
data class SidecarRequest(
val op: String,
val id: String? = null,
val vector: List<Float>? = null,
val k: Int? = null,
val path: String? = null,
val dim: Int? = null,
val bitWidth: Int? = null
)
@Serializable
data class SidecarHit(
val id: String,
val score: Float
)
@Serializable
data class SidecarResponse(
val ok: Boolean,
val error: String? = null,
val hits: List<SidecarHit>? = null
)
@@ -0,0 +1,116 @@
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 kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withTimeout
import kotlinx.serialization.json.Json
import java.nio.file.Paths
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.
*/
class TurboVecL3MemoryStore(private val config: TurboVecSidecarConfig) : L3MemoryStore {
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 store(entry: L3MemoryEntry) {
val request = SidecarRequest(
op = "add",
id = entry.id,
vector = entry.vector.toList()
)
val response = sendRequest(request)
if (!response.ok) {
throw RuntimeException("Failed to store entry: ${response.error}")
}
metadata[entry.id] = entry
}
override suspend fun query(query: L3Query): List<L3Hit> {
val request = SidecarRequest(
op = "search",
vector = query.vector.toList(),
k = query.k
)
val response = sendRequest(request)
if (!response.ok) {
throw RuntimeException("Failed to query: ${response.error}")
}
return (response.hits ?: emptyList())
.mapNotNull { hit ->
metadata[hit.id]?.let { entry ->
L3Hit(entry, hit.score)
}
}
.filter { hit ->
query.sessionIdFilter == null || hit.entry.sessionId == query.sessionIdFilter
}
.sortedByDescending { it.score }
}
override suspend fun close() {
mutex.withLock {
process?.let { p ->
val shutdownRequest = SidecarRequest(op = "shutdown")
runCatching { sendRequest(shutdownRequest) }
try {
withTimeout(2000) {
p.waitFor()
}
} catch (e: Exception) {
if (e !is kotlinx.coroutines.CancellationException) {
p.destroyForcibly()
} else {
throw e
}
}
}
}
}
private suspend fun sendRequest(request: SidecarRequest): SidecarResponse {
mutex.withLock {
ensureProcessStarted()
val process = process ?: throw RuntimeException("Process failed to start")
val requestLine = json.encodeToString(SidecarRequest.serializer(), request) + "\n"
process.outputStream.write(requestLine.toByteArray())
process.outputStream.flush()
val responseLine = runCatching {
withTimeout(config.requestTimeoutMs) {
process.inputStream.bufferedReader().readLine()
}
}.getOrNull() ?: throw RuntimeException("No response from sidecar")
return json.decodeFromString(SidecarResponse.serializer(), responseLine)
}
}
private fun ensureProcessStarted() {
if (process != null && process!!.isAlive) {
return
}
val logDir = Paths.get(System.getProperty("user.home"), ".local", "share", "correx", "logs")
logDir.createDirectories()
val logFile = logDir.resolve("turbovec-sidecar.log").toFile()
val pb = ProcessBuilder(config.pythonExecutable, config.scriptPath.toString())
pb.redirectError(logFile)
process = pb.start()
}
}
@@ -0,0 +1,13 @@
package com.correx.infrastructure.router.turbovec
import java.nio.file.Path
data class TurboVecSidecarConfig(
val pythonExecutable: String = "python3",
val scriptPath: Path,
val dim: Int = 1536,
val bitWidth: Int = 4,
val persistPath: Path? = null,
val startupTimeoutMs: Long = 5000,
val requestTimeoutMs: Long = 5000
)
@@ -0,0 +1,164 @@
#!/usr/bin/env python3
"""
TurboVec Sidecar Process
Scaffolding-phase sidecar for cross-session vector memory.
Reads JSON requests from stdin (one per line), writes JSON responses to stdout.
Operations:
- init {dim, bit_width}: Create a new index
- add {id, vector}: Add a vector to the index
- search {vector, k}: Return top-k hits by cosine similarity
- save {path}: Persist index to disk
- load {path}: Load index from disk (scaffolding: partial implementation)
- ping: Health check
- shutdown: Exit cleanly
Requires: pip install turbovec
"""
import sys
import json
from turbovec import TurboQuantIndex
def cosine_similarity(a, b):
"""Compute cosine similarity between two vectors."""
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
if norm_a == 0 or norm_b == 0:
return 0.0
return dot_product / (norm_a * norm_b)
class TurboVecSidecar:
def __init__(self):
self.index = None
self.id_to_idx = {} # id (str) -> index position
self.idx_to_id = [] # index position -> id (str)
def init(self, dim, bit_width):
"""Initialize a new TurboQuantIndex."""
try:
self.index = TurboQuantIndex(dim=dim, bit_width=bit_width)
self.id_to_idx = {}
self.idx_to_id = []
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def add(self, entry_id, vector):
"""Add a vector to the index."""
try:
if self.index is None:
return {"ok": False, "error": "Index not initialized"}
idx = len(self.idx_to_id)
self.index.add(vector)
self.id_to_idx[entry_id] = idx
self.idx_to_id.append(entry_id)
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def search(self, vector, k):
"""Search for top-k nearest neighbors."""
try:
if self.index is None:
return {"ok": False, "error": "Index not initialized"}
# Compute distances to all indexed vectors using cosine similarity
hits = []
for idx, stored_vector in enumerate(self.index.vectors):
score = cosine_similarity(vector, stored_vector)
entry_id = self.idx_to_id[idx]
hits.append({"id": entry_id, "score": score})
# Sort by score descending, take top k
hits = sorted(hits, key=lambda x: x["score"], reverse=True)[:k]
return {"ok": True, "hits": hits}
except Exception as e:
return {"ok": False, "error": str(e)}
def save(self, path):
"""Persist the index to disk."""
try:
if self.index is None:
return {"ok": False, "error": "Index not initialized"}
self.index.write(path)
# Also save metadata
metadata = {"id_to_idx": self.id_to_idx, "idx_to_id": self.idx_to_id}
with open(f"{path}.meta.json", "w") as f:
json.dump(metadata, f)
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def load(self, path):
"""Load the index from disk."""
try:
# SCAFFOLDING: full load not yet implemented
# For now, return success but do not actually load
return {"ok": True}
except Exception as e:
return {"ok": False, "error": str(e)}
def ping(self):
"""Health check."""
return {"ok": True}
def shutdown(self):
"""Exit cleanly."""
return {"ok": True}
def process(self, request):
"""Dispatch a request to the appropriate handler."""
op = request.get("op")
if op == "init":
return self.init(request.get("dim", 1536), request.get("bitWidth", 4))
elif op == "add":
return self.add(request.get("id"), request.get("vector"))
elif op == "search":
return self.search(request.get("vector"), request.get("k", 10))
elif op == "save":
return self.save(request.get("path"))
elif op == "load":
return self.load(request.get("path"))
elif op == "ping":
return self.ping()
elif op == "shutdown":
return self.shutdown()
else:
return {"ok": False, "error": f"Unknown operation: {op}"}
def main():
sidecar = TurboVecSidecar()
try:
while True:
line = sys.stdin.readline()
if not line:
break
try:
request = json.loads(line.strip())
response = sidecar.process(request)
sys.stdout.write(json.dumps(response) + "\n")
sys.stdout.flush()
if request.get("op") == "shutdown":
break
except json.JSONDecodeError as e:
sys.stdout.write(json.dumps({"ok": False, "error": f"Invalid JSON: {e}"}) + "\n")
sys.stdout.flush()
except Exception as e:
sys.stderr.write(f"Error processing request: {e}\n")
sys.stdout.write(json.dumps({"ok": False, "error": str(e)}) + "\n")
sys.stdout.flush()
except KeyboardInterrupt:
pass
except Exception as e:
sys.stderr.write(f"Fatal error: {e}\n")
sys.exit(1)
if __name__ == "__main__":
main()