import crypto from 'node:crypto'; import fs from 'node:fs'; import path from 'node:path'; import { DATA_DIR } from './config.js'; export const OPERATIONS_FILE = path.join(DATA_DIR, 'operations.jsonl'); const MUTATING = new Set(['POST', 'PUT', 'PATCH', 'DELETE']); // Reads worth auditing. A GET is normally none of this log's business — it // changes nothing — but `wait` is the one read that IS an event between two // agents: A blocked on B until B stopped working. Without it the log records // work being handed out and nothing ever coming back, which is exactly half of // "who called whom". Deliberately NOT here: `tail`, which every open pane polls // constantly and which says nothing a resolved wait does not already say. const LOGGED_READS = [/^\/api\/agents\/[^/]+\/wait$/]; const shouldLog = (req) => MUTATING.has(req.method) || (req.method === 'GET' && LOGGED_READS.some((re) => re.test(req.path))); // The body is stored WHOLE, on the operator's instruction: "just store all the // full api calls. why this arbitrary compression." So there is no allowlist of // routes, no size cap, and nothing is replaced by a summary of itself. The one // thing still withheld is a credential — that is not compression, it is not // writing secrets into a file that lives on the bucket. // // Above this length a string is stored as {present, chars, sha256, text} rather // than as a bare string. Nothing is lost either way: this only decides whether // the checksum travels beside the value, and `chars` is what the log's compact // list column reads. const MAX_TEXT = 500; // A backstop against a cyclic object, not a limit on how much is kept: a request // body is the output of a JSON or text parser and cannot contain a cycle, but // JSON.stringify throwing here would lose the whole entry. const MAX_DEPTH = 20; // How far back a read will go looking for complete records. One enormous entry // must not hide the log, and reading a whole year of it must not exhaust memory. const MAX_TAIL = 256 * 1024 * 1024; const SENSITIVE_KEY = /(authorization|credential|password|secret|subscription|token|endpoint|private.?key)/i; const CONTENT_KEY = /(body|content|data|prompt|text)/i; const digest = (value) => crypto.createHash('sha256').update(value).digest('hex'); // The value, plus the two things worth having beside it: sha256, because equal // checksums are how a repeated prompt or a scheduled job shows up, and chars, // because that is what the list column reads without touching the text. function textSummary(value) { return { present: value.length > 0, chars: value.length, sha256: digest(value), text: value }; } /** * The call as it was made, whole, with a checksum attached to anything long * enough to want one. Credentials are the single exception and are replaced by * `[redacted]` wherever they appear. */ export function summarizePayload(value, key = '', depth = 0) { if (value == null || typeof value === 'boolean' || typeof value === 'number') return value; // No route here parses a raw body today, but if one ever does, keep the bytes // rather than a description of them. if (Buffer.isBuffer(value)) { return { bytes: value.length, sha256: digest(value), base64: value.toString('base64') }; } if (typeof value === 'string') { if (SENSITIVE_KEY.test(key)) return '[redacted]'; if (CONTENT_KEY.test(key) || value.length > MAX_TEXT) return textSummary(value); return value; } if (depth >= MAX_DEPTH) return '[max-depth]'; if (Array.isArray(value)) return value.map((v) => summarizePayload(v, key, depth + 1)); if (typeof value === 'object') { const out = {}; for (const [k, v] of Object.entries(value)) { out[k] = SENSITIVE_KEY.test(k) ? '[redacted]' : summarizePayload(v, k, depth + 1); } return out; } return String(value); } function append(record) { try { fs.mkdirSync(path.dirname(OPERATIONS_FILE), { recursive: true }); fs.appendFileSync(OPERATIONS_FILE, `${JSON.stringify(record)}\n`, { mode: 0o600 }); } catch (e) { // Auditing must never turn a successfully completed user operation into an // HTTP failure. Make storage trouble loud in the server log instead. console.error('[operations.append]', e && e.message); } } const requestOrigin = (req) => String( req.query?.from || req.headers?.['x-am-origin'] || (req.body && !Array.isArray(req.body) && typeof req.body === 'object' ? req.body.from : '') || '', ).trim(); const cleanQuery = (query) => { const out = { ...(query || {}) }; delete out.from; return summarizePayload(out, 'query'); }; /** * Require an attributable origin for every state-changing API request and * append its outcome to a durable JSONL log. * * resolveOrigin(raw, req) returns {id,type,name?,cli?}, or null when the id is * unknown. It may derive an identity from a protocol route (remote agents do * this for backwards compatibility with already-running polling loops). */ export function operationMiddleware({ resolveOrigin, resolveTarget, allowMissing = false } = {}) { return (req, res, next) => { if (!req.path.startsWith('/api/') || !shouldLog(req)) return next(); const raw = requestOrigin(req); let origin = resolveOrigin ? resolveOrigin(raw, req) : (raw ? { id: raw, type: 'unknown' } : null); if (!origin && allowMissing) origin = { id: 'test', type: 'test' }; // A logged READ is never refused for want of an origin. `wait` is documented // as read-only and every watch loop running right now calls it without // `?from=`; rejecting those would break them the moment this ships. An // unattributed wait still records that someone finished waiting on B. if (!origin && MUTATING.has(req.method)) { return res.status(400).json({ error: raw ? `unknown origin '${raw}'` : 'from required — mutating calls must pass ?from= (agents use $AM_ID)', }); } if (origin) req.operationOrigin = origin; // BEFORE next(), not at response time: the handler for a delete removes the // session from the store and only then answers, so resolving this later // recorded `{id}` for a session whose name and cli had just been thrown // away — the one operation where the roster can never fill them back in. // A request-time snapshot also gives a rename the name it had when the call // arrived, which is the state the entry is describing. const target = resolveTarget ? resolveTarget(req) : null; const started = Date.now(); const operationId = crypto.randomUUID(); let responseBody; let recorded = false; const originalJson = res.json.bind(res); res.json = (body) => { responseBody = body; return originalJson(body); }; const record = () => { if (recorded) return; recorded = true; // A wait is a polling loop: only the call that RESOLVED is an event. The // ones that timed out say "still working", which the log already implies, // and logging them would multiply the entries by however long the job ran. // Same for a wait the caller abandoned (no body) or one whose target had // already gone: nothing came back, so there is nothing to draw. if (!MUTATING.has(req.method) && !(responseBody && responseBody.matched === true)) return; append({ version: 1, id: operationId, at: new Date(started).toISOString(), origin, // Who it was done TO, snapshotted above. The id is in the path already, // but a name read back later is the name the session has NOW — renamed // or deleted, and the audit trail stops making sense. ...(target ? { target } : {}), method: req.method, path: req.path, query: cleanQuery(req.query), request: summarizePayload(req.body, 'body'), status: res.statusCode, ok: res.statusCode < 400, durationMs: Date.now() - started, result: summarizePayload(responseBody, 'result'), }); }; res.once('finish', record); res.once('close', record); next(); }; } export function readOperations(limit = 200, before = null) { const take = Math.max(1, Math.min(1000, Number(limit) || 200)); let fd; try { fd = fs.openSync(OPERATIONS_FILE, 'r'); const size = fs.fstatSync(fd).size; // A bounded tail keeps this endpoint cheap after years of operations. But an // entry is now as big as the call it records — a file write can be eight // megabytes on one line — so a fixed window is not enough: it could contain // no COMPLETE line at all and the log would read as empty. Grow it until // there are enough whole records, or until the ceiling says stop. let rows = []; for (let window = 4 * 1024 * 1024; ; window *= 4) { const start = Math.max(0, size - window); const buf = Buffer.alloc(size - start); fs.readSync(fd, buf, 0, buf.length, start); let text = buf.toString('utf8'); // the first line is almost certainly cut in half by the window if (start > 0) text = text.slice(Math.max(0, text.indexOf('\n') + 1)); rows = text.split('\n').filter(Boolean).flatMap((line) => { try { return [JSON.parse(line)]; } catch { return []; } }); if (rows.length >= take || start === 0 || window >= MAX_TAIL) break; } // Sort, do not merely reverse. A record is appended when its response // finishes, but `at` is when the request STARTED — and a `wait` can block // for five minutes, so it lands in the file after calls that began later and // finished sooner. Reversing append order therefore returned rows out of // chronological order, which put an old wait above newer calls in the log // and, because the map derives its x from rank, could run its time axis // backwards. Ties keep newest-appended first, which is what reversing did. return rows .filter((row) => !before || row.at < before) .reverse() .sort((a, b) => (a.at < b.at ? 1 : a.at > b.at ? -1 : 0)) .slice(0, take); } catch { return []; } finally { if (fd !== undefined) try { fs.closeSync(fd); } catch {} } }