codex-office/server/watch.mjs

/**
 * Codex Office live watcher.
 *
 * Tails the rollout-*.jsonl session transcripts under ~/.codex/sessions/
 * (organised as YYYY/MM/DD/) and translates appended lines into office
 * events, streamed over Server-Sent Events at http://localhost:5181/events.
 *
 * Sessions are grouped into office agents by their working directory (cwd),
 * so each project appears as one pawn.
 *
 * Zero dependencies — plain Node 18+.
 */
import { createServer } from "node:http";
import { createReadStream, promises as fs } from "node:fs";
import { spawn } from "node:child_process";
import { homedir } from "node:os";
import path from "node:path";

const PORT = Number(process.env.PORT || 5181);
const POLL_MS = 1500;
const ACTIVE_WINDOW_DAYS = 7;

/** Origins allowed to dispatch tasks (browser CSRF guard; server is localhost-only). */
const ALLOWED_ORIGINS = new Set([
  "http://localhost:5173",
  "http://127.0.0.1:5173",
  "http://localhost:5182",
  "http://127.0.0.1:5182",
]);
const TASK_TIMEOUT_MS = 15 * 60_000;
const MAX_QUEUE = 3;

const SESSIONS_ROOT = path.join(homedir(), ".codex", "sessions");

/** slug → { slug, name, cwd, lastActivity } */
const projects = new Map();
/** absolute file path → { offset, remainder, cwd, slug } */
const tails = new Map();

const clients = new Set();

function broadcast(event) {
  const data = `data: ${JSON.stringify(event)}\n\n`;
  for (const res of clients) res.write(data);
}

function slugOf(cwd) {
  return cwd.replace(/[^a-zA-Z0-9]+/g, "-");
}

function nameOf(cwd) {
  const seg = cwd.split(/[\\/]/).filter(Boolean);
  return seg[seg.length - 1] || cwd;
}

function registerProject(cwd, lastActivity) {
  const slug = slugOf(cwd);
  let proj = projects.get(slug);
  if (!proj) {
    proj = { slug, name: nameOf(cwd), cwd, lastActivity };
    projects.set(slug, proj);
    broadcast({ type: "project", slug, name: proj.name, lastActivity });
  } else if (lastActivity > proj.lastActivity) {
    proj.lastActivity = lastActivity;
  }
  return proj;
}

function readRange(file, start, end) {
  return new Promise((resolve, reject) => {
    const chunks = [];
    createReadStream(file, { start, end: Math.max(start, end - 1) })
      .on("data", (c) => chunks.push(c))
      .on("end", () => resolve(Buffer.concat(chunks)))
      .on("error", reject);
  });
}

/** Rollouts put session_meta (with cwd) in the first lines — sniff the head. */
async function sniffCwd(file, size) {
  try {
    const buf = await readRange(file, 0, Math.min(size, 8192));
    const m = buf.toString("utf8").match(/"cwd":"((?:[^"\\]|\\.)*)"/);
    if (m) return JSON.parse(`"${m[1]}"`);
  } catch {
    /* ignore */
  }
  return null;
}

function handleLine(tail, line) {
  let o;
  try {
    o = JSON.parse(line);
  } catch {
    return;
  }
  const p = o.payload ?? {};

  // cwd can appear in session_meta and change per turn_context
  if ((o.type === "session_meta" || o.type === "turn_context") && typeof p.cwd === "string") {
    tail.cwd = p.cwd;
    tail.slug = slugOf(p.cwd);
    registerProject(p.cwd, Date.now());
    return;
  }

  if (!tail.slug) return;
  const slug = tail.slug;
  const proj = projects.get(slug);
  if (proj) proj.lastActivity = Date.now();

  if (o.type === "event_msg") {
    switch (p.type) {
      case "user_message":
        broadcast({ type: "prompt", slug });
        break;
      case "task_started":
        broadcast({ type: "thinking", slug });
        break;
      case "agent_message":
        if (p.message) broadcast({ type: "speech", slug, text: String(p.message).slice(0, 80) });
        break;
      case "task_complete":
        if (p.last_agent_message)
          broadcast({ type: "speech", slug, text: String(p.last_agent_message).slice(0, 80) });
        break;
      case "mcp_tool_call_begin":
      case "mcp_tool_call_end": {
        const inv = p.invocation ?? {};
        const name = [inv.server, inv.tool].filter(Boolean).join(":") || "mcp";
        broadcast({ type: "tool", slug, tool: name });
        break;
      }
    }
    return;
  }

  if (o.type === "response_item") {
    switch (p.type) {
      case "custom_tool_call":
      case "function_call":
        broadcast({ type: "tool", slug, tool: p.name || "tool" });
        break;
      case "local_shell_call":
        broadcast({ type: "tool", slug, tool: "shell" });
        break;
      case "reasoning":
        broadcast({ type: "thinking", slug });
        break;
    }
  }
}

