Add control board status scanner and MVP plan (#1503)

Step 1 of the control board MVP (decision MOSAIC-STACK-D-001): a plan page, Gitea #1503, and packages/control-board, which reads each agent's newest pi session log plus tmux liveness and writes one status file per agent under <dataRoot>/board/. 23/23 tests; independent review approved after three fixes (length stopReason as error, unknown liveness state, secrets-boundary test). CURRENT.md now points at step 2, the page. Co-Authored-By: Claude Fable 5.1 <[email protected]>
2026-09-12 07:26:03 -05:00
parent 1993039c76
commit b9f59a5903
10 changed files with 1003 additions and 15 deletions
@@ -0,0 +1,53 @@
+#!/usr/bin/env node
+// Usage: node packages/control-board/src/cli.mjs scan [--config PATH] [--repo PATH] [--fleet PATH|none] [--liveness tmux|assume-alive] [--print]
+// Exit 0 on success, 2 on a config refusal or bad usage.
+import { join } from "node:path";
+import { homedir } from "node:os";
+import { loadConfig, defaultConfigPath, discoverRepoAgents, discoverFleetAgents, scan, tmuxIsAlive, ConfigError } from "./scan.mjs";
+
+function parseArgs(argv) {
+  const opts = { command: argv[0], config: defaultConfigPath(), repo: process.cwd(), fleet: join(homedir(), ".mosaic", "fleet", "agents"), liveness: "tmux", print: false };
+  for (let i = 1; i < argv.length; i++) {
+    const a = argv[i];
+    const next = () => {
+      if (i + 1 >= argv.length) throw new ConfigError(`missing value for ${a}`);
+      return argv[++i];
+    };
+    if (a === "--config") opts.config = next();
+    else if (a === "--repo") opts.repo = next();
+    else if (a === "--fleet") opts.fleet = next();
+    else if (a === "--liveness") opts.liveness = next();
+    else if (a === "--print") opts.print = true;
+    else throw new ConfigError(`unknown argument: ${a}`);
+  }
+  if (opts.command !== "scan") throw new ConfigError("usage: mosaic-board scan [--config PATH] [--repo PATH] [--fleet PATH|none] [--liveness tmux|assume-alive] [--print]");
+  if (!["tmux", "assume-alive"].includes(opts.liveness)) throw new ConfigError(`unknown liveness mode: ${opts.liveness}`);
+  return opts;
+}
+
+function main() {
+  const opts = parseArgs(process.argv.slice(2));
+  const { dataRoot } = loadConfig(opts.config);
+  const specs = [...discoverRepoAgents(opts.repo), ...(opts.fleet === "none" ? [] : discoverFleetAgents(opts.fleet))];
+  const isAlive = opts.liveness === "tmux" ? tmuxIsAlive : () => true;
+  const boardDir = join(dataRoot, "board");
+  const index = scan(specs, { boardDir, isAlive });
+  if (opts.print) {
+    for (const s of index.sessions) {
+      const flag = s.waitingOnYou ? "*" : " ";
+      const age = s.ageSeconds == null ? "-" : `${Math.round(s.ageSeconds / 60)}m`;
+      process.stdout.write(`${flag} ${s.state.padEnd(8)} ${s.project.padEnd(14)} ${s.agent.padEnd(16)} ${age.padStart(7)}  ${s.lastAssistantText ? s.lastAssistantText.slice(0, 80) : ""}\n`);
+    }
+  }
+  process.stdout.write(`board: ${boardDir} (${index.sessions.length} sessions, ${index.waitingOnYou.length} waiting on you)\n`);
+}
+
+try {
+  main();
+} catch (err) {
+  if (err instanceof ConfigError) {
+    process.stderr.write(`refused: ${err.message}\n`);
+    process.exit(2);
+  }
+  throw err;
+}
@@ -0,0 +1,193 @@
+// 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 live tmux session for this agent
+//   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.
+
+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 };
+}
+
+// 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).
+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 === "stop") return "waiting";
+    if (m.stopReason === "error" || m.stopReason === "aborted" || m.stopReason === "length") return "error";
+    return "working";
+  }
+  return "working";
+}
+
+export function tmuxIsAlive({ socket, session }) {
+  const args = [];
+  if (socket) args.push("-L", socket);
+  args.push("has-session", "-t", `=${session}`);
+  const r = spawnSync("tmux", args, { encoding: "utf8", timeout: 5000 });
+  if (r.error) return null;
+  return r.status === 0;
+}
+
+// One agent -> one status record.
+export function scanAgent(spec, { isAlive = tmuxIsAlive, now = () => new Date() } = {}) {
+  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;
+  return {
+    agent: spec.agent,
+    project: spec.project,
+    state,
+    waitingOnYou: state === "waiting" || state === "error",
+    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 records = specs.map((spec) => scanAgent(spec, { isAlive, now }));
+  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((r) => `${r.project}/${r.agent}`),
+    sessions: records,
+  };
+  mkdirSync(boardDir, { recursive: true, mode: 0o700 });
+  writeAtomic(join(boardDir, "index.json"), index);
+  return index;
+}