From 3fc19194f2d7f3d935e2cd607b4e93afc8f27c44 Mon Sep 17 00:00:00 2001 From: Kartikeya <176560021+karti-ai@users.noreply.github.com> Date: Sat, 27 Jun 2026 15:34:57 -0700 Subject: [PATCH] feat(backend): persist memory to MongoDB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the in-memory store with a real MongoDB-backed store. Adds memory/db.ts (client singleton, typed collections, startup ping + indexes) and rewrites record{Observation,Collision,Intervention,Outcome} to insert into Mongo (best-effort — failures log, don't crash). Server connects on startup and exposes GET /api/memory/stats for verification. Co-Authored-By: Claude Opus 4.8 --- backend/src/memory/db.ts | 51 +++++++++++++++++++++++++++++++ backend/src/memory/store.ts | 60 ++++++++++++++++++------------------- backend/src/server.ts | 17 +++++++++-- 3 files changed, 95 insertions(+), 33 deletions(-) create mode 100644 backend/src/memory/db.ts diff --git a/backend/src/memory/db.ts b/backend/src/memory/db.ts new file mode 100644 index 0000000..8da2414 --- /dev/null +++ b/backend/src/memory/db.ts @@ -0,0 +1,51 @@ +import { MongoClient, type Db, type Collection } from 'mongodb'; +import type { EngineerContext, Collision, Intervention, InterventionOutcome } from '@podman/shared'; +import { env } from '../env.js'; + +let clientPromise: Promise | null = null; + +function getClient(): Promise { + if (!clientPromise) { + const client = new MongoClient(env.MONGODB_URI, { serverSelectionTimeoutMS: 3000 }); + clientPromise = client.connect(); + } + return clientPromise; +} + +/** The PodMan database (name comes from the MONGODB_URI path, e.g. `podman`). */ +export async function getDb(): Promise { + const client = await getClient(); + return client.db(); +} + +export interface PodCollections { + observations: Collection; + collisions: Collection; + interventions: Collection; + outcomes: Collection; +} + +export async function collections(): Promise { + const db = await getDb(); + return { + observations: db.collection('observations'), + collisions: db.collection('collisions'), + interventions: db.collection('interventions'), + outcomes: db.collection('outcomes'), + }; +} + +/** Connect, verify reachability, and create helpful indexes. Call once on startup. */ +export async function initMemory(): Promise { + const db = await getDb(); + await db.command({ ping: 1 }); + const c = await collections(); + await Promise.all([ + c.observations.createIndex({ podId: 1, observedAt: -1 }), + c.observations.createIndex({ engineerId: 1 }), + c.collisions.createIndex({ podId: 1, detectedAt: -1 }), + c.interventions.createIndex({ collisionId: 1 }), + c.outcomes.createIndex({ interventionId: 1 }), + ]); + console.log(`[memory] mongo connected -> ${db.databaseName}`); +} diff --git a/backend/src/memory/store.ts b/backend/src/memory/store.ts index c508514..c44b82e 100644 --- a/backend/src/memory/store.ts +++ b/backend/src/memory/store.ts @@ -1,47 +1,45 @@ -import type { EngineerContext, Collision, Intervention } from '@podman/shared'; -import type { InterventionOutcome } from '@podman/shared'; +import type { EngineerContext, Collision, Intervention, InterventionOutcome } from '@podman/shared'; +import { collections } from './db.js'; /** - * Continual-learning memory: persist observations + intervention outcomes to - * MongoDB Atlas and embed file/feature notes into Voyage vectors so later - * sessions are sharper ("more useful the more you use it"). + * Continual-learning memory: persist observations, collisions, interventions, + * and outcomes to MongoDB so later sessions get sharper. Writes are best-effort + * — a Mongo hiccup logs a warning rather than crashing the agent/server. */ -export interface PodMemory { - recordObservation(ctx: EngineerContext): Promise; - recordOutcome(intervention: Intervention, accepted: boolean): Promise; +async function persist(name: string, fn: () => Promise): Promise { + try { + await fn(); + } catch (err) { + console.warn(`[memory] ${name} persist failed: ${(err as Error).message}`); + } } -/** In-memory stub so the rest of the pipeline can run before Atlas is wired. */ -export function createInMemoryStore(): PodMemory { - const observations: EngineerContext[] = []; - return { - async recordObservation(ctx) { - observations.push(ctx); - }, - async recordOutcome() { - /* no-op until Atlas is wired */ - }, - }; -} - -// Standalone helpers used by the PodMan orchestrator and HTTP server. -const _observations: EngineerContext[] = []; -const _collisions: Collision[] = []; -const _interventions: Intervention[] = []; -const _outcomes: InterventionOutcome[] = []; - export async function recordObservation(ctx: EngineerContext): Promise { - _observations.push(ctx); + await persist('observation', async () => (await collections()).observations.insertOne({ ...ctx })); } export async function recordCollision(collision: Collision): Promise { - _collisions.push(collision); + await persist('collision', async () => (await collections()).collisions.insertOne({ ...collision })); } export async function recordIntervention(intervention: Intervention): Promise { - _interventions.push(intervention); + await persist('intervention', async () => + (await collections()).interventions.insertOne({ ...intervention }), + ); } export async function recordOutcome(outcome: InterventionOutcome): Promise { - _outcomes.push(outcome); + await persist('outcome', async () => (await collections()).outcomes.insertOne({ ...outcome })); +} + +/** Document counts per collection — used by the /api/memory/stats endpoint. */ +export async function memoryStats(): Promise> { + const c = await collections(); + const [observations, collisions, interventions, outcomes] = await Promise.all([ + c.observations.estimatedDocumentCount(), + c.collisions.estimatedDocumentCount(), + c.interventions.estimatedDocumentCount(), + c.outcomes.estimatedDocumentCount(), + ]); + return { observations, collisions, interventions, outcomes }; } diff --git a/backend/src/server.ts b/backend/src/server.ts index 414545d..11c79f6 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -5,7 +5,8 @@ import { WebSocketServer } from 'ws'; import { AccessToken } from 'livekit-server-sdk'; import { env } from './env.js'; import { createSyncPr } from './github/client.js'; -import { recordOutcome } from './memory/store.js'; +import { recordOutcome, memoryStats } from './memory/store.js'; +import { initMemory } from './memory/db.js'; import type { InterventionOutcome } from '@podman/shared'; const app = express(); @@ -44,6 +45,15 @@ app.post('/api/outcome', async (req, res) => { res.json({ ok: true }); }); +// Memory counts — quick way to confirm Mongo persistence is working. +app.get('/api/memory/stats', async (_req, res) => { + try { + res.json(await memoryStats()); + } catch (e) { + res.status(500).json({ error: (e as Error).message }); + } +}); + const http = createServer(app); // ws relay: the agent pushes collision/intervention JSON here; PWAs subscribed by pod receive it. @@ -58,4 +68,7 @@ wss.on('connection', (ws) => { }); }); -http.listen(env.PORT, '0.0.0.0', () => console.log(`[server] :${env.PORT}`)); +http.listen(env.PORT, '0.0.0.0', () => { + console.log(`[server] :${env.PORT}`); + initMemory().catch((e) => console.warn(`[memory] init failed: ${(e as Error).message}`)); +});