feat: harden podman deployment orchestration

backend/package.json

@@ -3,12 +3,15 @@
   "version": "0.0.0",
   "private": true,
   "type": "module",
-  "main": "./dist/index.js",
+  "main": "./dist/server.js",
   "scripts": {
-    "dev": "tsx watch src/server.ts",
+    "dev": "pnpm run dev:server",
     "dev:server": "tsx watch src/server.ts",
     "dev:agent": "tsx watch src/agent.ts",
     "start": "node dist/server.js",
+    "start:server": "node dist/server.js",
+    "start:agent": "node dist/agent.js",
+    "start:all": "node dist/hermes.js",
     "build": "tsc -p tsconfig.json",
     "typecheck": "tsc -p tsconfig.json --noEmit"
   },

backend/src/agent.ts

@@ -16,11 +16,12 @@ import { env } from './env.js';
 import { PodMan } from './agent/podman.js';
 
 const POD_ROOM = process.env.POD_ROOM ?? 'demo-pod';
+const HERMES_IDENTITY = 'podman-hermes';
 const SAMPLE_INTERVAL_MS = 1000; // ~1 fps to the vision model
 
 async function agentToken(room: string): Promise<string> {
   const at = new AccessToken(env.LIVEKIT_API_KEY, env.LIVEKIT_API_SECRET, {
-    identity: 'podman-agent',
+    identity: HERMES_IDENTITY,
     name: 'PodMan',
     ttl: '4h',
   });
@@ -36,7 +37,7 @@ async function main() {
     dynacast: true,
   });
   await podman.start();
-  console.log(`[agent] PodMan joined room ${POD_ROOM}`);
+  console.log(`[agent] ${HERMES_IDENTITY} joined room ${POD_ROOM}`);
 
   const lastSent = new Map<string, number>();
 

backend/src/env.ts

@@ -16,14 +16,15 @@ export const env = {
   LIVEKIT_API_SECRET: req('LIVEKIT_API_SECRET'),
   // Gemini
   GEMINI_API_KEY: req('GEMINI_API_KEY'),
-  GEMINI_VISION_MODEL: opt('GEMINI_VISION_MODEL', 'gemini-3.5-flash'),
-  GEMINI_LIVE_MODEL: opt('GEMINI_LIVE_MODEL', 'gemini-3.1-flash-live-preview'),
+  GEMINI_VISION_MODEL: opt('GEMINI_VISION_MODEL', 'gemini-2.0-flash'),
+  GEMINI_LIVE_MODEL: opt('GEMINI_LIVE_MODEL', 'gemini-live-2.5-flash'),
   // GitHub
   GITHUB_TOKEN: req('GITHUB_TOKEN'),
   GITHUB_REPO: req('GITHUB_REPO'), // owner/name
   // Mongo + Voyage
   MONGODB_URI: req('MONGODB_URI'),
   VOYAGE_API_KEY: opt('VOYAGE_API_KEY'),
+  VOYAGE_EMBEDDING_MODEL: opt('VOYAGE_EMBEDDING_MODEL', 'voyage-4-lite'),
   // Server
   PORT: Number(opt('PORT', '8787')),
 } as const;

backend/src/hermes.ts

