Files
stack/packages/control-board/src/serve.mjs
T
jason.woltjeandClaude Opus 5.5 a5beb6d97d feat(conversation): CHAT-02 read-only Pi history reader and two board routes (#1507)
packages/conversation is a library with no server: safe-fs, the Pi session
parser, CHAT-01 pages, pinned snapshots, cursors and follow. The control
board adds GET /api/conversations and /api/conversation behind the Host
and Origin guard. Both are read-only, their queries are validated, and
each refusal code maps to a status.

Dewey authored it (packet 0cf177b1, revision 2). Filbert reviewed the code:
R1 revise (branch ids moving on append, the assumed-link bridge merging
branches, one unreadable seat directory turning the catalogue into a 500),
then R2 approve (3b14d66c). Darkwing reviewed the routes: R1 approve
(07b10ad1), R2 approve (b9d92003). The package lands with the routes,
because serve.mjs imports the reader at load.

On an index export: the eight suites 24/90/43/17/14/15/63/18,
conversation and control-board 153/153, webui 9/9.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-09-26 16:36:20 -05:00

323 lines
15 KiB
JavaScript

// Control board step 2: a tiny local web server. No dependencies, no auth.
// It only ever binds to a loopback address (fail closed otherwise).
//
// GET / the page (src/page.html)
// GET /api/board re-runs the scanner and returns index.json as JSON
// POST /api/seen {project, agent, lastActivity, seen?} marks a row as seen
// (or clears the mark with seen:false); rescans, returns index
// POST /api/reply {agent: "<project>/<agent>", text} delivers text to the
// seat's tmux pane through tools/tmux/agent-send.sh (#1505);
// answers {delivered, exitCode, stdout, stderr, ...}
// GET /healthz {"ok":true}
// GET /api/conversations
// read-only Pi conversation catalogue (#1507, CHAT-02)
// GET /api/conversation?id=<conversation>[&branch=<branch>][&cursor=<cursor>]
// one CHAT-01 history page: the first page of a branch (the
// default one without branch), or the next page (or a
// follow) for a cursor. A cursor call repeats the page's
// branch; without it the answer is 400. Refusals carry
// {error, refusal: {code, reconcile}}.
//
// POST requires Content-Type: application/json. A plain form post from another
// site in the browser cannot set that header without a CORS preflight, and this
// server answers no preflight, so a stray page cannot flip marks.
//
// Every route first checks Host and Origin (#1507). Binding to loopback does
// not stop DNS rebinding: a page whose name now resolves to 127.0.0.1 reaches
// this server as its own origin, with its own name as Host, and could read
// /api/board or post /api/reply into a live pane. A Host that is not a loopback
// name on this server's port, or any Origin other than this server's own, gets
// 403 before anything else runs. Same check as packages/webui/src/serve.mjs.
// No CORS headers are ever sent.
//
// Every /api/board request rescans, so the page is never staler than its
// refresh timer. The scan rewrites the derived board files as a side effect.
//
// The conversation routes read only: packages/conversation lists the repository
// specs' Pi session roots (never fleet or connector ones), opens files
// O_NOFOLLOW and writes nothing. Registrations are hints there too. The
// conversation id is opaque and no path comes from the request. Cursors live
// in this server's memory, bound to the one actor this unauthenticated
// loopback route has, local-operator.
import { createServer as createHttpServer } from "node:http";
import { readFileSync } from "node:fs";
import { join, resolve } from "node:path";
import { isIP } from "node:net";
import { hostname } from "node:os";
import { spawnSync } from "node:child_process";
import { scan, markSeen, seenKey, ConfigError, loadRegistrations } from "./scan.mjs";
import { createReader, rootsFromSpecs } from "../../conversation/src/reader.mjs";
const MAX_BODY = 4096;
// Reply-from-board (#1505). The board owns no transport: a reply is handed to
// the repository's own inter-agent channel, tools/tmux/agent-send.sh, which
// prepends the "[<sender> -> <host>:<session>]" preamble and pastes into the
// seat's pane. The tool's exit code is the receipt; the board never retries,
// queues or broadcasts, and never calls tmux send-keys itself.
export const DEFAULT_AGENT_SEND = resolve(import.meta.dirname, "..", "..", "..", "tools", "tmux", "agent-send.sh");
export const REPLY_LIMIT = 2000;
export const REPLY_SENDER = "control-board";
// Appended to every message on its own line. The board is a sender without
// a pane: it reads the seat's transcript, so a seat must answer in its own
// session as usual and never agent-send back to "control-board" (that
// target does not exist and the tool refuses it). Jason's refinement after
// the first real exchange, 2026-09-12.
export const REPLY_TRAILER = "(control-board: answer in your own session as usual; the board reads your transcript. Do not agent-send to control-board.)";
const REPLY_TIMEOUT_MS = 15000;
// Decide and, when allowed, send. Returns { status, body } for the HTTP layer.
// Refusals (4xx) happen before the tool runs and carry { error }. Once the
// tool has run the answer is 200 with delivered true/false, the exit code and
// both output streams verbatim, whatever the code was.
export function replyToRow({ index, key, text, agentSend = DEFAULT_AGENT_SEND, exec = spawnSync, now = () => new Date(), host = hostname().split(".")[0] }) {
if (typeof key !== "string" || !key) return { status: 400, body: { error: "agent must be the row id <project>/<agent>" } };
if (typeof text !== "string" || !text.trim()) return { status: 400, body: { error: "text must be a non-empty string" } };
if (text.length > REPLY_LIMIT) return { status: 400, body: { error: `text is longer than ${REPLY_LIMIT} characters` } };
const rec = index.sessions.find((r) => seenKey(r) === key);
if (!rec) return { status: 404, body: { error: `unknown row: ${key}` } };
if (rec.connector || / \(discord: [a-z0-9][a-z0-9._-]{0,63}\)$/.test(rec.agent)) return { status: 409, body: { error: "board replies are disabled for Discord connectors" } };
const reg = rec.registered;
if (!reg) return { status: 409, body: { error: "reply needs a registered seat (start it through scripts/mosaic launch)" } };
if (reg.alive === false) return { status: 409, body: { error: `registration is stale: pid ${reg.pid} is gone` } };
if (!reg.tmux || !reg.tmux.session) return { status: 409, body: { error: "registration has no tmux session to address" } };
const session = reg.tmux.session;
const socket = reg.tmux.socket || null;
const args = ["-s", session, "-S", `${host}:${REPLY_SENDER}`];
if (socket) args.push("-L", socket);
args.push("-m", `${text}\n${REPLY_TRAILER}`);
// The registration says which socket the seat is on. A launcher-exported
// MOSAIC_TMUX_SOCKET in this server's own environment must not override it.
const env = { ...process.env };
delete env.MOSAIC_TMUX_SOCKET;
const r = exec(agentSend, args, { encoding: "utf8", timeout: REPLY_TIMEOUT_MS, env });
if (r.error) return { status: 500, body: { error: `could not run ${agentSend}: ${r.error.message}` } };
const exitCode = r.status;
return {
status: 200,
body: {
delivered: exitCode === 0,
exitCode,
signal: r.signal ?? null,
stdout: String(r.stdout ?? ""),
stderr: String(r.stderr ?? ""),
agent: key,
session,
socket,
sentAt: now().toISOString(),
},
};
}
function sendJson(res, status, body) {
res.writeHead(status, { "content-type": "application/json", "cache-control": "no-store" });
res.end(JSON.stringify(body) + "\n");
}
function readJsonBody(req) {
return new Promise((resolvePromise, reject) => {
const type = String(req.headers["content-type"] || "").split(";")[0].trim().toLowerCase();
if (type !== "application/json") return reject(new Error("Content-Type must be application/json"));
const chunks = [];
let size = 0;
req.on("data", (c) => {
size += c.length;
if (size > MAX_BODY) {
req.destroy();
reject(new Error(`body larger than ${MAX_BODY} bytes`));
return;
}
chunks.push(c);
});
req.on("end", () => {
try {
const parsed = JSON.parse(Buffer.concat(chunks).toString("utf8"));
if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw new Error("body must be a JSON object");
resolvePromise(parsed);
} catch (err) {
reject(new Error(`invalid JSON body: ${err.message}`));
}
});
req.on("error", reject);
});
}
const LOOPBACK = new Set(["127.0.0.1", "::1", "localhost"]);
export function isLoopbackHost(host) {
if (LOOPBACK.has(host)) return true;
return isIP(host) === 4 && host.startsWith("127.");
}
// Returns the refusal text for a request that did not come from this server's
// own loopback origin, or null. Uses the port the connection arrived on.
export function foreignRequest(req) {
let authority;
try {
authority = new URL(`http://${req.headers.host}`);
} catch {
return "non-local Host refused";
}
const plain = !authority.username && !authority.password && authority.pathname === "/" && !authority.search && !authority.hash;
if (!plain || !isLoopbackHost(authority.hostname.replace(/^\[|\]$/g, "")) || Number(authority.port || 80) !== req.socket.localPort) return "non-local Host refused";
if (req.headers.origin !== undefined && req.headers.origin !== `http://${req.headers.host}`) return "cross-origin request refused";
return null;
}
// HTTP status for each reader refusal. The body always carries the code. The
// tests hold every code the reader can raise to an entry here.
export const REFUSAL_STATUS = {
"unknown-conversation": 404,
"unknown-branch": 404,
unavailable: 404,
"cursor-unknown": 409,
"cursor-expired": 409,
"cursor-foreign": 409,
"source-replaced": 409,
"incomplete-header": 409,
"unsafe-path": 403,
"foreign-project": 403,
unreadable: 403,
"unsupported-harness": 422,
"not-a-pi-session": 422,
"too-large": 422,
"unknown-actor": 403,
"unsupported-purpose": 422,
};
const QUERY_VALUE = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
function sendConversationJson(res, status, body) {
res.writeHead(status, { "content-type": "application/json", "cache-control": "no-store", "x-content-type-options": "nosniff" });
res.end(JSON.stringify(body) + "\n");
}
// GET /api/conversation: id is required; branch and cursor are optional; any
// other, repeated or malformed parameter is a 400.
function conversationQuery(url) {
const out = {};
for (const key of new Set(url.searchParams.keys())) {
const values = url.searchParams.getAll(key);
if (!["id", "branch", "cursor"].includes(key)) return { error: `unknown parameter: ${key}` };
if (values.length !== 1 || !QUERY_VALUE.test(values[0])) return { error: `invalid ${key}` };
out[key] = values[0];
}
if (!out.id) return { error: "id is required" };
if (out.cursor && !out.branch) return { error: "a cursor call repeats the page's branch" };
return out;
}
export function conversationResponse(reader, url) {
if (url.pathname === "/api/conversations") {
if ([...url.searchParams.keys()].length) return { status: 400, body: { error: "no parameters are accepted" } };
return { status: 200, body: reader.catalogue() };
}
const query = conversationQuery(url);
if (query.error) return { status: 400, body: { error: query.error } };
const out = query.cursor
? reader.next({ cursor: query.cursor, conversation: query.id, branch: query.branch })
: reader.open({ conversation: query.id, branch: query.branch ?? null });
if (out.ok) return { status: 200, body: out };
return { status: REFUSAL_STATUS[out.refusal.code] ?? 422, body: { error: out.refusal.message, refusal: { code: out.refusal.code, reconcile: out.refusal.reconcile } } };
}
export function loadPage(path = join(import.meta.dirname, "page.html")) {
return readFileSync(path, "utf8");
}
// specs: agent specs to scan on each request. boardDir: where scan writes.
export function createServer({ specs, boardDir, isAlive, now, seatsDir = null, discordDataRoot = null, page = loadPage(), isPidAlive, agentSend = DEFAULT_AGENT_SEND, exec = spawnSync, conversationReader = null }) {
const rescan = () => scan(specs, { boardDir, isAlive, now, seatsDir, isPidAlive, discordDataRoot });
const reader = conversationReader ?? createReader({ roots: () => rootsFromSpecs(specs, loadRegistrations(seatsDir).registrations) });
return createHttpServer((req, res) => {
const refused = foreignRequest(req);
if (refused) {
req.resume();
return sendJson(res, 403, { error: refused });
}
const url = new URL(req.url, "http://localhost");
if (req.method === "POST" && url.pathname === "/api/reply") {
return readJsonBody(req)
.then((body) => {
let index;
try {
index = rescan();
} catch (err) {
return sendJson(res, 500, { error: err.message });
}
const out = replyToRow({ index, key: body.agent, text: body.text, agentSend, exec, now });
sendJson(res, out.status, out.body);
})
.catch((err) => sendJson(res, 400, { error: err.message }));
}
if (req.method === "POST" && url.pathname === "/api/seen") {
return readJsonBody(req)
.then((body) => {
try {
markSeen(boardDir, { project: body.project, agent: body.agent, lastActivity: body.lastActivity, seen: body.seen ?? true });
} catch (err) {
return sendJson(res, 400, { error: err.message });
}
try {
sendJson(res, 200, rescan());
} catch (err) {
sendJson(res, 500, { error: err.message });
}
})
.catch((err) => sendJson(res, 400, { error: err.message }));
}
if (req.method !== "GET" && req.method !== "HEAD") {
res.writeHead(405, { "content-type": "text/plain" });
return res.end("method not allowed\n");
}
if (url.pathname === "/" || url.pathname === "/index.html") {
res.writeHead(200, { "content-type": "text/html; charset=utf-8", "cache-control": "no-store" });
return res.end(page);
}
if (url.pathname === "/api/board") {
let index;
try {
index = rescan();
} catch (err) {
res.writeHead(500, { "content-type": "application/json", "cache-control": "no-store" });
return res.end(JSON.stringify({ error: err.message }) + "\n");
}
res.writeHead(200, { "content-type": "application/json", "cache-control": "no-store" });
return res.end(JSON.stringify(index) + "\n");
}
if (url.pathname === "/api/conversations" || url.pathname === "/api/conversation") {
let out;
try {
out = conversationResponse(reader, url);
} catch (err) {
process.stderr.write(`conversation read failed: ${err.message}\n`);
out = { status: 500, body: { error: "conversation read failed" } };
}
return sendConversationJson(res, out.status, out.body);
}
if (url.pathname === "/favicon.ico") {
res.writeHead(204);
return res.end();
}
if (url.pathname === "/healthz") {
res.writeHead(200, { "content-type": "application/json" });
return res.end('{"ok":true}\n');
}
res.writeHead(404, { "content-type": "text/plain" });
res.end("not found\n");
});
}
// Resolves to the listening server. Refuses any non-loopback host.
export async function startServer({ host = "127.0.0.1", port = 7331, ...rest }) {
if (!isLoopbackHost(host)) throw new ConfigError(`refusing to bind to non-loopback host: ${host} (no auth in the MVP)`);
const server = createServer(rest);
return new Promise((resolvePromise, reject) => {
server.once("error", reject);
server.listen(port, host, () => {
server.off("error", reject);
resolvePromise(server);
});
});
}