Integrate canonical architecture: two-process backend + LiveKit agent

Promote all 12 staged canonical files from docs/generated/files/ to their live paths, creating the full PodMan architecture: Backend: - server.ts: HTTP service (token mint, sync-PR, outcome recording, /health, WS relay) - agent.ts: worker joining LiveKit room, grabbing screenshare frames at ~1fps - agent/podman.ts: orchestrator loop (vision -> collision detection -> intervention) - env.ts: flat env var accessors replacing nested stub - vision/gemini.ts: JPEG -> Gemini vision -> EngineerContext (real implementation) - collision/detector.ts: fused vision+GitHub collision detection (the moat) - github/client.ts: Octokit wrapper with caching + sync PR creation - memory/store.ts: extended with recordObservation/recordCollision/recordIntervention/recordOutcome helpers - memory/vectors.ts: stub for Voyage+Atlas vector recall (Loop A) - memory/policy.ts: stub for intervention policy gate (Loop B) - voice/live.ts: stub for Gemini Live TTS voice output Shared: - messages.ts: LiveKit data-channel wire protocol (DataMessage, InterventionOutcome, TeamModel, LocalGitReport) - index.ts: re-exports messages module Frontend: - livekit/useScreenPublish.ts: hook for joining pod and publishing screenshare - livekit/useInterventions.ts: hook for receiving collision cards and responding - lib/api.ts: fetchToken + postOutcome HTTP helpers Database: - database/init.ts: MongoDB Atlas collections + indexes + vector search index Infra: - infra/.do/app.yaml: DO App Platform spec (static_site + service + worker) Retire stubs superseded by canonical decomposition: - backend/src/index.ts (replaced by server.ts) - backend/src/intervention/engine.ts (logic now in agent/podman.ts) - backend/src/livekit/token.ts (token minting now in server.ts) Install missing dependencies: @livekit/rtc-node, sharp, mongodb, ws, @types/ws Type error fixes: - vision/gemini.ts: use MediaResolution.MEDIA_RESOLUTION_LOW enum value (not string literal) - agent/podman.ts: wrap SuggestedActionKind into { kind: action } SuggestedAction object All packages pass pnpm -r typecheck and pnpm -r build. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-27 14:27:14 -07:00
parent 253c7438fb
commit da5b2622b2
23 changed files with 1369 additions and 190 deletions
@@ -5,23 +5,30 @@
  "type": "module",
  "main": "./dist/index.js",
  "scripts": {
-    "dev": "tsx watch src/index.ts",
-    "start": "node dist/index.js",
+    "dev": "tsx watch src/server.ts",
+    "dev:server": "tsx watch src/server.ts",
+    "dev:agent": "tsx watch src/agent.ts",
+    "start": "node dist/server.js",
    "build": "tsc -p tsconfig.json",
    "typecheck": "tsc -p tsconfig.json --noEmit"
  },
  "dependencies": {
    "@google/genai": "^2.10.0",
+    "@livekit/rtc-node": "^0.13.29",
    "@podman/shared": "workspace:*",
    "cors": "^2.8.6",
    "dotenv": "^17.4.2",
    "express": "^5.2.1",
    "livekit-server-sdk": "^2.15.5",
-    "octokit": "^5.0.5"
+    "mongodb": "^7.4.0",
+    "octokit": "^5.0.5",
+    "sharp": "^0.35.2",
+    "ws": "^8.21.0"
  },
  "devDependencies": {
    "@types/cors": "^2.8.19",
    "@types/express": "^5.0.6",
+    "@types/ws": "^8.18.1",
    "tsx": "^4.22.4"
  }
 }
