feat(live): add hands-free Codex voice calls for OMP panes

bridge/live/commands.ts

@@ -0,0 +1,24 @@
+import type { JsonValue } from "../json.ts";
+import { jsonRecord, jsonStringField } from "../stt/json.ts";
+import type { LiveCommand } from "./types.ts";
+
+export const MAX_LIVE_SDP_BYTES = 64 * 1024;
+export const MAX_LIVE_REQUEST_BYTES = 96 * 1024;
+export const LIVE_SESSION_HEADER = "x-collie-omp-session";
+const UUID = /^[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}$/i;
+
+/** Same admission rules at the phone bridge and at the private OMP listener. */
+export function parseLiveCommand(raw: JsonValue): LiveCommand | null {
+  const value = jsonRecord(raw);
+  const requestId = jsonStringField(value?.requestId);
+  if (!value || !requestId || !UUID.test(requestId)) return null;
+  const action = value.action;
+  if (action === "offer") {
+    const sdp = jsonStringField(value.sdp);
+    if (!sdp || Buffer.byteLength(sdp) > MAX_LIVE_SDP_BYTES || !sdp.startsWith("v=0")) return null;
+    return { action, requestId, sdp };
+  }
+  if (action === "mute" && (value.muted === true || value.muted === false)) return { action, requestId, muted: value.muted };
+  if (action === "ready" || action === "heartbeat" || action === "stop") return { action, requestId };
+  return null;
+}

bridge/live/proxy.test.ts

