stack/packages/control-board/src/scan.mjs

// Control board step 1: read each agent's newest pi session log and tmux
// liveness, and write one small JSON status file per agent.
//
// States (plain words):
//   working  - the agent is in the middle of a turn (thinking or running tools)
//   waiting  - the agent finished its turn; it is your move
//   error    - the agent's last turn ended in an error, was aborted, or was cut off; look at it
//   offline  - no tmux session for this agent, or its session no longer runs pi
//   idle     - the agent is live but has no conversation yet
//   unknown  - liveness could not be checked (tmux missing or unresponsive); not a guess
//
// Board files are derived and rewritable. They are not run records. The one
// exception is <boardDir>/seen.json, which holds Jason's "seen" marks and is
// only changed when he clicks; a scan reads it and never rewrites it.

import { existsSync, readFileSync, readdirSync, statSync, mkdirSync, writeFileSync, renameSync } from "node:fs";
import { join, basename, isAbsolute, resolve } from "node:path";
import { homedir } from "node:os";
import { spawnSync } from "node:child_process";

export const STATES = Object.freeze(["working", "waiting", "error", "offline", "idle", "unknown"]);
const TEXT_LIMIT = 240;

export class ConfigError extends Error {}

export function defaultConfigPath() {
  return join(homedir(), ".config", "mosaic-dev", "config.json");
}

// Fail closed: the config must exist, parse, and name an absolute dataRoot.
export function loadConfig(path = defaultConfigPath()) {
  if (!existsSync(path)) throw new ConfigError(`config not found: ${path}`);
  let raw;
  try {
    raw = JSON.parse(readFileSync(path, "utf8"));
  } catch (err) {
    throw new ConfigError(`config is not valid JSON: ${path} (${err.message})`);
  }
  if (!raw || typeof raw !== "object" || Array.isArray(raw)) throw new ConfigError(`config is not an object: ${path}`);
  if (typeof raw.dataRoot !== "string" || !isAbsolute(raw.dataRoot)) {
    throw new ConfigError(`config.dataRoot must be an absolute path: ${path}`);
  }
  return { dataRoot: raw.dataRoot };
}

// Newest *.jsonl under a directory tree, by mtime. Returns null when none.
export function findNewestSession(dir) {
  if (!existsSync(dir)) return null;
  let best = null;
  const walk = (d) => {
    for (const name of readdirSync(d)) {
      const p = join(d, name);
      const st = statSync(p);
      if (st.isDirectory()) walk(p);
      else if (name.endsWith(".jsonl") && (!best || st.mtimeMs > best.mtimeMs)) best = { path: p, mtimeMs: st.mtimeMs };
    }
  };
  walk(dir);
  return best ? best.path : null;
}

function collapse(text) {
  const one = text.replace(/\s+/g, " ").trim();
  return one.length > TEXT_LIMIT ? one.slice(0, TEXT_LIMIT - 1) + "…" : one;
}

// Read a pi session log. A partially written final line is skipped and counted,
// not treated as fatal, because pi may be appending while we read.
export function readSession(file) {
  const lines = readFileSync(file, "utf8").split("\n");
  let sessionId = null, cwd = null, lastTimestamp = null, lastMessage = null, lastAssistantText = null, lastError = null;
  let skippedLines = 0;
  for (const line of lines) {
    if (!line.trim()) continue;
    let entry;
    try {
      entry = JSON.parse(line);
    } catch {
      skippedLines += 1;
      continue;
    }
    if (entry.timestamp) lastTimestamp = entry.timestamp;
    if (entry.type === "session") {
      sessionId = entry.id ?? sessionId;
      cwd = entry.cwd ?? cwd;
    } else if (entry.type === "message" && entry.message) {
      lastMessage = entry.message;
      if (entry.message.role === "assistant" && Array.isArray(entry.message.content)) {
        const text = entry.message.content.filter((c) => c && c.type === "text" && typeof c.text === "string").map((c) => c.text).join("\n");
        if (text.trim()) lastAssistantText = collapse(text);
        lastError = entry.message.stopReason === "error" && typeof entry.message.errorMessage === "string" ? collapse(entry.message.errorMessage) : null;
      }
    }
  }
  return { file, sessionId, cwd, lastTimestamp, lastMessage, lastAssistantText, lastError, skippedLines };
}

