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 5b0127d7..ec85842d 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 @@ -291,6 +291,17 @@ fun main() { privilegedLocations = privilegedLocations, allowedWorkspaceRoots = allowedWorkspaceRoots, ) + val projectMemory = if (correxConfig.project.enabled) { + com.correx.apps.server.memory.ProjectMemoryService( + config = correxConfig.project, + embedder = embedder, + l3MemoryStore = l3MemoryStore, + journalRepository = decisionJournalRepository, + eventStore = eventStore, + ) + } else { + null + } val module = ServerModule( orchestrator = orchestrator, eventStore = eventStore, @@ -310,6 +321,7 @@ fun main() { resourceProbe = resourceProbe, workspaceResolver = workspaceResolver, narrationMaxPerRun = routerConfig.narration.maxPerRun, + projectMemory = projectMemory, ) module.start() log.info("==============================") diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 6c69d6d6..1f90c930 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -67,6 +67,8 @@ class ServerModule( val moduleScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default), approvalCoordinator: ApprovalCoordinator? = null, val narrationMaxPerRun: Int = 100, + // Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path). + val projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null, ) { val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator( orchestrator = orchestrator, @@ -127,8 +129,12 @@ class ServerModule( ) { activeSessionJobs.computeIfAbsent(sessionId) { moduleScope.launch { + // Seed prior-session project memory before the run so stages see it in context. + projectMemory?.let { pm -> runCatching { pm.retrieveAndSeed(sessionId, pm.repoRoot()) } } runCatching { orchestrator.run(sessionId, graph, sessionConfig) + // Distil this run's decisions into durable project memory on completion. + projectMemory?.let { pm -> pm.persist(sessionId, pm.repoRoot()) } }.onFailure { ex -> log.error("runSession: session={} failed: {}", sessionId.value, ex.message, ex) eventStore.append( diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt b/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt new file mode 100644 index 00000000..af0494da --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt @@ -0,0 +1,111 @@ +package com.correx.apps.server.memory + +import com.correx.core.config.ProjectConfig +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.SteeringNoteAddedEvent +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.inference.Embedder +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.journal.ProjectMemoryDistiller +import com.correx.core.router.l3.L3MemoryEntry +import com.correx.core.router.l3.L3MemoryStore +import com.correx.core.router.l3.L3Query +import kotlinx.coroutines.CancellationException +import kotlinx.datetime.Clock +import org.slf4j.LoggerFactory +import java.util.UUID + +private val log = LoggerFactory.getLogger(ProjectMemoryService::class.java) + +// Over-fetch factor: project lines share the L3 store with chat turns, so query wider +// than memoryK and post-filter by the repo-scoped turnId tag. +private const val OVERFETCH = 4 + +/** + * Cross-session, repo-scoped memory. Project lines are tagged in the L3 store via + * `turnId = "project:"` (no schema change). On session end the decision journal + * is distilled and persisted; on session start prior-session project lines are retrieved and + * seeded into the new session as a recorded steering note (invariant #9: the L3 read is + * recorded as an event, never re-observed at replay), so the decision-journal pinning surfaces + * them to every stage. All operations are gated on [ProjectConfig.enabled]. + */ +class ProjectMemoryService( + private val config: ProjectConfig, + private val embedder: Embedder, + private val l3MemoryStore: L3MemoryStore, + private val journalRepository: DefaultDecisionJournalRepository, + private val eventStore: EventStore, + private val distiller: ProjectMemoryDistiller = ProjectMemoryDistiller(), +) { + private fun tag(repoRoot: String) = "project:$repoRoot" + + /** Resolved repo-root key: configured [ProjectConfig.root], else the working dir. */ + fun repoRoot(): String = config.root.ifBlank { System.getProperty("user.dir") ?: "." } + + /** Distil the session's decision journal into durable, repo-tagged L3 entries. */ + suspend fun persist(sessionId: SessionId, repoRoot: String) { + if (!config.enabled) return + val lines = distiller.distill(journalRepository.getJournal(sessionId), repoRoot) + lines.forEach { line -> + runCatching { + val vector = embedder.embed(line) + l3MemoryStore.store( + L3MemoryEntry( + id = UUID.randomUUID().toString(), + sessionId = sessionId, + turnId = tag(repoRoot), + text = line, + vector = vector, + timestampMs = Clock.System.now().toEpochMilliseconds(), + ), + ) + }.exceptionOrNull()?.let { e -> + if (e is CancellationException) throw e + log.warn("project memory persist failed for {}: {}", repoRoot, e.message) + } + } + } + + /** + * Retrieve prior-session project lines for [repoRoot] and seed them into [sessionId] as a + * steering note. Returns the lines seeded (empty if disabled or none found). + */ + suspend fun retrieveAndSeed(sessionId: SessionId, repoRoot: String): List { + if (!config.enabled) return emptyList() + val lines = runCatching { + val vector = embedder.embed(repoRoot) + l3MemoryStore.query(L3Query(vector = vector, k = config.memoryK * OVERFETCH)) + .map { it.entry } + .filter { it.turnId == tag(repoRoot) && it.sessionId != sessionId } + .map { it.text } + .distinct() + .take(config.memoryK) + }.getOrElse { e -> + if (e is CancellationException) throw e + log.warn("project memory retrieval failed for {}: {}", repoRoot, e.message) + emptyList() + } + if (lines.isEmpty()) return emptyList() + + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = SteeringNoteAddedEvent( + sessionId = sessionId, + content = "Project memory (prior sessions):\n" + lines.joinToString("\n"), + ), + ), + ) + return lines + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceTest.kt new file mode 100644 index 00000000..29012032 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceTest.kt @@ -0,0 +1,89 @@ +package com.correx.apps.server.memory + +import com.correx.core.config.ProjectConfig +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.SteeringNoteAddedEvent +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId +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.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.util.UUID + +private class OnesEmbedder(override val dimension: Int = 8) : Embedder { + override suspend fun embed(text: String): FloatArray = FloatArray(dimension) { 1f } +} + +class ProjectMemoryServiceTest { + + private fun store(sessionId: SessionId, payload: com.correx.core.events.events.EventPayload, es: InMemoryEventStore) = + runBlocking { + es.append( + NewEvent( + EventMetadata(EventId(UUID.randomUUID().toString()), sessionId, Clock.System.now(), 1, null, null), + payload, + ), + ) + } + + private fun service(config: ProjectConfig, es: InMemoryEventStore, l3: InMemoryL3MemoryStore) = + ProjectMemoryService( + config = config, + embedder = OnesEmbedder(), + l3MemoryStore = l3, + journalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(es, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ), + eventStore = es, + ) + + @Test + fun `decisions persisted in one session are retrieved and seeded in the next`(): Unit = runBlocking { + val eventStore = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val config = ProjectConfig(enabled = true, root = "/repo", memoryK = 5) + + val sessionA = SessionId("A") + store(sessionA, SteeringNoteAddedEvent(sessionA, "use jwt for auth"), eventStore) + store(sessionA, TransitionExecutedEvent(sessionA, StageId("plan"), StageId("impl"), TransitionId("t1")), eventStore) + + service(config, eventStore, l3).persist(sessionA, "/repo") + + val sessionB = SessionId("B") + val seeded = service(config, eventStore, l3).retrieveAndSeed(sessionB, "/repo") + + assertTrue(seeded.any { it.contains("use jwt for auth") }, "expected prior decision retrieved") + val note = eventStore.read(sessionB).mapNotNull { it.payload as? SteeringNoteAddedEvent }.firstOrNull() + assertTrue(note != null && note.content.contains("Project memory"), "expected seeded steering note in session B") + } + + @Test + fun `disabled project memory is a no-op`(): Unit = runBlocking { + val eventStore = InMemoryEventStore() + val l3 = InMemoryL3MemoryStore() + val config = ProjectConfig(enabled = false, root = "/repo") + + val sessionA = SessionId("A") + store(sessionA, SteeringNoteAddedEvent(sessionA, "secret"), eventStore) + service(config, eventStore, l3).persist(sessionA, "/repo") + + val sessionB = SessionId("B") + val seeded = service(config, eventStore, l3).retrieveAndSeed(sessionB, "/repo") + + assertTrue(seeded.isEmpty()) + assertFalse(eventStore.read(sessionB).any { it.payload is SteeringNoteAddedEvent }) + } +}