@@ -0,0 +1,121 @@
+import { afterEach, describe, expect, spyOn, test } from "bun:test";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import type { AgentView } from "../types.ts";
+import { OmpLiveProxy } from "./proxy.ts";
+import { LIVE_SESSION_HEADER, parseLiveCommand } from "./commands.ts";
+import type { LiveDescriptor } from "./types.ts";
+
+const dirs: string[] = [];
+afterEach(async () => {
+  await Promise.all(dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })));
+});
+
+async function fixture() {
+  const dir = await mkdtemp(join(tmpdir(), "collie-live-"));
+  dirs.push(dir);
+  const pane: AgentView = {
+    paneId: "w1:p1", workspaceId: "w1", workspaceLabel: "Live", workspaceNumber: 1,
+    tabId: "t1", agent: "omp", status: "idle", cwd: dir, focused: false,
+    agentSession: { kind: "path", value: join(dir, "one.jsonl") },
+  };
+  const descriptor: LiveDescriptor = {
+    version: 1, pid: process.pid, port: 12345, token: "a".repeat(64),
+    paneId: pane.paneId, sessionId: "one", sessionRef: pane.agentSession!,
+  };
+  await Bun.write(join(dir, `${process.pid}.json`), JSON.stringify(descriptor));
+  return { proxy: new OmpLiveProxy(dir), pane, dir, descriptor };
+}
+
+function getRequest() {
+  return new Request("http://localhost/api/pane/w1%3Ap1/live");
+}
+
+const requestId = "12345678-1234-1234-1234-123456789abc";
+
+describe("OMP Live session boundary", () => {
+  test("a fresh session works before its journal exists, but a raced session switch is refused", async () => {
+    const { proxy, pane, dir, descriptor } = await fixture();
+    delete pane.agentSession;
+    let currentSessionId = descriptor.sessionId;
+    const host = Bun.serve({
+      hostname: "127.0.0.1", port: 0,
+      fetch(request) {
+        if (request.headers.get(LIVE_SESSION_HEADER) !== currentSessionId) {
+          return Response.json({ ok: false, error: "Session changed" }, { status: 409 });
+        }
+        return Response.json({ available: true, phase: "idle", muted: false, transcripts: [] });
+      },
+    });
+    try {
+      if (host.port === undefined) throw new Error("No TCP port");
+      descriptor.port = host.port;
+      await Bun.write(join(dir, `${process.pid}.json`), JSON.stringify(descriptor));
+      expect((await (await proxy.handle(getRequest(), pane)).json()).available).toBe(true);
+      currentSessionId = "next-session";
+      expect((await proxy.handle(getRequest(), pane)).status).toBe(409);
+    } finally { await host.stop(true); }
+  });
+
+  test("a pane switched to another conversation cannot reach its previous live host", async () => {
+    const { proxy, pane } = await fixture();
+    const network = spyOn(globalThis, "fetch");
+    try {
+      pane.agentSession = { kind: "path", value: join(pane.cwd, "another.jsonl") };
+      const result = await proxy.handle(getRequest(), pane);
+      expect((await result.json()).available).toBe(false);
+      expect(network).not.toHaveBeenCalled();
+    } finally { network.mockRestore(); }
+  });
+
+  test("a reused pane belonging to a different harness cannot reach OMP", async () => {
+    const { proxy, pane } = await fixture();
+    pane.agent = "codex";
+    const result = await proxy.handle(getRequest(), pane);
+    expect((await result.json()).available).toBe(false);
+  });
+
+  test("local credentials and paths never enter the browser response", async () => {
+    const { proxy, pane, descriptor } = await fixture();
+    const network = spyOn(globalThis, "fetch").mockResolvedValue(Response.json({
+      available: true, phase: "listening", muted: false,
+      transcripts: [{ role: "assistant", text: "Connected", final: true, accessToken: "secret" }],
+      token: descriptor.token, sessionRef: descriptor.sessionRef, accessToken: "secret",
+    }));
+    try {
+      const result = await proxy.handle(getRequest(), pane);
+      expect(await result.json()).toEqual({
+        available: true, phase: "listening", muted: false,
+        transcripts: [{ role: "assistant", text: "Connected", final: true }],
+      });
+      expect(result.headers.get("cache-control")).toBe("no-store");
+    } finally { network.mockRestore(); }
+  });
+
+  test("invalid call commands are refused before local host discovery or network I/O", async () => {
+    const { proxy, pane } = await fixture();
+    const network = spyOn(globalThis, "fetch");
+    try {
+      const result = await proxy.handle(new Request(getRequest(), {
+        method: "POST", headers: { "content-type": "application/json" },
+        body: JSON.stringify({ action: "offer", requestId, sdp: "not an SDP", url: "https://example.com" }),
+      }), pane);
+      expect(result.status).toBe(400);
+      expect(network).not.toHaveBeenCalled();
+    } finally { network.mockRestore(); }
+  });
+});
+
+describe("OMP Live command admission", () => {
+  test("a stale or missing call lease cannot address call controls", () => {
+    expect(parseLiveCommand({ action: "stop" })).toBeNull();
+    expect(parseLiveCommand({ action: "stop", requestId: "../other-session" })).toBeNull();
+    expect(parseLiveCommand({ action: "mute", requestId, muted: "false" })).toBeNull();
+  });
+
+  test("large or non-audio signaling payloads cannot become an unbounded upstream request", () => {
+    expect(parseLiveCommand({ action: "offer", requestId, sdp: "v=0" + "a".repeat(65536) })).toBeNull();
+    expect(parseLiveCommand({ action: "offer", requestId, sdp: "file:///secret" })).toBeNull();
+  });
+});

bridge/live/proxy.ts

