kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): ProjectMemoryService — cross-session repo-scoped memory wiring

Browse Source
This commit is contained in:
kami
2026-06-04 01:06:10 +04:00
parent d552148048
commit 0ca7d4c86b
4 changed files with 218 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/main/kotlin/com/correx/apps/server/Main.kt
+12
View File
@@ -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("==============================")
apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt
+6
View File
@@ -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(
apps/server/src/main/kotlin/com/correx/apps/server/memory/ProjectMemoryService.kt
+111
View File
@@ -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:<repoRoot>"` (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<String> {
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
}
}
apps/server/src/test/kotlin/com/correx/apps/server/memory/ProjectMemoryServiceTest.kt
+89
View File
@@ -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 })
}
}