/** Recursively collect .jsonl files under the YYYY/MM/DD tree (depth ≤ 3). */
async function collectFiles(dir, depth) {
  let entries;
  try {
    entries = await fs.readdir(dir, { withFileTypes: true });
  } catch {
    return [];
  }
  const files = [];
  for (const e of entries) {
    const full = path.join(dir, e.name);
    if (e.isDirectory() && depth < 3) {
      files.push(...(await collectFiles(full, depth + 1)));
    } else if (e.isFile() && e.name.endsWith(".jsonl")) {
      files.push(full);
    }
  }
  return files;
}

async function pollOnce(initial) {
  const files = await collectFiles(SESSIONS_ROOT, 0);
  const cutoff = Date.now() - ACTIVE_WINDOW_DAYS * 86400000;

  for (const full of files) {
    let stat;
    try {
      stat = await fs.stat(full);
    } catch {
      continue;
    }

    let tail = tails.get(full);
    if (!tail) {
      // Initial scan: skip history. Files born while we're running
      // (e.g. a freshly dispatched task session) are read from the top.
      const bornNow = !initial && Date.now() - stat.mtimeMs < 120_000;
      tail = { offset: bornNow ? 0 : stat.size, remainder: "", cwd: null, slug: null, file: full, mtime: stat.mtimeMs };
      tails.set(full, tail);
      const cwd = await sniffCwd(full, stat.size);
      if (cwd) {
        tail.cwd = cwd;
        tail.slug = slugOf(cwd);
        if (stat.mtimeMs >= cutoff) {
          const proj = registerProject(cwd, stat.mtimeMs);
          if (initial) proj.lastActivity = stat.mtimeMs;
        }
      }
      if (!bornNow) continue;
    }
    tail.mtime = stat.mtimeMs;

    if (stat.size < tail.offset) {
      tail.offset = stat.size;
      tail.remainder = "";
    }
    if (stat.size > tail.offset) {
      const buf = await readRange(full, tail.offset, stat.size);
      tail.offset = stat.size;
      const chunk = tail.remainder + buf.toString("utf8");
      const lines = chunk.split("\n");
      tail.remainder = lines.pop() ?? "";
      for (const line of lines) {
        if (line.trim()) handleLine(tail, line);
      }
    }
  }
}

/* ── task dispatch: turn the watcher into a tiny gateway ── */

/** slug → { child, startedAt, queue: [{prompt, resume}] } */
const taskRuns = new Map();

/** Newest rollout for this project → session id (uuid in the filename), for resume. */
function newestSessionId(slug) {
  let best = null;
  let bestM = 0;
  for (const tail of tails.values()) {
    if (tail.slug === slug && tail.mtime > bestM) {
      bestM = tail.mtime;
      best = tail.file;
    }
  }
  if (!best) return null;
  const m = path.basename(best).match(/([0-9a-fA-F]{8}-[0-9a-fA-F-]{27,})\.jsonl$/);
  return m ? m[1] : null;
}

function launchTask(slug, prompt, resume) {
  const proj = projects.get(slug);
  if (!proj?.cwd) return;

  const args = ["exec"];
  if (resume) {
    const sessionId = newestSessionId(slug);
    if (sessionId) args.push("resume", sessionId);
  }
  args.push("--full-auto", "--skip-git-repo-check", "-");

  // Prompt goes through stdin ("-") — never through shell arguments.
  const child = spawn("codex", args, {
    cwd: proj.cwd,
    shell: true,
    stdio: ["pipe", "ignore", "pipe"],
    windowsHide: true,
  });
  const run = { child, startedAt: Date.now(), queue: taskRuns.get(slug)?.queue ?? [] };
  taskRuns.set(slug, run);

  let stderrTail = "";
  child.stderr.on("data", (c) => {
    stderrTail = (stderrTail + c.toString("utf8")).slice(-400);
  });
  child.stdin.write(prompt);
  child.stdin.end();

  const killer = setTimeout(() => child.kill(), TASK_TIMEOUT_MS);
  broadcast({ type: "task_accepted", slug, resume: Boolean(resume) });

  child.on("close", (code) => {
    clearTimeout(killer);
    broadcast({
      type: "task_done",
      slug,
      code,
      error: code === 0 ? undefined : stderrTail.trim().slice(-200) || undefined,
    });
    const next = run.queue.shift();
    if (next) {
      launchTask(slug, next.prompt, next.resume);
    } else {
      taskRuns.delete(slug);
    }
  });
  child.on("error", (err) => {
    clearTimeout(killer);
    taskRuns.delete(slug);
    broadcast({ type: "task_done", slug, code: -1, error: String(err).slice(0, 200) });
  });
}