// True when an assistant message carries a tool call in its content.
export function hasToolCall(message) {
  return Array.isArray(message?.content) && message.content.some((c) => c && c.type === "toolCall");
}

// Pure state rule. alive: true/false, or null when liveness could not be checked.
// A null check is reported as "unknown" rather than assumed alive (fail closed).
// Acceptance rule (plan page, 2026-09-12): a seat mid-tool-call is working,
// never waiting. The newest entry being an assistant message with a tool call,
// or a tool result, means working even if the last text looked like a question.
// waiting needs a text-only assistant message whose turn ended (stopReason stop).
export function deriveState({ alive, session }) {
  if (alive === false) return "offline";
  if (alive !== true) return "unknown";
  if (!session || !session.lastMessage) return "idle";
  const m = session.lastMessage;
  if (m.role === "assistant") {
    if (m.stopReason === "error" || m.stopReason === "aborted" || m.stopReason === "length") return "error";
    if (hasToolCall(m)) return "working";
    if (m.stopReason === "stop") return "waiting";
    return "working";
  }
  return "working";
}

// Programs that count as a live pi agent in a tmux pane. A tmux session that
// still exists but only runs a shell (or another harness) is not alive: its
// pi session log is history, not status.
export const PI_COMMANDS = Object.freeze(["pi"]);

export function panesRunPi(listPanesOutput) {
  return String(listPanesOutput)
    .split("\n")
    .map((l) => l.trim())
    .some((cmd) => PI_COMMANDS.includes(cmd));
}

// true: a pane in the tmux session runs pi. false: no such session, or no pane
// runs pi. null: tmux could not be run at all (reported as "unknown", never
// assumed alive). `exec` is injectable for tests.
export function tmuxIsAlive({ socket, session }, { exec = spawnSync } = {}) {
  const args = [];
  if (socket) args.push("-L", socket);
  args.push("list-panes", "-s", "-t", `=${session}`, "-F", "#{pane_current_command}");
  const r = exec("tmux", args, { encoding: "utf8", timeout: 5000 });
  if (r.error) return null;
  if (r.status !== 0) return false;
  return panesRunPi(r.stdout ?? "");
}

// "Seen" marks: { "<project>/<agent>": "<lastActivity ISO>" }. A mark only
// applies while the agent's newest message still has that timestamp; anything
// the agent writes afterwards clears it automatically.
export function seenKey(rec) {
  return `${rec.project}/${rec.agent}`;
}

export function seenPath(boardDir) {
  return join(boardDir, "seen.json");
}

// Fail closed: a present but unreadable seen.json refuses the scan rather than
// silently dropping every mark.
export function loadSeen(boardDir) {
  const path = seenPath(boardDir);
  if (!existsSync(path)) return {};
  let parsed;
  try {
    parsed = JSON.parse(readFileSync(path, "utf8"));
  } catch (err) {
    throw new ConfigError(`seen marks file is not valid JSON: ${path} (${err.message})`);
  }
  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw new ConfigError(`seen marks file must be a JSON object: ${path}`);
  for (const [k, v] of Object.entries(parsed)) {
    if (typeof v !== "string") throw new ConfigError(`seen marks file has a non-string value for ${JSON.stringify(k)}: ${path}`);
  }
  return parsed;
}

export function saveSeen(boardDir, marks) {
  mkdirSync(boardDir, { recursive: true, mode: 0o700 });
  writeAtomic(seenPath(boardDir), marks);
}

// Set or clear one mark. Returns the updated map.
export function markSeen(boardDir, { project, agent, lastActivity, seen = true }) {
  for (const [name, v] of Object.entries({ project, agent, lastActivity })) {
    if (typeof v !== "string" || v.length === 0 || v.length > 512) throw new ConfigError(`${name} must be a non-empty string`);
  }
  if (project.includes("/")) throw new ConfigError("project must not contain '/'");
  if (typeof seen !== "boolean") throw new ConfigError("seen must be true or false");
  const marks = loadSeen(boardDir);
  const key = seenKey({ project, agent });
  if (seen) marks[key] = lastActivity;
  else delete marks[key];
  saveSeen(boardDir, marks);
  return marks;
}

