// 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 /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"; import { readRegistration, samePath, SeatError, LAYOUTS } from "../../seat/src/seat.mjs"; 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 firstUserText = 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 === "user" && firstUserText === null) { const text = userText(entry.message.content); if (text.trim()) firstUserText = collapse(text); } 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, firstUserText, skippedLines }; } // A user message's text: pi writes either a plain string or a list of blocks. function userText(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("\n"); } // Nearest ancestor (including dir itself) that holds a .git entry. A .git // file counts too, because git worktrees use one. Null when there is none. export function findRepoRoot(dir) { if (typeof dir !== "string" || !isAbsolute(dir)) return null; let cur = resolve(dir); for (;;) { if (existsSync(join(cur, ".git"))) return cur; const parent = resolve(cur, ".."); if (parent === cur) return null; cur = parent; } } // 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 parsePanes(listPanesOutput).some((p) => PI_COMMANDS.includes(p.command)); } // One line per pane: "\t" (the path column is optional). export function parsePanes(listPanesOutput) { return String(listPanesOutput) .split("\n") .filter((l) => l.trim()) .map((l) => { const [command = "", path = ""] = l.split("\t"); return { command: command.trim(), path: path.trim() || null }; }); } // Ask tmux about one session. Returns { alive, workspace }: // alive true when a pane runs pi; false when no such session or no pane // runs pi; null when tmux could not be run at all (reported as // "unknown", never assumed alive). // workspace the current path of the first pane running pi, else null. // `exec` is injectable for tests. export function tmuxInspect({ socket, session }, { exec = spawnSync } = {}) { const args = []; if (socket) args.push("-L", socket); args.push("list-panes", "-s", "-t", `=${session}`, "-F", "#{pane_current_command}\t#{pane_current_path}"); const r = exec("tmux", args, { encoding: "utf8", timeout: 5000 }); if (r.error) return { alive: null, workspace: null }; if (r.status !== 0) return { alive: false, workspace: null }; const pane = parsePanes(r.stdout ?? "").find((p) => PI_COMMANDS.includes(p.command)); return { alive: Boolean(pane), workspace: pane?.path ?? null }; } // Liveness only, for callers that do not need the pane path. export function tmuxIsAlive(tmux, opts) { return tmuxInspect(tmux, opts).alive; } // "Seen" marks: { "/": "" }. 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; } // `isAlive` may return a bare liveness value (true/false/null) or the richer // { alive, workspace } shape from tmuxInspect. Both are accepted. function liveness(result) { if (result && typeof result === "object") return { alive: result.alive ?? null, workspace: result.workspace ?? null }; return { alive: result ?? null, workspace: null }; } // Registrations written by `mosaic launch ` (packages/seat): one // record per seat under /seats///registration.json. Returns // the readable records plus one error line per unreadable one; a bad record // must not take the whole board down, but it is not silently dropped either. export function loadRegistrations(seatsDir) { const registrations = []; const errors = []; if (!seatsDir || !existsSync(seatsDir)) return { registrations, errors }; for (const layout of LAYOUTS) { const layoutDir = join(seatsDir, layout); if (!existsSync(layoutDir) || !statSync(layoutDir).isDirectory()) continue; for (const seat of readdirSync(layoutDir).sort()) { if (!statSync(join(layoutDir, seat)).isDirectory()) continue; try { const rec = readRegistration(seatsDir, seat, layout); if (rec) registrations.push(rec); } catch (err) { if (!(err instanceof SeatError)) throw err; errors.push(`seat ${layout}/${seat}: ${err.message}`); } } } return { registrations, errors }; } // A registration belongs to a row when it names the same sessions directory. // Seat names alone are not enough: the repo and fleet layouts both have a // "darkwing", and they are different seats. export function matchRegistration(spec, registrations) { return registrations.find((r) => r.sessionsDir && samePath(r.sessionsDir, spec.sessionsDir)) ?? null; } // One agent -> one status record. // // Three fields answer "what is this seat doing, and where" (Gate A ask, // 2026-09-12). Each is derived, never guessed; null means "unknown". A seat // started through `mosaic launch` has a registration, and a registered task, // project or workspace wins over the derived value; the *Source field says // which one the row shows. // task registration.task, else the session's first user message // (the log has no task envelope entry). // workspace registration.workspace, else the live pane path from tmux, // else the session log's cwd. // activeProject registration.project, else the basename of the nearest git // repo root above the workspace. export function scanAgent(spec, { isAlive = tmuxInspect, now = () => new Date(), seen = {}, registration = null } = {}) { const live = liveness(isAlive(spec.tmux)); const alive = live.alive; 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; const cwd = session?.cwd ?? null; const reg = registration && typeof registration === "object" ? registration : null; const derivedWorkspace = live.workspace ?? cwd; const workspace = reg?.workspace ?? derivedWorkspace; const workspaceSource = reg?.workspace ? "registration" : live.workspace ? "tmux-pane" : cwd ? "session-cwd" : null; const repoRoot = derivedWorkspace ? findRepoRoot(derivedWorkspace) : null; const derivedProject = repoRoot ? basename(repoRoot) : null; const activeProject = reg?.project ?? derivedProject; const activeProjectSource = reg?.project ? "registration" : derivedProject ? "workspace-git-root" : null; const firstUserText = session?.firstUserText ?? null; const task = reg?.task ? reg.task : firstUserText; const taskSource = reg?.task ? "registration" : firstUserText ? "first-user-message" : null; return { agent: spec.agent, project: spec.project, state, waitingOnYou: needsYou && !isSeen, seen: isSeen, alive, tmux: spec.tmux, sessionFile: file, sessionId: session?.sessionId ?? null, cwd, task, taskSource, workspace, workspaceSource, activeProject, activeProjectSource, registered: reg ? { startedAt: reg.startedAt, updatedAt: reg.updatedAt, harness: reg.harness, pid: reg.pid, tmux: reg.tmux, layout: reg.layout, launchScript: reg.launchScript } : null, lastActivity, ageSeconds, lastAssistantText: session?.lastAssistantText ?? null, lastError: session?.lastError ?? null, skippedLines: session?.skippedLines ?? 0, scannedAt: scannedAt.toISOString(), }; } // Repo agents: /.pi/state//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: //.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 /sessions//.json plus index.json. // seatsDir (optional): where `mosaic launch` registrations live; read only. export function scan(specs, { boardDir, isAlive, now, seatsDir = null } = {}) { if (!boardDir || !isAbsolute(boardDir)) throw new ConfigError("boardDir must be an absolute path"); if (seatsDir !== null && (typeof seatsDir !== "string" || !isAbsolute(seatsDir))) throw new ConfigError("seatsDir must be an absolute path or null"); const seen = loadSeen(boardDir); const { registrations, errors: registrationErrors } = loadRegistrations(seatsDir); const records = specs.map((spec) => scanAgent(spec, { isAlive, now, seen, registration: matchRegistration(spec, registrations) })); 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), registered: records.filter((r) => r.registered).map(seenKey), registrationErrors, sessions: records, }; mkdirSync(boardDir, { recursive: true, mode: 0o700 }); writeAtomic(join(boardDir, "index.json"), index); return index; }