fleet update

This commit is contained in:
thePR0M3TH3AN committed 2026-10-26 14:58:19 -04:00
1 parent 0ffd92a8d2
commit 2d695e7acd
20 files changed
+546 -38

No files matched your search

+87 -12
View File
@@ -12,8 +12,10 @@ export const HEALTH_DEFAULTS = {
diskFreeGb: 5,
memoryAvailablePct: 15,
swapUsedPct: 50,
// Occupancy alone is not tight: Linux keeps swapped pages. swapUsedPct counts
// only when available memory is also low. swapFullPct counts on its own.
// Occupancy alone is not tight, and a full swap file is not tight while
// plenty of RAM is free. swapUsedPct counts only when available memory is
// under twice memoryAvailablePct. swapFullPct is kept so older configs load;
// it is not a hold by itself. Live swap-in (pswpin) is pressure on its own.
swapFullPct: 90,
cpuTempC: 90,
batteryPct: 20,
@@ -189,6 +191,35 @@ export function parseDf(text) {
return { mount: parts.slice(5).join(' '), totalBytes: total, usedBytes: used, freeBytes: free, freePct };
}
/** Linux /proc/vmstat counters. pswpin is pages read back from swap. */
export function parseLinuxVmstat(text) {
const out = {};
for (const line of String(text || '').split('\n')) {
const m = /^([A-Za-z0-9_]+)\s+(\d+)$/.exec(line.trim());
if (m) out[m[1]] = Number(m[2]);
}
return out;
}
/** Pages swapped in between two /proc/vmstat snapshots. Null when either side lacks pswpin. */
export function swapInPages(before, after) {
if (before?.pswpin == null || after?.pswpin == null) return null;
return Math.max(0, after.pswpin - before.pswpin);
}
/** Top non-trivial CPU lines from `ps -ax -o pid=,pcpu=,comm=`. */
export function parseBusyPs(text) {
const rows = [];
for (const line of String(text || '').split('\n')) {
const m = /^\s*(\d+)\s+(\d+(?:\.\d+)?)\s+(\S+)/.exec(line);
if (!m) continue;
const cpu = Number(m[2]);
if (cpu < 5) continue;
rows.push({ pid: Number(m[1]), cpu, comm: m[3] });
}
return rows.sort((a, b) => b.cpu - a.cpu).slice(0, 3);
}
export function parseVmStat(text) {
const page = /page size of (\d+) bytes/i.exec(text || '');
const pages = {};
@@ -521,20 +552,24 @@ function readWindowsBattery() {
function linuxSample(sysRoot, opt) {
if (opt.statTexts || opt.procSamples) {
const vm = Array.isArray(opt.vmstat) ? opt.vmstat.map((text) => parseLinuxVmstat(text)) : null;
return {
usagePct: opt.statTexts ? cpuUsagePct(parseCpuStat(opt.statTexts[0]), parseCpuStat(opt.statTexts[1])) : null,
before: opt.procSamples ? opt.procSamples[0] : [],
after: opt.procSamples ? opt.procSamples[1] : [],
elapsedMs: opt.elapsedMs ?? 1000,
swapInPages: vm ? swapInPages(vm[0], vm[1]) : null,
};
}
if (opt.sampleMs === 0) return { usagePct: null, before: [], after: [], elapsedMs: 0 };
if (opt.sampleMs === 0) return { usagePct: null, before: [], after: [], elapsedMs: 0, swapInPages: null };
const stat1 = parseCpuStat(readText(path.join(sysRoot, 'proc/stat')) || '');
const vm1 = parseLinuxVmstat(readText(path.join(sysRoot, 'proc/vmstat')) || '');
const pages = pageSize();
const before = readProcSnapshot(sysRoot, pages);
const t0 = Date.now();
pause(opt.sampleMs ?? 200);
const stat2 = parseCpuStat(readText(path.join(sysRoot, 'proc/stat')) || '');
const vm2 = parseLinuxVmstat(readText(path.join(sysRoot, 'proc/vmstat')) || '');
const after = readProcSnapshot(sysRoot, pages);
let usagePct = cpuUsagePct(stat1, stat2);
if (usagePct == null) {
@@ -542,7 +577,7 @@ function linuxSample(sysRoot, opt) {
pause(opt.sampleMs ?? 200);
usagePct = cpuUsagePct(a, cpuTimes());
}
return { usagePct, before, after, elapsedMs: Math.max(1, Date.now() - t0) };
return { usagePct, before, after, elapsedMs: Math.max(1, Date.now() - t0), swapInPages: swapInPages(vm1, vm2) };
}
function readLoad(platform, sysRoot) {
@@ -606,13 +641,14 @@ function judge(snap, limits) {
if (mem?.swapTotalBytes > 0 && mem.swapUsedBytes != null) {
const pct = (100 * mem.swapUsedBytes) / mem.swapTotalBytes;
const availablePct = mem.totalBytes > 0 && mem.availableBytes != null ? (100 * mem.availableBytes) / mem.totalBytes : null;
const memoryLow = availablePct != null && availablePct < limits.memoryAvailablePct;
const nearFull = pct >= limits.swapFullPct;
// Occupancy alone is normal on Linux. Count it when memory is also low, or when swap is near full.
if (nearFull || (memoryLow && pct >= limits.swapUsedPct)) {
const words = nearFull
? `swap ${Math.round(pct)}% used, near full (at or over ${limits.swapFullPct}%)`
: `swap ${Math.round(pct)}% used while available memory is under ${limits.memoryAvailablePct}%`;
const softLimit = limits.memoryAvailablePct * 2;
const memorySoft = availablePct != null && availablePct < softLimit;
const swappingIn = Number(mem.swapInPages) > 0;
// A full swap file with plenty of free RAM is parked pages, not pressure.
if (swappingIn || (memorySoft && pct >= limits.swapUsedPct)) {
const words = swappingIn
? `swap is being read in (${mem.swapInPages} pages in the sample)`
: `swap ${Math.round(pct)}% used while available memory is under ${softLimit}%`;
crossed.push({ id: 'swap', words });
}
}
@@ -649,12 +685,45 @@ function judge(snap, limits) {
if (snap.cpu?.load1 != null && snap.cpu.cores) {
const per = snap.cpu.load1 / snap.cpu.cores;
if (per >= limits.loadPerCore) {
crossed.push({ id: 'load', words: `load ${snap.cpu.load1} on ${snap.cpu.cores} cores (${Math.round(per * 10) / 10} per core), at or over ${limits.loadPerCore}` });
const fleet = fleetCpuPct(snap.areas);
const words = `load ${snap.cpu.load1} on ${snap.cpu.cores} cores (${Math.round(per * 10) / 10} per core)`;
if (fleet != null && fleet >= 10) {
const who = (snap.areas || []).filter((a) => a.cpuPct >= 10).map((a) => `${a.id} ${a.cpuPct}%`).join(', ');
crossed.push({ id: 'load', words: `${words}, and fleet areas are using CPU${who ? ` (${who})` : ''}` });
} else {
const outside = describeBusy(snap.busy);
const sample = snap.cpu.usagePct == null ? '' : ` A short CPU sample is ${snap.cpu.usagePct}%.`;
const fleetWords = fleet == null ? 'fleet area CPU was not measured' : 'fleet areas are idle';
snap.notes.push({ id: 'load', words: `${words} is above ${limits.loadPerCore} per core, but ${fleetWords}.${sample}${outside} This does not hold fleet work.` });
}
}
}
return crossed;
}
function fleetCpuPct(areas) {
const known = (areas || []).filter((a) => a.found && a.cpuPct != null);
if (!known.length) return null;
return known.reduce((sum, a) => sum + a.cpuPct, 0);
}
function describeBusy(busy) {
const rows = (busy || []).filter((row) => row?.comm && Number(row.cpu) >= 1).slice(0, 3);
if (!rows.length) return '';
return ` Busy outside the fleet: ${rows.map((row) => `${row.comm} ${row.cpu}%`).join(', ')}.`;
}
function collectBusy(snap, limits, opt) {
if (snap.cpu?.load1 == null || !snap.cpu.cores) return [];
if (snap.cpu.load1 / snap.cpu.cores < limits.loadPerCore) return [];
const fleet = fleetCpuPct(snap.areas);
if (fleet != null && fleet >= 10) return [];
if (Array.isArray(opt.busy)) return opt.busy;
if (opt.psBusy == null && (opt.sampleMs === 0 || opt.skipBusy)) return [];
const text = opt.psBusy != null ? opt.psBusy : sh('ps', ['-ax', '-o', 'pid=,pcpu=,comm='], 2000);
return parseBusyPs(text);
}
function verdictOf(crossed, battery) {
const onBattery = battery?.present && battery.state === 'discharging';
if (!crossed.length && onBattery) {
@@ -690,6 +759,7 @@ export function formatHealth(snap) {
const left = battery.minutesRemaining == null ? 'time unavailable' : `${battery.minutesRemaining} min left`;
lines.push(`battery ${battery.percent ?? '?'}% ${battery.state} ${left}`);
}
for (const note of snap.notes || []) lines.push(`note ${note.words}`);
if (!snap.areas?.length) lines.push('areas none');
for (const area of snap.areas || []) {
if (!area.found) lines.push(`area ${area.id} window not found`);
@@ -742,6 +812,9 @@ export function fleetHealth(c = {}, opt = {}) {
areas = readPsAreas(panes, opt);
}
}
if (memory?.status === 'ok') {
memory.swapInPages = opt.swapInPages != null ? opt.swapInPages : (sampled?.swapInPages ?? null);
}
const snap = {
at: opt.at || new Date().toISOString(),
platform,
@@ -753,7 +826,9 @@ export function fleetHealth(c = {}, opt = {}) {
battery,
areas,
thresholds: limits,
notes: [],
};
snap.busy = collectBusy(snap, limits, opt);
snap.crossed = judge(snap, limits);
snap.verdict = verdictOf(snap.crossed, snap.battery);
return snap;
+26 -3
View File
@@ -3,9 +3,10 @@
import fs from 'node:fs';
import path from 'node:path';
import { spawn } from 'node:child_process';
import { KIT, area as getArea, die, pidAlive, tryGit, git, mainRef, writeAtomic, today, landerIsUp, matchesAny, startLander, runTimedSync } from '../lib.mjs';
import { KIT, area as getArea, die, pidAlive, tryGit, git, mainRef, writeAtomic, today, landerIsUp, matchesAny, startLander, runTimedSync, readRun } from '../lib.mjs';
import { send as mailboxSendFile, read as mailboxReadDir, archive as mailboxArchiveDir } from '../../adapters/mailbox.mjs';
import { appendWakeLog, priorWakes } from './wake-log.mjs';
import { readHoldMap, readProgressMap, STALL_MIN } from './progress.mjs';
import { tmuxCommand, tmuxSocketFacts } from './tmux.mjs';
const STATUSES = ['open', 'in-progress', 'blocked', 'done', 'deferred'];
@@ -549,17 +550,39 @@ function senderOf(file) {
* That is the only recovery. It runs when someone looks at status, reads mail,
* hears, or tells. A filename already nudged is not nudged again. There is no timer.
*/
export function fleetStatus(c, { recoverAfterMs = RECOVER_AFTER_MS, idleAfterMs = IDLE_AFTER_MS } = {}) {
export function fleetStatus(c, { recoverAfterMs = RECOVER_AFTER_MS, idleAfterMs = IDLE_AFTER_MS, views = null } = {}) {
const state = readNudgeState(c);
let stateDirty = false;
const recovered = [];
const configured = Number(c.progress?.stallMin);
const stallMin = Number.isFinite(configured) && configured >= 0 ? configured : STALL_MIN;
const progress = readProgressMap(c);
const holds = readHoldMap(c);
const areas = c.areas.map((a) => {
const view = paneView(c, a.id, idleAfterMs);
const view = views?.[a.id] || paneView(c, a.id, idleAfterMs);
const unread = unreadFiles(c, a.id);
const mark = progress[a.id] || null;
const ageMin = mark?.at ? (Date.now() - Date.parse(mark.at)) / 60000 : null;
const row = {
id: a.id, role: a.role, harness: a.harness,
window: view.window, unread: unread.length, activity: view.activity,
progress: mark && Number.isFinite(ageMin) ? { at: mark.at, line: mark.line, ageMin: Math.round(ageMin) } : null,
hold: holds[a.id] || null,
stalled: false,
stallReason: null,
};
const quiet = view.window === 'alive' && unread.length === 0;
if (quiet && mark && Number.isFinite(ageMin) && ageMin >= stallMin) {
row.stalled = true;
row.stallReason = `no progress for ${Math.round(ageMin)} min and no unread mail`;
} else if (quiet && !mark) {
const started = Date.parse(readRun(c, a.id)?.startedAt || '');
const upMin = Number.isFinite(started) ? (Date.now() - started) / 60000 : null;
if (upMin != null && upMin >= stallMin) {
row.stalled = true;
row.stallReason = `no checkpoint for ${Math.round(upMin)} min and no unread mail`;
}
}
const oldest = unread[0];
const age = oldest ? Date.now() - oldest.mtimeMs : 0;
const already = (state[a.id] || []).includes(oldest?.file);
+70
View File
@@ -0,0 +1,70 @@
// Per-area checkpoint and hold records. Status and the check-in read them.
// There is no timer. An area writes a line when work advances, and a hold
// when something is blocked, naming who clears it and what is still allowed.
import fs from 'node:fs';
import path from 'node:path';
import { area as getArea, die, writeAtomic } from '../lib.mjs';
export const STALL_MIN = 20;
function safeId(id) {
if (!/^[a-z0-9][a-z0-9-]{0,62}$/.test(id || '')) die(`bad area "${id ?? ''}"`);
return id;
}
function fileIn(c, kind, id) {
return path.join(c.stateDir, kind, `${safeId(id)}.json`);
}
function readMap(c, kind) {
const dir = path.join(c.stateDir, kind);
const out = {};
if (!fs.existsSync(dir)) return out;
for (const name of fs.readdirSync(dir)) {
if (!name.endsWith('.json')) continue;
try { out[name.slice(0, -5)] = JSON.parse(fs.readFileSync(path.join(dir, name), 'utf8')); }
catch { /* a torn write is skipped; the next checkpoint replaces it */ }
}
return out;
}
export function readProgressMap(c) {
return readMap(c, 'progress');
}
export function readHoldMap(c) {
return readMap(c, 'holds');
}
export function writeProgress(c, area, line) {
const id = safeId(area);
getArea(c, id);
const text = String(line || '').trim();
if (!text) die('progress needs one line saying what advanced');
const rec = { at: new Date().toISOString(), area: id, line: text.slice(0, 400) };
writeAtomic(fileIn(c, 'progress', id), JSON.stringify(rec) + '\n');
return rec;
}
export function writeHold(c, area, { owner, unblock, next } = {}) {
const id = safeId(area);
getArea(c, id);
const rec = {
at: new Date().toISOString(),
area: id,
owner: String(owner || '').trim(),
unblock: String(unblock || '').trim(),
next: String(next || '').trim(),
};
if (!rec.owner || !rec.unblock || !rec.next) die('a hold needs --owner, --unblock, and --next');
writeAtomic(fileIn(c, 'holds', id), JSON.stringify(rec) + '\n');
return rec;
}
export function clearHold(c, area) {
const id = safeId(area);
getArea(c, id);
const file = fileIn(c, 'holds', id);
if (fs.existsSync(file)) fs.unlinkSync(file);
return { area: id, cleared: true };
}
+169
View File
@@ -0,0 +1,169 @@
// Startup readiness. One mailbox round trip, the Codex approval path, health,
// durable deps, and the last checkpoints. No model call and no daemon.
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import { send, read, archive } from '../../adapters/mailbox.mjs';
import { fleetHealth } from './health.mjs';
import { readHoldMap, readProgressMap } from './progress.mjs';
const USAGE_RE = /five-hour|usage limit|%\s*remaining|rate limit/i;
export function codexConfigCandidates(repo, env = process.env, home = os.homedir()) {
const root = env.CODEX_HOME || path.join(home, '.codex');
return [path.join(root, 'config.toml'), path.join(repo, '.codex', 'config.toml')];
}
/**
* Read the agent-fleet approval mode out of a Codex config.toml.
* `approve` is the always-allow value. `auto` still requires approval for
* tools that are not marked read-only, and on-request auto-review then runs.
*/
export function parseCodexMcpApproval(text, server = 'agent-fleet') {
let section = '';
const result = { server: false, defaultMode: null, tools: {} };
for (const raw of String(text || '').split('\n')) {
const line = raw.replace(/\s+#.*$/, '').trim();
if (!line || line.startsWith('#')) continue;
const table = /^\[([^\]]+)\]$/.exec(line);
if (table) { section = table[1].trim(); continue; }
const kv = /^([A-Za-z0-9_.-]+)\s*=\s*(.+)$/.exec(line);
if (!kv) continue;
const val = kv[2].trim().replace(/^['"]|['"]$/g, '');
if (section === `mcp_servers.${server}` && kv[1] === 'default_tools_approval_mode') {
result.server = true;
result.defaultMode = val;
}
const tool = new RegExp(`^mcp_servers\\.${server}\\.tools\\.(.+)$`).exec(section);
if (tool && kv[1] === 'approval_mode') {
result.server = true;
result.tools[tool[1]] = val;
}
}
const sendMode = result.tools.mailbox_send;
const tellMode = result.tools.tell;
const mailOk = result.defaultMode === 'approve' || (sendMode === 'approve' && tellMode === 'approve');
return { ...result, mailOk };
}
export function readCodexApproval(repo, opt = {}) {
const files = [];
for (const file of codexConfigCandidates(repo, opt.env, opt.home)) {
if (!file || files.some((row) => row.file === file)) continue;
if (!fs.existsSync(file)) {
files.push({ file, found: false });
continue;
}
const parsed = parseCodexMcpApproval(fs.readFileSync(file, 'utf8'));
files.push({ file, found: true, ...parsed });
}
const hit = files.find((row) => row.mailOk) || files.find((row) => row.found && row.server) || null;
return {
files,
mailOk: Boolean(hit?.mailOk),
mode: hit?.defaultMode ?? null,
file: hit?.file ?? null,
note: 'fleet up runs codex with no --profile and does not set CODEX_HOME. The session reads the same config.toml as any other Codex process in that environment. The server name is agent-fleet. Shell fleet tell is a sandboxed command; default_tools_approval_mode does not apply to it.',
};
}
/** Prove send, read, and archive on an isolated mailbox. Does not wake a pane. */
export function mailboxRoundTrip(dir) {
const probe = path.join(dir, 'probe');
fs.rmSync(probe, { recursive: true, force: true });
const body = `ready ${Date.now()} ${process.pid}`;
try {
fs.mkdirSync(dir, { recursive: true });
const stamp = path.join(dir, `.write-${process.pid}`);
fs.writeFileSync(stamp, 'ok');
fs.unlinkSync(stamp);
const file = send(probe, 'ready', 'ready', body, 'round trip');
const msgs = read(probe, 'ready');
const hit = msgs.find((m) => m.text.includes(body));
if (!hit) return { ok: false, error: 'read did not see the message just written' };
archive(probe, 'ready', hit.file);
const left = read(probe, 'ready');
if (left.length) return { ok: false, error: 'archive left the message unread' };
return { ok: true, file: path.basename(file) };
} catch (e) {
return { ok: false, error: e instanceof Error ? e.message : String(e) };
} finally {
fs.rmSync(probe, { recursive: true, force: true });
}
}
export function depsReady(c) {
const root = path.join(c.stateDir, 'deps');
const manifest = path.join(root, 'manifest.txt');
const missing = [];
if (fs.existsSync(manifest)) {
for (const line of fs.readFileSync(manifest, 'utf8').split('\n')) {
const rel = line.trim();
if (!rel || rel.startsWith('#')) continue;
const file = path.resolve(root, rel);
if (file !== root && !file.startsWith(root + path.sep)) { missing.push(rel); continue; }
if (!fs.existsSync(file)) missing.push(rel);
}
}
const names = ['package.json', 'package-lock.json', 'pnpm-lock.yaml', 'yarn.lock', 'Cargo.lock', 'go.sum'];
const repoManifests = names.filter((name) => fs.existsSync(path.join(c.repo, name)));
return {
repoManifests,
manifest: fs.existsSync(manifest),
missing,
ok: missing.length === 0,
note: 'Keep runners, lockfiles, pinned archives, and evidence in the worktree or stateDir/deps. /tmp and the npm cache are disposable.',
};
}
export function usageSignal(c) {
const file = path.join(c.stateDir, 'wake-log.jsonl');
if (!fs.existsSync(file)) return { seen: false, line: null };
const lines = fs.readFileSync(file, 'utf8').split('\n').filter(Boolean).slice(-80);
const hit = [...lines].reverse().find((line) => USAGE_RE.test(line)) || null;
return { seen: Boolean(hit), line: hit ? hit.slice(0, 240) : null };
}
export function fleetReady(c, opt = {}) {
const mailbox = mailboxRoundTrip(path.join(c.stateDir, 'ready-mailbox'));
const approval = readCodexApproval(c.repo, opt);
const health = opt.health || fleetHealth(c, { sampleMs: 0, ...(opt.healthOpt || {}) });
const deps = depsReady(c);
const checkpoints = Object.values(readProgressMap(c));
const holds = Object.values(readHoldMap(c));
const usage = usageSignal(c);
return {
ok: mailbox.ok && deps.ok,
mailbox,
approval,
health: { verdict: health.verdict, notes: health.notes || [] },
deps,
checkpoints,
holds,
usage,
};
}
export function formatReady(report) {
const lines = [];
lines.push(report.mailbox.ok ? 'mailbox round trip ok' : `mailbox round trip failed: ${report.mailbox.error}`);
if (report.approval.mailOk) lines.push(`approval ${report.approval.mode || 'per-tool approve'} in ${report.approval.file}`);
else {
const seen = (report.approval.files || []).filter((row) => row.found && row.server);
const where = seen.length ? seen.map((row) => `${row.file} mode=${row.defaultMode || 'unset'}`).join('; ') : 'no [mcp_servers.agent-fleet] table found';
lines.push(`approval want default_tools_approval_mode = "approve" (or per-tool approval_mode = "approve" on mailbox_send and tell). ${where}. auto still asks, and on-request auto-review can reject the call. Offer to write approve only after the user says yes, then reload Codex.`);
}
lines.push(`health ${report.health.verdict}`);
for (const note of report.health.notes || []) lines.push(`health note ${note.words}`);
const manifests = report.deps.repoManifests.length ? report.deps.repoManifests.join(', ') : 'none in the repo root';
lines.push(`deps repo manifests: ${manifests}${report.deps.manifest ? '; stateDir/deps/manifest.txt present' : ''}`);
if (report.deps.missing.length) lines.push(`deps missing ${report.deps.missing.join(', ')}`);
lines.push(`deps note ${report.deps.note}`);
if (!report.checkpoints.length) lines.push('checkpoints none yet');
for (const row of report.checkpoints) lines.push(`checkpoint ${row.area} ${row.at} ${row.line}`);
if (!report.holds.length) lines.push('holds none');
for (const row of report.holds) lines.push(`hold ${row.area} owner ${row.owner}; unblock ${row.unblock}; next ${row.next}`);
if (report.usage.seen) lines.push('usage a recent wake line mentioned a usage limit. Checkpoint and hand off; resume from the checkpoints after the window resets.');
lines.push('resume continue from the checkpoints. A live window is not progress.');
return lines.join('\n');
}