claude-office/server/watch.mjs

/**
 * Claude Office live watcher.
 *
 * Tails the .jsonl session transcripts under ~/.claude/projects/ and
 * translates appended lines into office events, streamed to the frontend
 * over Server-Sent Events at http://localhost:5179/events.
 *
 * 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 || 5179);
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:5180",
  "http://127.0.0.1:5180",
]);
const TASK_TIMEOUT_MS = 15 * 60_000;
const MAX_QUEUE = 3;
/** Headless permission mode — change to taste (acceptEdits / plan / bypassPermissions). */
const PERMISSION_MODE = process.env.OFFICE_PERMISSION_MODE || "acceptEdits";

const PROJECTS_ROOT = path.join(homedir(), ".claude", "projects");

/** slug → { slug, name, cwd, lastActivity } */
const projects = new Map();
/** absolute file path → { offset, remainder, slug } */
const tails = new Map();
/** tool_use_id → { slug, desc, kind, startedAt } — pending Agent (subagent) calls */
const pendingSubs = 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 prettifyName(slug, cwd) {
  if (cwd) {
    const seg = cwd.split(/[\\/]/).filter(Boolean);
    if (seg.length > 0) return seg[seg.length - 1];
  }
  const parts = slug.split("-").filter(Boolean);
  return parts[parts.length - 1] || slug;
}

/** Read the tail of the newest transcript to discover the project cwd. */
async function sniffCwd(file) {
  try {
    const stat = await fs.stat(file);
    const start = Math.max(0, stat.size - 16384);
    const buf = await readRange(file, start, stat.size);
    const lines = buf.toString("utf8").split("\n");
    for (let i = lines.length - 1; i >= 0; i--) {
      const m = lines[i].match(/"cwd":"((?:[^"\\]|\\.)*)"/);
      if (m) return JSON.parse(`"${m[1]}"`);
    }
  } catch {
    /* ignore */
  }
  return null;
}

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);
  });
}

function textOf(content) {
  if (typeof content === "string") return content;
  if (!Array.isArray(content)) return "";
  return content
    .filter((c) => c && c.type === "text" && typeof c.text === "string")
    .map((c) => c.text)
    .join(" ");
}

function handleLine(slug, line) {
  let o;
  try {
    o = JSON.parse(line);
  } catch {
    return;
  }
  const proj = projects.get(slug);
  if (proj) {
    proj.lastActivity = Date.now();
    if (!proj.cwd && typeof o.cwd === "string") {
      proj.cwd = o.cwd;
      proj.name = prettifyName(slug, o.cwd);
    }
  }

  const msg = o.message;
  const content = msg?.content;

  // Subagent completion: any tool_result whose id matches a pending Agent call.
  if (Array.isArray(content)) {
    for (const c of content) {
      if (c && c.type === "tool_result" && c.tool_use_id && pendingSubs.has(c.tool_use_id)) {
        const sub = pendingSubs.get(c.tool_use_id);
        pendingSubs.delete(c.tool_use_id);
        broadcast({ type: "subagent_end", slug: sub.slug, subId: c.tool_use_id });
      }
    }
  }

  // Skip subagent-internal lines for the main pawn's status.
  if (o.isSidechain) return;

  if (o.type === "assistant" && msg?.role === "assistant") {
    if (Array.isArray(content)) {
      for (const c of content) {
        if (!c || c.type !== "tool_use") continue;
        if (c.name === "Agent" || c.name === "Task") {
          const input = c.input ?? {};
          const sub = {
            slug,
            desc: input.description || input.prompt?.slice(0, 40) || "subagent",
            kind: input.subagent_type || "Task",
            startedAt: Date.now(),
          };
          pendingSubs.set(c.id, sub);
          broadcast({ type: "subagent_spawn", slug, subId: c.id, desc: sub.desc, kind: sub.kind });
        } else {
          broadcast({ type: "tool", slug, tool: c.name });
        }
      }
    }
    const text = textOf(content).trim();
    if (text) broadcast({ type: "speech", slug, text: text.slice(0, 80) });
    return;
  }

  if (o.type === "user" && msg?.role === "user" && !o.isMeta) {
    const hasToolResult =
      Array.isArray(content) && content.some((c) => c && c.type === "tool_result");
    if (!hasToolResult) {
      const text = textOf(content).trim();
      if (text) broadcast({ type: "prompt", slug });
    }
  }
}