/* ── two-way chat: one office thread per project, stitched with `codex exec resume` ── */

/** slug → { threadId, child } */
const chatThreads = new Map();

function launchChat(slug, message, fromProject) {
  const proj = projects.get(slug);
  if (!proj?.cwd) return { ok: false, error: "unknown project" };

  let thread = chatThreads.get(slug);
  if (!thread) {
    thread = { threadId: null, child: null };
    chatThreads.set(slug, thread);
  }
  if (thread.child) return { ok: false, error: "busy" };

  let resumeId = thread.threadId;
  if (!resumeId && fromProject) {
    const sid = newestSessionId(slug);
    if (sid) resumeId = sid;
  }

  const args = ["exec"];
  if (resumeId) args.push("resume", resumeId);
  args.push("--json", "--full-auto", "--skip-git-repo-check", "-");

  // Message goes through stdin ("-") — never through shell arguments.
  const child = spawn("codex", args, {
    cwd: proj.cwd,
    shell: true,
    stdio: ["pipe", "pipe", "pipe"],
    windowsHide: true,
  });
  thread.child = child;

  let stderrTail = "";
  let stdoutRemainder = "";
  let sawDone = false;
  let lastText = "";

  child.stderr.on("data", (c) => {
    stderrTail = (stderrTail + c.toString("utf8")).slice(-400);
  });

  child.stdout.on("data", (c) => {
    const chunk = stdoutRemainder + c.toString("utf8");
    const lines = chunk.split("\n");
    stdoutRemainder = lines.pop() ?? "";
    for (const line of lines) {
      if (!line.trim()) continue;
      let o;
      try {
        o = JSON.parse(line);
      } catch {
        continue;
      }
      if (typeof o.thread_id === "string") thread.threadId = o.thread_id;
      if (o.type === "thread.started") {
        broadcast({ type: "chat_start", slug });
      } else if (o.type === "item.completed" && o.item?.type === "agent_message" && o.item.text) {
        lastText = String(o.item.text);
        broadcast({ type: "chat_text", slug, text: lastText });
      } else if (o.type === "turn.completed") {
        sawDone = true;
        broadcast({ type: "chat_done", slug, ok: true, text: lastText });
      } else if (o.type === "turn.failed" || o.type === "error") {
        sawDone = true;
        broadcast({
          type: "chat_done",
          slug,
          ok: false,
          text: "",
          error: String(o.error?.message ?? o.message ?? "turn failed").slice(0, 200),
        });
      }
    }
  });

  child.stdin.write(message);
  child.stdin.end();

  const killer = setTimeout(() => child.kill(), TASK_TIMEOUT_MS);
  child.on("close", (code) => {
    clearTimeout(killer);
    thread.child = null;
    if (!sawDone) {
      broadcast({
        type: "chat_done",
        slug,
        ok: code === 0 && Boolean(lastText),
        text: lastText,
        error: code === 0 ? undefined : stderrTail.trim().slice(-200) || `exit ${code}`,
      });
    }
  });
  child.on("error", (err) => {
    clearTimeout(killer);
    thread.child = null;
    broadcast({ type: "chat_done", slug, ok: false, text: "", error: String(err).slice(0, 200) });
  });

  return { ok: true, resumed: Boolean(resumeId) };
}