@@ -0,0 +1,161 @@
+import { readdir } from "node:fs/promises";
+import { homedir } from "node:os";
+import { join, normalize } from "node:path";
+import { containedRealpath } from "../journal/files.ts";
+import type { AgentSessionRef } from "../journal/types.ts";
+import type { JsonValue } from "../json.ts";
+import { jsonRecord, jsonStringField, jsonNumberField } from "../stt/json.ts";
+import type { AgentView } from "../types.ts";
+import type { LiveCommand, LiveDescriptor, LiveReply, LiveStatus, LiveTranscript } from "./types.ts";
+import { LIVE_SESSION_HEADER, MAX_LIVE_REQUEST_BYTES, MAX_LIVE_SDP_BYTES, parseLiveCommand } from "./commands.ts";
+
+const PHASES = {
+  idle: true, connecting: true, listening: true, working: true, muted: true, error: true,
+} satisfies Record<LiveStatus["phase"], true>;
+
+function sameReference(left: AgentSessionRef, right: AgentSessionRef): boolean {
+  if (left.kind !== right.kind) return false;
+  if (left.kind === "id") return left.value === right.value;
+  const a = normalize(left.value);
+  const b = normalize(right.value);
+  return process.platform === "win32" ? a.toLowerCase() === b.toLowerCase() : a === b;
+}
+
+function descriptorFrom(raw: JsonValue): LiveDescriptor | null {
+  const value = jsonRecord(raw);
+  if (!value || value.version !== 1) return null;
+  const pid = jsonNumberField(value.pid);
+  const port = jsonNumberField(value.port);
+  const token = jsonStringField(value.token);
+  const paneId = jsonStringField(value.paneId);
+  const sessionId = jsonStringField(value.sessionId);
+  const sessionRef = jsonRecord(value.sessionRef);
+  if (pid === null || !Number.isSafeInteger(pid) || pid < 1) return null;
+  if (port === null || !Number.isInteger(port) || port < 1 || port > 65535) return null;
+  if (!token || !/^[\w-]{43,128}$/.test(token) || !paneId || !sessionId) return null;
+  if (!sessionRef || (sessionRef.kind !== "path" && sessionRef.kind !== "id")) return null;
+  const ref = jsonStringField(sessionRef.value);
+  if (!ref) return null;
+  return { version: 1, pid, port, token, paneId, sessionId, sessionRef: { kind: sessionRef.kind, value: ref } };
+}
+
+
+/** Whitelist the browser response: a local descriptor/OAuth field can never ride through a spread. */
+function statusFrom(raw: JsonValue | undefined): LiveStatus | null {
+  const value = jsonRecord(raw);
+  if (!value || (value.available !== true && value.available !== false) || (value.muted !== true && value.muted !== false)) return null;
+  const phase = jsonStringField(value.phase);
+  if (!phase || !Object.hasOwn(PHASES, phase) || !Array.isArray(value.transcripts)) return null;
+  const transcripts: LiveTranscript[] = [];
+  for (const item of value.transcripts.slice(-8)) {
+    const row = jsonRecord(item);
+    const text = jsonStringField(row?.text);
+    if (!row || (row.role !== "user" && row.role !== "assistant") || text === null || (row.final !== true && row.final !== false)) return null;
+    transcripts.push({ role: row.role, text: text.slice(0, 4000), final: row.final });
+  }
+  // SAFETY: PHASES has only the members of the LiveStatus phase union.
+  const result: LiveStatus = { available: value.available, phase: phase as LiveStatus["phase"], muted: value.muted, transcripts };
+  const error = jsonStringField(value.error);
+  if (error !== null) result.error = error.slice(0, 2048);
+  return result;
+}
+
+export function liveJson(body: LiveStatus | LiveReply | { ok: false; error: string }, status = 200): Response {
+  return Response.json(body, { status, headers: { "cache-control": "no-store" } });
+}
+
+function unavailable(error: string): LiveStatus {
+  return { available: false, phase: "idle", muted: false, transcripts: [], error };
+}
+
+/** Only host-authored records are read; the phone supplies a pane id, never a path or URL. */
+export class OmpLiveProxy {
+  constructor(private readonly directory = process.env.COLLIE_OMP_LIVE_DIR || join(homedir(), ".omp", "collie-live")) {}
+
+  private async find(pane: AgentView): Promise<LiveDescriptor | null> {
+    if (pane.agent !== "omp") return null;
+    let names: string[];
+    try {
+      names = await readdir(this.directory);
+    } catch {
+      return null;
+    }
+    let selected: LiveDescriptor | null = null;
+    for (const name of names.slice(0, 512)) {
+      if (!/^\d+\.json$/.test(name)) continue;
+      const safe = await containedRealpath(join(this.directory, name), this.directory);
+      if (!safe) continue;
+      try {
+        const file = Bun.file(safe);
+        if (file.size > 16 * 1024) continue;
+        const descriptor = descriptorFrom(await file.json());
+        if (!descriptor || name !== `${descriptor.pid}.json` || descriptor.paneId !== pane.paneId) continue;
+        // Herdr has no journal reference until a fresh OMP session writes its first turn.
+        // The unique interactive listener remains pinned by session ID on every HTTP request.
+        if (pane.agentSession && !sameReference(descriptor.sessionRef, pane.agentSession)) continue;
+        process.kill(descriptor.pid, 0);
+        // More than one OMP claiming this pane/session is ambiguous, never choose at random.
+        if (selected) return null;
+        selected = descriptor;
+      } catch {
+        // Dead processes and half-written/stale records are not call targets.
+      }
+    }
+    return selected;
+  }
+
+  async handle(req: Request, pane: AgentView): Promise<Response> {
+    if (req.method !== "GET" && req.method !== "POST") return liveJson({ ok: false, error: "Method not allowed" }, 405);
+    let command: LiveCommand | null = null;
+    if (req.method === "POST") {
+      if (req.headers.get("content-type")?.split(";", 1)[0]?.trim() !== "application/json") return liveJson({ ok: false, error: "JSON required" }, 415);
+      const body = await req.text();
+      if (Buffer.byteLength(body) > MAX_LIVE_REQUEST_BYTES) return liveJson({ ok: false, error: "Live request too large" }, 413);
+      try { command = parseLiveCommand(JSON.parse(body)); } catch { /* Invalid JSON is a client error. */ }
+      if (!command) return liveJson({ ok: false, error: "Invalid live request" }, 400);
+    }
+    const descriptor = await this.find(pane);
+    if (!descriptor) {
+      const status = unavailable("Open a new OMP pane to load Collie Live, or restart OMP and resume this session.");
+      return req.method === "GET" ? liveJson(status) : liveJson({ ok: false, error: status.error! }, 503);
+    }
+    try {
+      const response = await fetch(`http://127.0.0.1:${descriptor.port}/live`, {
+        method: req.method,
+        headers: {
+          host: `localhost:${descriptor.port}`, authorization: `Bearer ${descriptor.token}`,
+          "content-type": "application/json", [LIVE_SESSION_HEADER]: descriptor.sessionId,
+        },
+        body: command ? JSON.stringify(command) : undefined,
+        redirect: "error",
+        signal: AbortSignal.any([req.signal, AbortSignal.timeout(command?.action === "ready" || command?.action === "offer" ? 40_000 : 5_000)]),
+      });
+      const body = await response.text();
+      if (Buffer.byteLength(body) > MAX_LIVE_REQUEST_BYTES) throw new Error("Oversized live response");
+      const value: JsonValue = JSON.parse(body);
+      const object = jsonRecord(value);
+      if (!response.ok) {
+        const error = jsonStringField(object?.error)?.slice(0, 2048) ?? "OMP refused the live request";
+        return liveJson({ ok: false, error }, response.status);
+      }
+      if (req.method === "GET") {
+        const status = statusFrom(value);
+        if (!status) throw new Error("Invalid live status");
+        return liveJson(status);
+      }
+      if (!object || object.ok !== true) throw new Error("Invalid live response");
+      const status = statusFrom(object.status);
+      if (!status) throw new Error("Invalid live status");
+      const reply: LiveReply = { ok: true, status };
+      if (command?.action === "offer") {
+        const sdp = jsonStringField(object.sdp);
+        if (!sdp || Buffer.byteLength(sdp) > MAX_LIVE_SDP_BYTES || !sdp.startsWith("v=0")) throw new Error("Invalid SDP answer");
+        reply.sdp = sdp;
+      }
+      return liveJson(reply);
+    } catch {
+      const error = "The OMP live connection is unavailable or timed out. Reopen Live to reconnect.";
+      return req.method === "GET" ? liveJson(unavailable(error)) : liveJson({ ok: false, error }, 502);
+    }
+  }
+}

bridge/live/types.ts

@@ -0,0 +1,40 @@
+import type { AgentSessionRef } from "../journal/types.ts";
+
+export type LivePhase = "idle" | "connecting" | "listening" | "working" | "muted" | "error";
+
+export interface LiveTranscript {
+  role: "user" | "assistant";
+  text: string;
+  final: boolean;
+}
+
+/** No credentials or local paths cross the browser boundary. */
+export interface LiveStatus {
+  available: boolean;
+  phase: LivePhase;
+  muted: boolean;
+  transcripts: LiveTranscript[];
+  error?: string;
+}
+
+export type LiveCommand =
+  | { action: "offer"; requestId: string; sdp: string }
+  | { action: "ready" | "heartbeat" | "stop"; requestId: string }
+  | { action: "mute"; requestId: string; muted: boolean };
+
+export interface LiveReply {
+  ok: true;
+  status: LiveStatus;
+  sdp?: string;
+}
+
+/** Owner-only discovery record. Read by Collie, never returned to the browser. */
+export interface LiveDescriptor {
+  version: 1;
+  pid: number;
+  port: number;
+  token: string;
+  paneId: string;
+  sessionId: string;
+  sessionRef: AgentSessionRef;
+}