async function pollOnce(initial) {
  let dirs;
  try {
    dirs = await fs.readdir(PROJECTS_ROOT, { withFileTypes: true });
  } catch {
    return;
  }

  for (const d of dirs) {
    if (!d.isDirectory()) continue;
    const slug = d.name;
    const dir = path.join(PROJECTS_ROOT, slug);

    let files;
    try {
      files = (await fs.readdir(dir)).filter((f) => f.endsWith(".jsonl"));
    } catch {
      continue;
    }

    let newestMtime = 0;
    let newestFile = null;

    for (const f of files) {
      const full = path.join(dir, f);
      let stat;
      try {
        stat = await fs.stat(full);
      } catch {
        continue;
      }
      if (stat.mtimeMs > newestMtime) {
        newestMtime = stat.mtimeMs;
        newestFile = full;
      }

      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: "", slug };
        tails.set(full, tail);
        if (!bornNow) continue;
      }
      if (stat.size < tail.offset) {
        tail.offset = stat.size; // truncated / rotated
        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(slug, line);
        }
      }
    }

    // Register the project if recently active.
    const ageDays = (Date.now() - newestMtime) / 86400000;
    if (newestFile && ageDays <= ACTIVE_WINDOW_DAYS && !projects.has(slug)) {
      const cwd = await sniffCwd(newestFile);
      projects.set(slug, {
        slug,
        name: prettifyName(slug, cwd),
        cwd,
        lastActivity: newestMtime,
      });
      if (!initial) {
        broadcast({ type: "project", slug, name: projects.get(slug).name, lastActivity: newestMtime });
      }
    }
  }
}

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

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

/** Newest transcript's basename = session id, for --resume. */
async function newestSessionId(slug) {
  try {
    const dir = path.join(PROJECTS_ROOT, slug);
    const files = (await fs.readdir(dir)).filter((f) => f.endsWith(".jsonl"));
    let best = null;
    let bestM = 0;
    for (const f of files) {
      const st = await fs.stat(path.join(dir, f));
      if (st.mtimeMs > bestM) {
        bestM = st.mtimeMs;
        best = f;
      }
    }
    return best ? best.replace(/\.jsonl$/, "") : null;
  } catch {
    return null;
  }
}

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

  const args = ["-p", "--permission-mode", PERMISSION_MODE];
  if (resume) {
    const sessionId = await newestSessionId(slug);
    if (sessionId && /^[0-9a-fA-F-]+$/.test(sessionId)) args.push("--resume", sessionId);
  }

  // Prompt goes through stdin — never through shell arguments.
  const child = spawn("claude", 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 --resume ── */

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

async 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 = { sessionId: null, child: null };
    chatThreads.set(slug, thread);
  }
  if (thread.child) return { ok: false, error: "busy" };

  let resumeId = thread.sessionId;
  if (!resumeId && fromProject) {
    const sid = await newestSessionId(slug);
    if (sid && /^[0-9a-fA-F-]+$/.test(sid)) resumeId = sid;
  }

  const args = [
    "-p",
    "--output-format",
    "stream-json",
    "--verbose",
    "--permission-mode",
    PERMISSION_MODE,
  ];
  if (resumeId) args.push("--resume", resumeId);

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

  let stderrTail = "";
  let stdoutRemainder = "";
  let sawResult = false;

  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 (o.session_id) thread.sessionId = o.session_id;
      if (o.type === "system" && o.subtype === "init") {
        broadcast({ type: "chat_start", slug });
      } else if (o.type === "assistant") {
        const text = (o.message?.content ?? [])
          .filter((b) => b && b.type === "text" && b.text)
          .map((b) => b.text)
          .join("\n")
          .trim();
        if (text) broadcast({ type: "chat_text", slug, text });
      } else if (o.type === "result") {
        sawResult = true;
        broadcast({
          type: "chat_done",
          slug,
          ok: o.subtype === "success",
          text: typeof o.result === "string" ? o.result : "",
        });
      }
    }
  });

  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 (!sawResult) {
      broadcast({
        type: "chat_done",
        slug,
        ok: false,
        text: "",
        error: 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 = await 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;
    }
    await 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: [...pendingSubs.entries()].map(([subId, s]) => ({
      subId,
      slug: s.slug,
      desc: s.desc,
      kind: s.kind,
    })),
  };
}

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(`✳ Claude Office watcher`);
  console.log(`  watching  ${PROJECTS_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);