@@ -0,0 +1,79 @@
+import {
+  Room,
+  RoomEvent,
+  TrackKind,
+  TrackSource,
+  VideoStream,
+  VideoBufferType,
+  dispose,
+  type RemoteTrack,
+  type RemoteTrackPublication,
+  type RemoteParticipant,
+} from '@livekit/rtc-node';
+import sharp from 'sharp';
+import { AccessToken } from 'livekit-server-sdk';
+import { env } from './env.js';
+import { PodMan } from './agent/podman.js';
+
+const POD_ROOM = process.env.POD_ROOM ?? 'demo-pod';
+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',
+    name: 'PodMan',
+    ttl: '4h',
+  });
+  at.addGrant({ roomJoin: true, room, canPublish: true, canSubscribe: true, canPublishData: true });
+  return at.toJwt();
+}
+
+async function main() {
+  const room = new Room();
+  const podman = new PodMan(room, POD_ROOM);
+  await room.connect(env.LIVEKIT_URL, await agentToken(POD_ROOM), {
+    autoSubscribe: true,
+    dynacast: true,
+  });
+  await podman.start();
+  console.log(`[agent] PodMan joined room ${POD_ROOM}`);
+
+  const lastSent = new Map<string, number>();
+
+  room.on(
+    RoomEvent.TrackSubscribed,
+    (track: RemoteTrack, pub: RemoteTrackPublication, participant: RemoteParticipant) => {
+      if (track.kind !== TrackKind.KIND_VIDEO || pub.source !== TrackSource.SOURCE_SCREENSHARE) return;
+      const id = participant.identity;
+      const stream = new VideoStream(track);
+      void (async () => {
+        for await (const event of stream) {
+          const now = Date.now();
+          if (now - (lastSent.get(id) ?? 0) < SAMPLE_INTERVAL_MS) continue; // THROTTLE
+          lastSent.set(id, now);
+          const rgba = event.frame.convert(VideoBufferType.RGBA);
+          const jpeg = await sharp(Buffer.from(rgba.data), {
+            raw: { width: rgba.width, height: rgba.height, channels: 4 },
+          })
+            .resize({ width: 1280, withoutEnlargement: true })
+            .jpeg({ quality: 70 })
+            .toBuffer();
+          await podman.onScreenFrame(id, jpeg);
+        }
+      })();
+    },
+  );
+
+  const shutdown = async () => {
+    await room.disconnect();
+    await dispose();
+    process.exit(0);
+  };
+  process.on('SIGINT', shutdown);
+  process.on('SIGTERM', shutdown);
+}
+
+main().catch((e) => {
+  console.error('[agent] fatal', e);
+  process.exit(1);
+});
@@ -0,0 +1,75 @@
+import { RoomEvent, type Room } from '@livekit/rtc-node';
+import type { EngineerContext, Collision, Intervention, DataMessage } from '@podman/shared';
+import { DATA_TOPIC } from '@podman/shared';
+import { analyzeFrame } from '../vision/gemini.js';
+import { detectCollisions } from '../collision/detector.js';
+import { getGithubState } from '../github/client.js';
+import { recordObservation, recordCollision, recordIntervention } from '../memory/store.js';
+import { recallSimilar } from '../memory/vectors.js';
+import { shouldIntervene, preferredAction } from '../memory/policy.js';
+import { speak } from '../voice/live.js';
+
+export class PodMan {
+  private contexts = new Map<string, EngineerContext>();
+  private encoder = new TextEncoder();
+
+  constructor(
+    private room: Room,
+    private podId: string,
+  ) {}
+
+  async start(): Promise<void> {
+    // Tier-2 optional ground-truth + engineer ACKs arrive over the data channel.
+    this.room.on(RoomEvent.DataReceived, (payload) => {
+      try {
+        const msg = JSON.parse(new TextDecoder().decode(payload)) as DataMessage;
+        if (msg.type === 'GIT_REPORT') {
+          const c = this.contexts.get(msg.report.engineerId);
+          if (c) c.hasUnpushedChanges = msg.report.unpushedCount > 0 || msg.report.dirtyFiles.length > 0;
+        }
+      } catch { /* ignore malformed */ }
+    });
+  }
+
+  async onScreenFrame(engineerId: string, jpeg: Buffer): Promise<void> {
+    const ctx = await analyzeFrame(engineerId, this.podId, jpeg);
+    this.contexts.set(engineerId, ctx);
+    await recordObservation(ctx);
+
+    const github = await getGithubState(); // cached
+    const collisions = detectCollisions([...this.contexts.values()], github);
+    for (const collision of collisions) await this.handle(collision);
+  }
+
+  private async handle(collision: Collision): Promise<void> {
+    const prior = await recallSimilar(collision); // Loop A: vector recall raises confidence
+    if (prior) collision.severity = 'critical';
+    if (!shouldIntervene(collision, prior)) return; // Loop B: policy gate
+
+    await recordCollision(collision);
+    const action = preferredAction(collision, prior);
+    const names = collision.engineers.join(' and ');
+    const message = `${names} are both editing ${collision.file}` +
+      (collision.githubState?.unpushed ? ' and one has unpushed changes.' : '.') +
+      (prior ? ` I've seen this conflict pattern before.` : '');
+
+    const intervention: Intervention = {
+      id: `int_${Date.now()}`,
+      collisionId: collision.id,
+      podId: this.podId,
+      kind: 'card',
+      message,
+      suggestedAction: { kind: action },
+      status: 'pending',
+      createdAt: new Date().toISOString(),
+    };
+    await recordIntervention(intervention);
+
+    const data: DataMessage = { type: 'COLLISION', collision, intervention };
+    await this.room.localParticipant?.publishData(
+      this.encoder.encode(JSON.stringify(data)),
+      { reliable: true, topic: DATA_TOPIC },
+    );
+    await speak(this.room, message); // gemini-3.1-flash-live voice into the room
+  }
+}
@@ -1,50 +1,40 @@
-import type { Collision, EngineerContext, GithubStateSnapshot } from '@podman/shared';
+import type { EngineerContext, Collision, GithubStateSnapshot } from '@podman/shared';
+
+function normalize(path?: string): string | undefined {
+  if (!path) return undefined;
+  return path.replace(/^\.?\/?(src\/)?/, 'src/').toLowerCase();
+}