@@ -0,0 +1,50 @@
+import { spawn, type ChildProcess } from 'node:child_process';
+import { dirname, join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const here = dirname(fileURLToPath(import.meta.url));
+
+const processMode = process.env.PODMAN_PROCESS ?? 'all';
+type ProcessMode = 'server' | 'agent';
+
+function commandFor(mode: ProcessMode): string[] {
+  return [join(here, `${mode}.js`)];
+}
+
+const modes: ProcessMode[] =
+  processMode === 'server' || processMode === 'agent' ? [processMode] : ['server', 'agent'];
+
+const children: ChildProcess[] = modes.map((mode) =>
+  spawn(process.execPath, commandFor(mode), { stdio: 'inherit' }),
+);
+
+let shuttingDown = false;
+
+function stopAll(signal: NodeJS.Signals = 'SIGTERM') {
+  for (const child of children) {
+    if (!child.killed) child.kill(signal);
+  }
+}
+
+for (const child of children) {
+  child.on('exit', (code, signal) => {
+    if (shuttingDown) return;
+    shuttingDown = true;
+    stopAll();
+    if (signal) {
+      process.kill(process.pid, signal);
+      return;
+    }
+    process.exit(code ?? 1);
+  });
+}
+
+process.on('SIGINT', () => {
+  shuttingDown = true;
+  stopAll('SIGINT');
+});
+
+process.on('SIGTERM', () => {
+  shuttingDown = true;
+  stopAll('SIGTERM');
+});

backend/src/memory/db.ts

@@ -24,6 +24,13 @@ export async function getDb(): Promise<Db> {
   return client.db();
 }
 
+export async function closeMemory(): Promise<void> {
+  if (!clientPromise) return;
+  const client = await clientPromise;
+  clientPromise = null;
+  await client.close();
+}
+
 export interface PodCollections {
   pods: Collection<Pod>;
   observations: Collection<EngineerContext>;
@@ -88,6 +95,10 @@ export async function initMemory(): Promise<void> {
     ['observations.podId', () => c.observations.createIndex({ podId: 1, observedAt: -1 })],
     ['observations.engineerId', () => c.observations.createIndex({ engineerId: 1 })],
     ['collisions.podId', () => c.collisions.createIndex({ podId: 1, detectedAt: -1 })],
+    [
+      'collisions.memorySignature',
+      () => c.collisions.createIndex({ podId: 1, memorySignature: 1 }),
+    ],
     ['interventions.collisionId', () => c.interventions.createIndex({ collisionId: 1 })],
     ['outcomes.interventionId', () => c.outcomes.createIndex({ interventionId: 1 })],
   ];

backend/src/memory/store.ts

@@ -1,5 +1,6 @@
 import type { EngineerContext, Collision, Intervention, InterventionOutcome } from '@podman/shared';
 import { collections } from './db.js';
+import { enrichCollisionMemory } from './vectors.js';
 
 /**
  * Continual-learning memory: persist observations, collisions, interventions,
@@ -22,7 +23,7 @@ export async function recordObservation(ctx: EngineerContext): Promise<void> {
 
 export async function recordCollision(collision: Collision): Promise<void> {
   await persist('collision', async () =>
-    (await collections()).collisions.insertOne({ ...collision }),
+    (await collections()).collisions.insertOne(await enrichCollisionMemory(collision)),
   );
 }
 

backend/src/memory/vectors.ts

@@ -1,10 +1,123 @@
 import type { Collision } from '@podman/shared';
+import { env } from '../env.js';
+import { getDb } from './db.js';
+
+type StoredCollision = Collision & {
+  memorySignature?: string;
+  memoryText?: string;
+  embedding?: number[];
+};
+
+interface VoyageEmbeddingResponse {
+  data?: Array<{ embedding?: number[] }>;
+}
+
+function normalize(value: string | undefined): string {
+  return (value ?? '').trim().toLowerCase();
+}
+
+function signature(collision: Collision): string {
+  return [normalize(collision.file), normalize(collision.symbol)].filter(Boolean).join('#');
+}
+
+function memoryText(collision: Collision): string {
+  return [
+    `file: ${collision.file}`,
+    collision.symbol ? `symbol: ${collision.symbol}` : undefined,
+    `engineers: ${collision.engineers.join(', ')}`,
+    `severity: ${collision.severity}`,
+    collision.githubState?.unpushed ? 'unpushed local changes present' : undefined,
+  ]
+    .filter(Boolean)
+    .join('\n');
+}
+
+async function embed(text: string, inputType: 'document' | 'query'): Promise<number[] | null> {
+  if (!env.VOYAGE_API_KEY) return null;
+  try {
+    const res = await fetch('https://api.voyageai.com/v1/embeddings', {
+      method: 'POST',
+      headers: {
+        Authorization: `Bearer ${env.VOYAGE_API_KEY}`,
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        input: text,
+        model: env.VOYAGE_EMBEDDING_MODEL,
+        input_type: inputType,
+      }),
+    });
+    if (!res.ok) {
+      console.warn(`[memory] voyage embedding failed: ${res.status} ${await res.text()}`);
+      return null;
+    }
+    const body = (await res.json()) as VoyageEmbeddingResponse;
+    return body.data?.[0]?.embedding ?? null;
+  } catch (err) {
+    console.warn(`[memory] voyage embedding failed: ${(err as Error).message}`);
+    return null;
+  }
+}
+
+export async function enrichCollisionMemory(collision: Collision): Promise<StoredCollision> {
+  const text = memoryText(collision);
+  const embedding = await embed(text, 'document');
+  return {
+    ...collision,
+    memorySignature: signature(collision),
+    memoryText: text,
+    ...(embedding ? { embedding } : {}),
+  };
+}
+
+async function recallByVector(collision: Collision): Promise<Collision | null> {
+  const queryVector = await embed(memoryText(collision), 'query');
+  if (!queryVector) return null;
+
+  try {
+    const db = await getDb();
+    const [match] = await db
+      .collection<StoredCollision>('collisions')
+      .aggregate<StoredCollision>([
+        {
+          $vectorSearch: {
+            index: 'collision_embedding',
+            path: 'embedding',
+            queryVector,
+            numCandidates: 50,
+            limit: 5,
+            filter: { podId: collision.podId },
+          },
+        },
+        { $match: { id: { $ne: collision.id } } },
+        { $project: { _id: 0, embedding: 0 } },
+      ])
+      .toArray();
+    return match ?? null;
+  } catch (err) {
+    console.warn(`[memory] vector recall unavailable: ${(err as Error).message}`);
+    return null;
+  }
+}
+
+async function recallBySignature(collision: Collision): Promise<Collision | null> {
+  const db = await getDb();
+  const sig = signature(collision);
+  const match = await db.collection<StoredCollision>('collisions').findOne(
+    {
+      podId: collision.podId,
+      id: { $ne: collision.id },
+      $or: [{ memorySignature: sig }, { file: collision.file }],
+    },
+    { sort: { detectedAt: -1 }, projection: { _id: 0, embedding: 0 } },
+  );
+  return match ?? null;
+}
 
 /**
- * Vector-based recall of prior collision patterns (Loop A).
- * Stub: returns null until Voyage + Atlas Vector Search are wired.
+ * Recall prior collision patterns. Exact Mongo recall is always available;
+ * Voyage + Atlas Vector Search is used first when configured.
  */
-export async function recallSimilar(_collision: Collision): Promise<Collision | null> {
-  // TODO(memory): embed collision.file via Voyage, query Atlas vector index
-  return null;
+export async function recallSimilar(collision: Collision): Promise<Collision | null> {
+  return (await recallByVector(collision)) ?? recallBySignature(collision);
 }

backend/src/server.ts

@@ -6,7 +6,7 @@ import { AccessToken, RoomConfiguration } from 'livekit-server-sdk';
 import { env } from './env.js';
 import { createSyncPr } from './github/client.js';
 import { recordOutcome, memoryStats } from './memory/store.js';
-import { initMemory } from './memory/db.js';
+import { closeMemory, initMemory } from './memory/db.js';
 import {
   listPods,
   getPod,
@@ -161,3 +161,18 @@ http.listen(env.PORT, '0.0.0.0', () => {
     .then(() => seedDefaultPods())
     .catch((e) => console.warn(`[memory] init failed: ${(e as Error).message}`));
 });
+
+let shuttingDown = false;
+async function shutdown(signal: NodeJS.Signals): Promise<void> {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  console.log(`[server] ${signal} received; shutting down`);
+  for (const client of clients) client.close();
+  wss.close();
+  await new Promise<void>((resolve) => http.close(() => resolve()));
+  await closeMemory().catch((e) => console.warn(`[memory] close failed: ${(e as Error).message}`));
+  process.exit(0);
+}
+
+process.on('SIGINT', () => void shutdown('SIGINT'));
+process.on('SIGTERM', () => void shutdown('SIGTERM'));

backend/src/voice/live.ts

@@ -1,10 +1,101 @@
-import type { Room } from '@livekit/rtc-node';
+import {
+  AudioFrame,
+  AudioSource,
+  LocalAudioTrack,
+  TrackPublishOptions,
+  TrackSource,
+  type Room,
+} from '@livekit/rtc-node';
+import { GoogleGenAI, Modality, type LiveServerMessage, type Session } from '@google/genai';
+import { DATA_TOPIC, type DataMessage } from '@podman/shared';
+import { env } from '../env.js';
+
+const SAMPLE_RATE = 24_000;
+const CHANNELS = 1;
+const encoder = new TextEncoder();
+const ai = new GoogleGenAI({ apiKey: env.GEMINI_API_KEY });
+
+async function publishVoiceCue(room: Room, message: string): Promise<void> {
+  const cue: DataMessage = { type: 'VOICE_CUE', text: message };
+  await room.localParticipant?.publishData(encoder.encode(JSON.stringify(cue)), {
+    reliable: true,
+    topic: DATA_TOPIC,
+  });
+}
+
+function audioFrameFromBase64(data: string, mimeType?: string): AudioFrame | null {
+  if (mimeType && !mimeType.includes('audio')) return null;
+  const buf = Buffer.from(data, 'base64');
+  if (buf.byteLength < 2) return null;
+  const bytes = buf.byteLength % 2 === 0 ? buf : buf.subarray(0, buf.byteLength - 1);
+  const samples = new Int16Array(bytes.buffer, bytes.byteOffset, bytes.byteLength / 2);
+  return new AudioFrame(samples, SAMPLE_RATE, CHANNELS, samples.length / CHANNELS);
+}
+
+function audioFrames(message: LiveServerMessage): AudioFrame[] {
+  const parts = message.serverContent?.modelTurn?.parts ?? [];
+  const out: AudioFrame[] = [];
+  for (const part of parts) {
+    const data = part.inlineData?.data;
+    if (!data) continue;
+    const frame = audioFrameFromBase64(data, part.inlineData?.mimeType);
+    if (frame) out.push(frame);
+  }
+  return out;
+}
 
 /**
- * Speak a message into the LiveKit room using Gemini Live voice.
- * Stub: logs until Gemini Live audio track wiring is complete.
+ * Speak a message into the LiveKit room using Gemini Live audio. A data-channel
+ * VOICE_CUE is sent first so clients still get the cue if audio generation or
+ * publishing fails.
  */
-export async function speak(_room: Room, message: string): Promise<void> {
-  // TODO(voice): use Gemini Live streaming TTS -> publish audio track into room
-  console.log(`[voice] ${message}`);
+export async function speak(room: Room, message: string): Promise<void> {
+  await publishVoiceCue(room, message);
+  if (!room.localParticipant) return;
+
+  const source = new AudioSource(SAMPLE_RATE, CHANNELS);
+  const track = LocalAudioTrack.createAudioTrack('podman-hermes-voice', source);
+  const options = new TrackPublishOptions();
+  options.source = TrackSource.SOURCE_MICROPHONE;
+
+  try {
+    const publication = await room.localParticipant.publishTrack(track, options);
+    let done: () => void = () => {};
+    const donePromise = new Promise<void>((resolve) => {
+      done = resolve;
+    });
+    let session: Session | null = null;
+
+    session = await ai.live.connect({
+      model: env.GEMINI_LIVE_MODEL,
+      config: { responseModalities: [Modality.AUDIO] },
+      callbacks: {
+        onmessage: (event) => {
+          void (async () => {
+            for (const frame of audioFrames(event)) await source.captureFrame(frame);
+            if (event.serverContent?.turnComplete || event.serverContent?.generationComplete)
+              done();
+          })();
+        },
+        onerror: (event) => {
+          console.warn(`[voice] Gemini Live error: ${event.message}`);
+          done();
+        },
+        onclose: done,
+      },
+    });
+
+    session.sendClientContent({
+      turns: [{ role: 'user', parts: [{ text: message }] }],
+      turnComplete: true,
+    });
+
+    await Promise.race([donePromise, new Promise((resolve) => setTimeout(resolve, 15_000))]);
+    session.close();
+    if (publication.sid) await room.localParticipant.unpublishTrack(publication.sid, true);
+    await source.close();
+  } catch (err) {
+    console.warn(`[voice] Gemini Live publish failed: ${(err as Error).message}`);
+    await source.close().catch(() => {});
+  }
 }