// One agent -> one status record.
export function scanAgent(spec, { isAlive = tmuxIsAlive, now = () => new Date(), seen = {} } = {}) {
  const alive = isAlive(spec.tmux);
  const file = findNewestSession(spec.sessionsDir);
  const session = file ? readSession(file) : null;
  const state = deriveState({ alive, session });
  const scannedAt = now();
  const lastActivity = session?.lastTimestamp ?? null;
  const ageSeconds = lastActivity ? Math.max(0, Math.round((scannedAt.getTime() - Date.parse(lastActivity)) / 1000)) : null;
  const needsYou = state === "waiting" || state === "error";
  const isSeen = needsYou && lastActivity !== null && seen[seenKey(spec)] === lastActivity;
  return {
    agent: spec.agent,
    project: spec.project,
    state,
    waitingOnYou: needsYou && !isSeen,
    seen: isSeen,
    alive,
    tmux: spec.tmux,
    sessionFile: file,
    sessionId: session?.sessionId ?? null,
    cwd: session?.cwd ?? null,
    lastActivity,
    ageSeconds,
    lastAssistantText: session?.lastAssistantText ?? null,
    lastError: session?.lastError ?? null,
    skippedLines: session?.skippedLines ?? 0,
    scannedAt: scannedAt.toISOString(),
  };
}

// Repo agents: <repo>/.pi/state/<agent>/sessions, tmux default socket, session = agent.
export function discoverRepoAgents(repoRoot) {
  const stateDir = join(repoRoot, ".pi", "state");
  if (!existsSync(stateDir)) return [];
  const project = basename(resolve(repoRoot));
  return readdirSync(stateDir)
    .filter((n) => existsSync(join(stateDir, n, "sessions")))
    .sort()
    .map((agent) => ({ agent, project, sessionsDir: join(stateDir, agent, "sessions"), tmux: { socket: null, session: agent } }));
}

// Fleet agents: <fleet>/<agent>/.pi/agent/sessions, tmux socket mosaic-fleet, session = agent.
export function discoverFleetAgents(fleetRoot, { project = "fleet", socket = "mosaic-fleet" } = {}) {
  if (!existsSync(fleetRoot)) return [];
  return readdirSync(fleetRoot)
    .filter((n) => existsSync(join(fleetRoot, n, ".pi", "agent", "sessions")))
    .sort()
    .map((agent) => ({ agent, project, sessionsDir: join(fleetRoot, agent, ".pi", "agent", "sessions"), tmux: { socket, session: agent } }));
}

function writeAtomic(path, data) {
  const tmp = `${path}.tmp-${process.pid}`;
  writeFileSync(tmp, JSON.stringify(data, null, 2) + "\n", { mode: 0o600 });
  renameSync(tmp, path);
}

// Scan every spec and write <boardDir>/sessions/<project>/<agent>.json plus index.json.
export function scan(specs, { boardDir, isAlive, now } = {}) {
  if (!boardDir || !isAbsolute(boardDir)) throw new ConfigError("boardDir must be an absolute path");
  const seen = loadSeen(boardDir);
  const records = specs.map((spec) => scanAgent(spec, { isAlive, now, seen }));
  for (const rec of records) {
    const dir = join(boardDir, "sessions", rec.project);
    mkdirSync(dir, { recursive: true, mode: 0o700 });
    writeAtomic(join(dir, `${rec.agent}.json`), rec);
  }
  const generatedAt = (now ? now() : new Date()).toISOString();
  const index = {
    generatedAt,
    counts: Object.fromEntries(STATES.map((s) => [s, records.filter((r) => r.state === s).length])),
    waitingOnYou: records.filter((r) => r.waitingOnYou).map(seenKey),
    seen: records.filter((r) => r.seen).map(seenKey),
    sessions: records,
  };
  mkdirSync(boardDir, { recursive: true, mode: 0o700 });
  writeAtomic(join(boardDir, "index.json"), index);
  return index;
}