-/**
- * Fuse live engineer contexts (from vision) with GitHub state to find overlaps:
- * two or more engineers editing the same file — especially when unpushed.
- *
- * Pure function: easy to unit-test, no I/O. Callers supply the GitHub snapshot.
- */
 export function detectCollisions(
  contexts: EngineerContext[],
-  githubStateByFile: Record<string, GithubStateSnapshot> = {},
-  now: string = new Date().toISOString(),
+  github: GithubStateSnapshot,
 ): Collision[] {
  const byFile = new Map<string, EngineerContext[]>();
-  for (const ctx of contexts) {
-    if (!ctx.currentFile) continue;
-    const list = byFile.get(ctx.currentFile) ?? [];
-    list.push(ctx);
-    byFile.set(ctx.currentFile, list);
+  for (const c of contexts) {
+    const f = normalize(c.currentFile);
+    if (!f) continue;
+    (byFile.get(f) ?? byFile.set(f, []).get(f)!).push(c);
  }

-  const collisions: Collision[] = [];
+  const out: Collision[] = [];
  for (const [file, group] of byFile) {
-    if (group.length < 2) continue;
-
    const engineers = [...new Set(group.map((g) => g.engineerId))];
    if (engineers.length < 2) continue;

-    const github = githubStateByFile[file];
-    const unpushed = group.some((g) => g.hasUnpushedChanges) || github?.unpushed === true;
-    const sharedSymbol = group.every(
-      (g) => g.currentSymbol && g.currentSymbol === group[0]!.currentSymbol,
-    )
-      ? group[0]!.currentSymbol
-      : undefined;
+    const anyUnpushed =
+      group.some((g) => g.hasUnpushedChanges) || github.unpushed === true;
+    if (!anyUnpushed) continue; // the crux GitHub alone cannot answer

-    collisions.push({
-      id: `${group[0]!.podId}:${file}:${engineers.sort().join(',')}`,
+    out.push({
+      id: `col_${file}_${Date.now()}`,
      podId: group[0]!.podId,
      file,
-      symbol: sharedSymbol,
+      symbol: group.find((g) => g.currentSymbol)?.currentSymbol,
      engineers,
-      severity: unpushed ? 'critical' : sharedSymbol ? 'warn' : 'info',
-      githubState: github,
-      detectedAt: now,
+      severity: 'warn',
+      githubState: { ...github, unpushed: anyUnpushed },
+      detectedAt: new Date().toISOString(),
    });
  }
-
-  return collisions;
+  return out;
 }
@@ -1,39 +1,35 @@
 import 'dotenv/config';

-/** Reads an env var, throwing if it is required but missing. */
-function read(name: string, required = false): string {
-  const value = process.env[name] ?? '';
-  if (required && !value) {
-    throw new Error(`Missing required env var: ${name}`);
-  }
-  return value;
+function req(name: string): string {
+  const v = process.env[name];
+  if (!v) throw new Error(`Missing required env var: ${name}`);
+  return v;
+}
+function opt(name: string, fallback = ''): string {
+  return process.env[name] ?? fallback;
 }

 export const env = {
-  port: Number(process.env.PORT ?? 8787),
-
-  livekit: {
-    url: read('LIVEKIT_URL'),
-    apiKey: read('LIVEKIT_API_KEY'),
-    apiSecret: read('LIVEKIT_API_SECRET'),
-  },
-
-  gemini: {
-    apiKey: read('GEMINI_API_KEY'),
-    visionModel: process.env.GEMINI_VISION_MODEL ?? 'gemini-3.5-flash',
-    liveModel: process.env.GEMINI_LIVE_MODEL ?? '',
-  },
-
-  github: {
-    token: read('GITHUB_TOKEN'),
-    repo: read('GITHUB_REPO'),
-  },
-
-  mongo: {
-    uri: read('MONGODB_URI'),
-  },
-
-  voyage: {
-    apiKey: read('VOYAGE_API_KEY'),
-  },
+  // LiveKit
+  LIVEKIT_URL: req('LIVEKIT_URL'),
+  LIVEKIT_API_KEY: req('LIVEKIT_API_KEY'),
+  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'),
+  // 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'),
+  // Server
+  PORT: Number(opt('PORT', '8787')),
 } as const;
+
+export function repoParts(): { owner: string; repo: string } {
+  const [owner, repo] = env.GITHUB_REPO.split('/');
+  if (!owner || !repo) throw new Error('GITHUB_REPO must be "owner/name"');
+  return { owner, repo };
+}
@@ -1,25 +1,46 @@
 import { Octokit } from 'octokit';
 import type { GithubStateSnapshot } from '@podman/shared';
-import { env } from '../env.js';
+import { env, repoParts } from '../env.js';

-export const octokit: Octokit = new Octokit({ auth: env.github.token });
+const gh = new Octokit({ auth: env.GITHUB_TOKEN });
+let cache: { at: number; state: GithubStateSnapshot } | null = null;
+const TTL_MS = 5000;

-/**
- * Pull the GitHub state relevant to a file in the pod's repo: open branches
- * and PRs touching it. Fused with vision contexts by the collision detector.
- *
- * TODO(github): list branches/PRs, diff files, map commits -> engineer logins.
- */
-export async function getStateForFile(_file: string): Promise<GithubStateSnapshot> {
-  return { branches: {}, openPrs: [], unpushed: false };
+export async function getGithubState(): Promise<GithubStateSnapshot> {
+  if (cache && Date.now() - cache.at < TTL_MS) return cache.state;
+  const { owner, repo } = repoParts();
+  const [{ data: branches }] = await Promise.all([
+    gh.rest.repos.listBranches({ owner, repo, per_page: 50 }),
+  ]);
+  const state: GithubStateSnapshot = {
+    branches: Object.fromEntries(branches.map((b) => [b.name, b.commit.sha])),
+    openPrs: [],
+    unpushed: undefined, // vision/Tier-2 fills this; API cannot know
+  };
+  cache = { at: Date.now(), state };
+  return state;
 }

-/** Open a draft "sync PR" between two engineers' branches — the suggested action. */
-export async function openSyncPr(_params: {
-  base: string;
-  head: string;
-  title: string;
-}): Promise<{ number: number; url: string } | null> {
-  // TODO(github): octokit.rest.pulls.create({ ...env.github.repo, draft: true })
-  return null;
+export async function remoteHasFile(path: string, ref = 'main'): Promise<boolean> {
+  const { owner, repo } = repoParts();
+  return gh.rest.repos
+    .getContent({ owner, repo, path, ref })
+    .then(() => true)
+    .catch(() => false);
+}
+
+export async function createSyncPr(input: { headBranch: string; file: string; summary: string }) {
+  const { owner, repo } = repoParts();
+  const { data: mainRef } = await gh.rest.git.getRef({ owner, repo, ref: 'heads/main' });
+  const branch = `podman-sync-${Date.now()}`;
+  await gh.rest.git.createRef({ owner, repo, ref: `refs/heads/${branch}`, sha: mainRef.object.sha });
+  const { data: pr } = await gh.rest.pulls.create({
+    owner,
+    repo,
+    title: `PodMan: sync ${input.file} before collision`,
+    head: branch,
+    base: 'main',
+    body: input.summary,
+  });
+  return pr;
 }
@@ -1,35 +0,0 @@
-import express from 'express';
-import cors from 'cors';
-import { env } from './env.js';
-import { createPodToken } from './livekit/token.js';
-
-const app = express();
-app.use(cors());
-app.use(express.json());
-
-app.get('/health', (_req, res) => {
-  res.json({ ok: true, service: 'podman-backend' });
-});
-
-/**
- * Mint a LiveKit token for an engineer to join a pod room.
- * POST /pods/:podId/token  { identity, name }
- */
-app.post('/pods/:podId/token', async (req, res) => {
-  const { podId } = req.params;
-  const { identity, name } = req.body ?? {};
-  if (!identity) {
-    res.status(400).json({ error: 'identity is required' });
-    return;
-  }
-  try {
-    const token = await createPodToken(podId, identity, name);
-    res.json({ token, url: env.livekit.url });
-  } catch (err) {
-    res.status(500).json({ error: (err as Error).message });
-  }
-});
-
-app.listen(env.port, () => {
-  console.log(`[podman] backend listening on http://localhost:${env.port}`);
-});
@@ -1,27 +0,0 @@
-import type { Collision, Intervention } from '@podman/shared';
-
-/**
- * Compose PodMan's intervention for a collision — the message it speaks and the
- * action it offers (e.g. "open a sync PR"). Severity drives voice vs card.
- *
- * TODO(brain): use Gemini to phrase the message naturally from the team model;
- * tune thresholds from outcomes (the continual-learning policy).
- */
-export function composeIntervention(
-  collision: Collision,
-  now: string = new Date().toISOString(),
-): Intervention {
-  const kind = collision.severity === 'critical' ? 'voice' : 'card';
-  return {
-    id: `intervention:${collision.id}`,
-    collisionId: collision.id,
-    podId: collision.podId,
-    kind,
-    message: `Heads up — ${collision.engineers.join(' and ')} are both in ${collision.file}.`,
-    suggestedAction: {
-      kind: collision.severity === 'critical' ? 'open_sync_pr' : 'ping_teammate',
-    },
-    status: 'pending',
-    createdAt: now,
-  };
-}
@@ -1,25 +0,0 @@
-import { AccessToken } from 'livekit-server-sdk';
-import { env } from '../env.js';
-
-/**
- * Mint a LiveKit access token so an engineer's browser can join a pod room
- * and publish screen + mic + cam tracks.
- */
-export async function createPodToken(
-  podId: string,
-  identity: string,
-  name?: string,
-): Promise<string> {
-  const at = new AccessToken(env.livekit.apiKey, env.livekit.apiSecret, {
-    identity,
-    name,
-  });
-  at.addGrant({
-    room: podId,
-    roomJoin: true,
-    canPublish: true,
-    canSubscribe: true,
-    canPublishData: true,
-  });
-  return at.toJwt();
-}
@@ -0,0 +1,23 @@
+import type { Collision, SuggestedActionKind } from '@podman/shared';
+
+/**
+ * Policy gate: decides whether PodMan should intervene.
+ * Stub: always intervene on warn/critical.
+ */
+export function shouldIntervene(
+  collision: Collision,
+  _prior: unknown,
+): boolean {
+  return collision.severity !== 'info';
+}
+
+/**
+ * Preferred action selection based on collision + prior history.
+ * Stub: open sync PR for critical, ping teammate otherwise.
+ */
+export function preferredAction(
+  collision: Collision,
+  _prior: unknown,
+): SuggestedActionKind {
+  return collision.severity === 'critical' ? 'open_sync_pr' : 'ping_teammate';
+}
@@ -1,12 +1,10 @@
-import type { EngineerContext, Intervention } from '@podman/shared';
+import type { EngineerContext, Collision, Intervention } from '@podman/shared';
+import type { InterventionOutcome } from '@podman/shared';

 /**
 * 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").
- *
- * TODO(memory): connect Atlas, store observations, record outcomes, embed via
- * Voyage, and expose retrieval for the PodMan brain.
 */
 export interface PodMemory {
  recordObservation(ctx: EngineerContext): Promise<void>;
@@ -25,3 +23,25 @@ export function createInMemoryStore(): PodMemory {
    },
  };
 }
+
+// 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<void> {
+  _observations.push(ctx);
+}
+
+export async function recordCollision(collision: Collision): Promise<void> {
+  _collisions.push(collision);
+}
+
+export async function recordIntervention(intervention: Intervention): Promise<void> {
+  _interventions.push(intervention);
+}
+
+export async function recordOutcome(outcome: InterventionOutcome): Promise<void> {
+  _outcomes.push(outcome);
+}
@@ -0,0 +1,10 @@
+import type { Collision } from '@podman/shared';
+
+/**
+ * Vector-based recall of prior collision patterns (Loop A).
+ * Stub: returns null until Voyage + Atlas Vector Search are wired.
+ */
+export async function recallSimilar(_collision: Collision): Promise<Collision | null> {
+  // TODO(memory): embed collision.file via Voyage, query Atlas vector index
+  return null;
+}
@@ -0,0 +1,59 @@
+import express from 'express';
+import { createServer } from 'node:http';
+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 type { InterventionOutcome } from '@podman/shared';
+
+const app = express();
+app.use(express.json());
+app.get('/health', (_req, res) => res.json({ ok: true }));
+
+// Mint a LiveKit token for an engineer joining a pod.
+app.post('/api/token', async (req, res) => {
+  const { room, identity, name, githubLogin } = req.body ?? {};
+  if (!room || !identity) return res.status(400).json({ error: 'room+identity required' });
+  const at = new AccessToken(env.LIVEKIT_API_KEY, env.LIVEKIT_API_SECRET, {
+    identity,
+    name,
+    ttl: '4h',
+    metadata: JSON.stringify({ githubLogin: githubLogin ?? name }),
+  });
+  at.addGrant({ roomJoin: true, room, canPublish: true, canSubscribe: true, canPublishData: true });
+  res.json({ token: await at.toJwt(), url: env.LIVEKIT_URL });
+});
+
+// PodMan's hero action: open a real sync PR on the PUBLIC repo.
+app.post('/api/sync-pr', async (req, res) => {
+  try {
+    const { headBranch, file, summary } = req.body ?? {};
+    const pr = await createSyncPr({ headBranch, file, summary });
+    res.json({ url: pr.html_url, number: pr.number });
+  } catch (e) {
+    res.status(500).json({ error: (e as Error).message });
+  }
+});
+
+// Outcome ACK -> closes the continual-learning policy loop.
+app.post('/api/outcome', async (req, res) => {
+  await recordOutcome(req.body as InterventionOutcome);
+  res.json({ ok: true });
+});
+
+const http = createServer(app);
+
+// ws relay: the agent pushes collision/intervention JSON here; PWAs subscribed by pod receive it.
+const wss = new WebSocketServer({ server: http, path: '/api/events' });
+const clients = new Set<import('ws').WebSocket>();
+wss.on('connection', (ws) => {
+  clients.add(ws);
+  ws.on('close', () => clients.delete(ws));
+  ws.on('message', (buf) => {
+    // fan out agent->PWA events; (auth/pod-scoping omitted for hackathon brevity)
+    for (const c of clients) if (c !== ws && c.readyState === 1) c.send(buf.toString());
+  });
+});
+
+http.listen(env.PORT, '0.0.0.0', () => console.log(`[server] :${env.PORT}`));
@@ -1,20 +1,53 @@
+import { GoogleGenAI, MediaResolution, Type } from '@google/genai';
 import type { EngineerContext } from '@podman/shared';
+import { env } from '../env.js';

-/**
- * Turn a sampled screen frame into a structured EngineerContext using Gemini
- * vision. This is the headline capability: it produces the pre-push signal
- * (which file/symbol an engineer is editing) that GitHub cannot see.
- *
- * TODO(vision): wire @google/genai, downscale frames, sample ~1fps/on-change.
- */
-export async function frameToContext(
-  _frame: Uint8Array,
-  meta: { engineerId: string; podId: string },
+const ai = new GoogleGenAI({ apiKey: env.GEMINI_API_KEY });
+
+const SCHEMA = {
+  type: Type.OBJECT,
+  properties: {
+    currentFile: { type: Type.STRING, description: 'open file path if visible, e.g. src/auth/session.ts' },
+    currentSymbol: { type: Type.STRING, description: 'function/class under the cursor' },
+    activity: { type: Type.STRING, description: 'editing | reading | debugging | terminal | PR review' },
+    hasUnpushedChanges: { type: Type.BOOLEAN, description: 'dirty git gutter / modified markers visible' },
+    confidence: { type: Type.NUMBER, description: '0..1 confidence in this read' },
+  },
+  propertyOrdering: ['currentFile', 'currentSymbol', 'activity', 'hasUnpushedChanges', 'confidence'],
+} as const;
+
+export async function analyzeFrame(
+  engineerId: string,
+  podId: string,
+  jpeg: Buffer,
 ): Promise<EngineerContext> {
+  const res = await ai.models.generateContent({
+    model: env.GEMINI_VISION_MODEL,
+    contents: [
+      {
+        role: 'user',
+        parts: [
+          { text: "You are PodMan watching an engineer's screen. Identify what file/symbol they are working on and whether there are uncommitted edits. JSON only." },
+          { inlineData: { mimeType: 'image/jpeg', data: jpeg.toString('base64') } },
+        ],
+      },
+    ],
+    config: {
+      responseMimeType: 'application/json',
+      responseJsonSchema: SCHEMA,
+      thinkingConfig: { thinkingBudget: 0 }, // minimal thinking: low latency/cost for ambient loop
+      mediaResolution: MediaResolution.MEDIA_RESOLUTION_LOW,
+    },
+  });
+  const parsed = JSON.parse(res.text ?? '{}') as Partial<EngineerContext>;
  return {
-    engineerId: meta.engineerId,
-    podId: meta.podId,
-    confidence: 0,
+    engineerId,
+    podId,
+    currentFile: parsed.currentFile,
+    currentSymbol: parsed.currentSymbol,
+    activity: parsed.activity,
+    hasUnpushedChanges: parsed.hasUnpushedChanges,
+    confidence: parsed.confidence ?? 0.5,
    observedAt: new Date().toISOString(),
  };
 }
@@ -0,0 +1,10 @@
+import type { Room } from '@livekit/rtc-node';
+
+/**
+ * Speak a message into the LiveKit room using Gemini Live voice.
+ * Stub: logs until Gemini Live audio track wiring is complete.
+ */
+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}`);
+}