async function handleChatRequest(req, res) {
  const cors = corsFor(req);
  if (req.method === "OPTIONS") {
    res.writeHead(Object.keys(cors).length ? 204 : 403, cors);
    res.end();
    return;
  }
  if (!Object.keys(cors).length) {
    res.writeHead(403, { "Content-Type": "application/json" });
    res.end(JSON.stringify({ ok: false, error: "origin not allowed" }));
    return;
  }
  const reply = (status, payload) => {
    res.writeHead(status, { ...cors, "Content-Type": "application/json" });
    res.end(JSON.stringify(payload));
  };
  try {
    const body = JSON.parse(await readBody(req));
    const slug = String(body.slug ?? "");
    if (!projects.has(slug)) return reply(404, { ok: false, error: "unknown project" });

    if (body.reset) {
      const thread = chatThreads.get(slug);
      if (thread?.child) thread.child.kill();
      chatThreads.delete(slug);
      return reply(200, { ok: true });
    }

    const message = String(body.message ?? "").trim();
    if (!message || message.length > 8000) {
      return reply(400, { ok: false, error: "message must be 1–8000 chars" });
    }
    const result = launchChat(slug, message, Boolean(body.fromProject));
    reply(result.ok ? 200 : 409, result);
  } catch {
    reply(400, { ok: false, error: "bad request" });
  }
}

function corsFor(req) {
  const origin = req.headers.origin;
  if (origin && ALLOWED_ORIGINS.has(origin)) {
    return {
      "Access-Control-Allow-Origin": origin,
      "Access-Control-Allow-Headers": "content-type",
      "Access-Control-Allow-Methods": "POST, OPTIONS",
      Vary: "Origin",
    };
  }
  return {};
}

function readBody(req, limit = 100_000) {
  return new Promise((resolve, reject) => {
    let size = 0;
    const chunks = [];
    req.on("data", (c) => {
      size += c.length;
      if (size > limit) {
        reject(new Error("too large"));
        req.destroy();
        return;
      }
      chunks.push(c);
    });
    req.on("end", () => resolve(Buffer.concat(chunks).toString("utf8")));
    req.on("error", reject);
  });
}

async function handleTaskRequest(req, res) {
  const cors = corsFor(req);
  if (req.method === "OPTIONS") {
    res.writeHead(Object.keys(cors).length ? 204 : 403, cors);
    res.end();
    return;
  }
  if (!Object.keys(cors).length) {
    res.writeHead(403, { "Content-Type": "application/json" });
    res.end(JSON.stringify({ ok: false, error: "origin not allowed" }));
    return;
  }
  const fail = (status, error) => {
    res.writeHead(status, { ...cors, "Content-Type": "application/json" });
    res.end(JSON.stringify({ ok: false, error }));
  };
  try {
    const body = JSON.parse(await readBody(req));
    const slug = String(body.slug ?? "");
    const prompt = String(body.prompt ?? "").trim();
    const resume = Boolean(body.resume);
    if (!projects.has(slug)) return fail(404, "unknown project");
    if (!prompt || prompt.length > 8000) return fail(400, "prompt must be 1–8000 chars");

    const running = taskRuns.get(slug);
    if (running) {
      if (running.queue.length >= MAX_QUEUE) return fail(429, "queue full");
      running.queue.push({ prompt, resume });
      res.writeHead(202, { ...cors, "Content-Type": "application/json" });
      res.end(JSON.stringify({ ok: true, queued: running.queue.length }));
      return;
    }
    launchTask(slug, prompt, resume);
    res.writeHead(200, { ...cors, "Content-Type": "application/json" });
    res.end(JSON.stringify({ ok: true, queued: 0 }));
  } catch {
    fail(400, "bad request");
  }
}

function snapshot() {
  return {
    type: "snapshot",
    projects: [...projects.values()]
      .sort((a, b) => b.lastActivity - a.lastActivity)
      .map((p) => ({ slug: p.slug, name: p.name, lastActivity: p.lastActivity })),
    subs: [],
  };
}

const server = createServer((req, res) => {
  if (req.url === "/task") {
    handleTaskRequest(req, res);
    return;
  }
  if (req.url === "/chat") {
    handleChatRequest(req, res);
    return;
  }
  if (req.url === "/events") {
    res.writeHead(200, {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
      "Access-Control-Allow-Origin": "*",
    });
    res.write(`data: ${JSON.stringify(snapshot())}\n\n`);
    clients.add(res);
    req.on("close", () => clients.delete(res));
    return;
  }
  res.writeHead(404, { "Access-Control-Allow-Origin": "*" });
  res.end("not found");
});

await pollOnce(true);
// Transcripts are sensitive — never expose beyond this machine.
server.listen(PORT, "127.0.0.1", () => {
  console.log(">_ Codex Office watcher");
  console.log(`   watching  ${SESSIONS_ROOT}`);
  console.log(`   projects  ${projects.size} active in last ${ACTIVE_WINDOW_DAYS} days`);
  console.log(`   SSE       http://localhost:${PORT}/events`);
});
setInterval(() => pollOnce(false).catch(() => {}), POLL_MS);