src/server/background-jobs.ts
Rama main · réplica de lectura del repositorio original.
// Background shell jobs that OUTLIVE the turn that started them.
//
// The bug this exists for: an agent dispatches long work, promises "I'll report back", and the turn
// is then cut at the engine's wall clock. The work finishes fine — and nobody is left to say so.
// Any dispatch-and-poll pattern loses that race whenever the work outlasts the remaining budget, so
// polling is not the fix; something has to WAKE the session when the process exits.
//
// Shape:
// • The registry is owned by the SERVER (module singleton), not by the turn. An add-on engine builds a fresh
// `Agent` per turn, so a registry created there would die with the turn — the very failure above.
// Hosts hand each turn a thin per-session VIEW (`sessionJobRegistry`) over this one store; the
// view is duck-compatible with the add-on engine's `ShellJobRegistry`, so `Shell({background:true})` and the
// ShellOutput/ShellStatus/ShellKill tools light up with no folklore `nohup`.
// • On exit we report through wake.ts — the same "close then report" path polls.ts already uses in
// production, so a job outcome lands wherever the session speaks (Slack thread / web UI).
// • No double-report: reading a finished job's OUTPUT from inside a turn marks it `observed` — the
// model had the result in hand and can speak for itself. Only that. `status`/`list` return no
// output, so seeing "exited" there is not holding the result; marking those observed suppressed
// the follow-up and lost the outcome entirely. We wake for every job whose output nobody read.
// (Known narrow gap: a turn that reads the output and is then cut before it speaks gets no
// follow-up. Much smaller window than the one being fixed.)
// • No collision with a live turn: the report defers while `isSessionLocked` — with a cap, past
// which it is delivered as plain text rather than never.
//
// Restart survival is PARTIAL and deliberately visible. Children are detached, so they keep running
// across a server restart and their output keeps landing in the job's log file; boot re-adopts them
// by pid and reports on exit as usual. What cannot survive is the exit CODE of a job that finished
// while we were down — nothing was waiting on the process. Those are reported with an explicit
// "exit code unknown (server restarted)" instead of a fabricated success.
import { spawn, execFileSync } from 'node:child_process';
import { mkdirSync, openSync, closeSync, readSync, readFileSync, writeFileSync, readdirSync, statSync, rmSync } from 'node:fs';
import path from 'node:path';
import { dataPath } from './paths.ts';
import { isSessionLocked, getSession } from './sessions.ts';
import { wakeSession, deliverToSession, wakeReady } from './wake.ts';
const PREFIX = '[jobs]';
// ── Bounds. Every one of these caps a blast radius the incident's `nohup` had none of. ──
/** Hard cap on a job's log, enforced by `head -c` inside the job's own pipeline (see buildShell).
* Polling a size and killing late cannot bound anything: at a 5s cadence an 8MB cap measured 830MB
* on disk. `head` closes the pipe at exactly this many bytes instead — the writer then dies of
* SIGPIPE, so the bound is the kernel's, not a timer's. */
const MAX_LOG_BYTES = 8 * 1024 * 1024;
/** How much of the tail we ever read back into memory (for a tool result or a report). */
const TAIL_BYTES = 16 * 1024;
/** How much of that tail goes into the wake prompt (the rest stays readable via ShellOutput). */
const REPORT_CHARS = 4_000;
const MAX_RUNNING_PER_SESSION = 4;
const MAX_RUNNING_GLOBAL = 16;
/** Grace before reporting: lets a still-live turn poll the just-finished job and own the telling. */
const GRACE_MS = 8_000;
/** Re-check cadence while the session is busy, or while an adopted (post-restart) job still runs. */
const POLL_MS = 5_000;
/** Give up deferring behind a busy session after this long and deliver the outcome as plain text. */
const MAX_DEFER_MS = 30 * 60_000;
/** Delete finished job records + logs after this long. */
const PRUNE_AFTER_MS = 24 * 60 * 60_000;
export type JobStatus = 'running' | 'exited' | 'killed' | 'error';
export interface JobOwner {
sessionId: string;
uid: string;
userEmail?: string;
/** Working directory for the job's shell (the deployment/project root). */
cwd: string;
/** Extra env merged over the server's own for the child. */
env?: Record<string, string>;
}
interface JobRecord {
id: string;
sessionId: string;
uid: string;
userEmail?: string;
command: string;
cwd: string;
pid?: number;
startedAt: number;
status: JobStatus;
exitCode?: number;
endedAt?: number;
/** A turn read this job's terminal state — the model can report it itself, so we must not. */
observed?: boolean;
/** Terminal outcome of the follow-up delivery (absent = not attempted yet). */
reported?: 'woke' | 'raw' | 'skipped-observed' | 'no-session' | 'failed';
/** The exit code is unknown because the server restarted while this job ran. */
orphaned?: boolean;
/** The job hit MAX_LOG_BYTES: its output was cut off and the command was killed by SIGPIPE. */
truncated?: boolean;
}
// ── Storage ────────────────────────────────────────────────────────────────────
const dir = (): string => { const d = dataPath('jobs'); mkdirSync(d, { recursive: true }); return d; };
const recFile = (id: string): string => path.join(dir(), `${id}.json`);
const logFile = (id: string): string => path.join(dir(), `${id}.log`);
/** The command's REAL exit status, written by the job's own shell (see buildShell). */
const statusFile = (id: string): string => path.join(dir(), `${id}.status`);
/** The command's exit code as its shell recorded it, or null if it never got that far. */
function readStatus(id: string): number | null {
try {
const n = Number(readFileSync(statusFile(id), 'utf-8').trim());
return Number.isInteger(n) ? n : null;
} catch { return null; }
}
/** Current size of a job's log, 0 if it does not exist yet. */
function logSize(id: string): number {
try { return statSync(logFile(id)).size; } catch { return 0; }
}
/**
* Wrap the user's command so the log is bounded and the real exit code survives.
*
* `{ ( cmd ); echo $? > status; } 2>&1 | head -c N`, and every piece of that shape is load-bearing:
* • `head -c N` — the actual disk bound. It closes the pipe at exactly N bytes; the writer then
* takes SIGPIPE. Chosen over `ulimit -f`, which is also kernel-enforced but caps EVERY file the
* job writes — that would kill an ordinary build the moment it emitted a >8MB artifact.
* • the inner `( … )` subshell — a command ending in `exit 3` (agents write those) would otherwise
* exit the group before the status line ever ran, and we'd lose the code. Verified.
* • the status file — the pipeline's own exit code is `head`'s (always 0), so without this every
* job would report success. The shell is not the process writing to the pipe, so it survives the
* SIGPIPE that kills the command and still records the code.
* It also outlives us: the whole pipeline is in the job's detached process group, so a job adopted
* after a server restart can be given its true exit code instead of "unknown".
*/
function buildShell(id: string, cmd: string): string {
return `{\n(\n${cmd}\n)\n__st=$?\necho $__st > '${statusFile(id)}'\n} 2>&1 | head -c ${MAX_LOG_BYTES}`;
}
/** In-memory mirror of the on-disk records (disk is the restart-durable copy; this is the hot path). */
const records = new Map<string, JobRecord>();
/** Live timers per job, so nothing is left ticking after a job is reported or the record pruned. */
const timers = new Map<string, ReturnType<typeof setTimeout>>();
function save(j: JobRecord): void {
records.set(j.id, j);
try { writeFileSync(recFile(j.id), JSON.stringify(j, null, 2)); }
catch (e) { console.error(`${PREFIX} persist ${j.id} failed:`, (e as Error).message); }
}
function setTimer(id: string, ms: number, fn: () => void): void {
clearTimer(id);
timers.set(id, setTimeout(() => { timers.delete(id); fn(); }, ms));
}
function clearTimer(id: string): void {
const t = timers.get(id);
if (t) { clearTimeout(t); timers.delete(id); }
}
/** Last `TAIL_BYTES` of a job's log, decoded as utf-8. Never loads the whole file. */
function readTail(id: string): string {
let fd: number | undefined;
try {
const size = statSync(logFile(id)).size;
const start = Math.max(0, size - TAIL_BYTES);
const len = size - start;
if (len <= 0) return '';
const buf = Buffer.allocUnsafe(len);
fd = openSync(logFile(id), 'r');
readSync(fd, buf, 0, len, start);
// A head-truncated read can split a multi-byte char — the replacement char is cosmetic, not corruption.
return (start > 0 ? `…(truncated, showing last ${TAIL_BYTES >> 10}KB)\n` : '') + buf.toString('utf-8');
} catch { return ''; }
finally { if (fd !== undefined) try { closeSync(fd); } catch { /* already closed */ } }
}
const runningCount = (sessionId?: string): number =>
[...records.values()].filter((j) => j.status === 'running' && (!sessionId || j.sessionId === sessionId)).length;
/**
* Seconds this pid has been alive, or null if there is no such process.
*
* `etime` (formatted `[[DD-]HH:]MM:SS`), NOT `etimes` (raw seconds): etimes is a procps/Linux
* extension and BSD ps — macOS, which is what the live box runs — rejects it outright, so the whole
* adoption check silently degraded to "process gone" and every job that survived a restart was
* reported as `exit code unknown`. etime is in POSIX and works on both.
*/
function pidAgeSeconds(pid: number): number | null {
let out: string;
try {
// stderr piped: a missing pid is an expected outcome here, not something to print.
out = execFileSync('ps', ['-o', 'etime=', '-p', String(pid)], { encoding: 'utf-8', stdio: ['ignore', 'pipe', 'pipe'] }).trim();
} catch { return null; /* no such process */ }
const m = out.match(/^(?:(?:(\d+)-)?(\d+):)?(\d+):(\d+)$/);
if (!m) return null;
const [, d, h, mi, sec] = m;
return Number(d ?? 0) * 86400 + Number(h ?? 0) * 3600 + Number(mi) * 60 + Number(sec);
}
/**
* Env for a job's shell: the server's own, minus anything that looks like a credential.
*
* An add-on engine's foreground `Bash` redacts these by default (RealShellOptions.redactEnv). Backgrounding a
* command must not be a way around that — otherwise `Bash({background:true})` becomes a strictly
* weaker sandbox than `Bash({})`, and `echo $ANTHROPIC_API_KEY` lands in a job log we then feed back
* into the model. Mirrors the add-on engine's SECRET_ENV_RE; a job that genuinely needs a credential gets it
* explicitly via `JobOwner.env`.
*/
const SECRET_ENV_RE = /(API_KEY|_TOKEN|_SECRET|_PASSWORD|_PRIVATE_KEY|^AWS_|^GITHUB_TOKEN$|^OPENAI_|^ANTHROPIC_|^GOOGLE_|^GEMINI_|^GROQ_|^NPM_TOKEN$|^SLACK_)/i;
function childEnv(): Record<string, string | undefined> {
const out: Record<string, string | undefined> = {};
for (const [k, v] of Object.entries(process.env)) if (!SECRET_ENV_RE.test(k) && k !== 'PORT') out[k] = v; // PORT = the server's own port; a job honouring it would shadow us
return out;
}
// ── Lifecycle ──────────────────────────────────────────────────────────────────
/** Start `command` detached, in its own process group, with stdout+stderr appended to the job log.
* Detached (setsid) for two reasons: the group is killable as a subtree, and the child is NOT torn
* down with the server — which is what makes restart adoption possible at all. */
export async function startJob(owner: JobOwner, command: string): Promise<string> {
const cmd = command.trim();
if (!cmd) throw new Error('empty command');
if (runningCount(owner.sessionId) >= MAX_RUNNING_PER_SESSION)
throw new Error(`too many background jobs in this session (max ${MAX_RUNNING_PER_SESSION}) — wait for one to finish or ShellKill it`);
if (runningCount() >= MAX_RUNNING_GLOBAL)
throw new Error(`too many background jobs on this server (max ${MAX_RUNNING_GLOBAL})`);
const id = `job-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
const j: JobRecord = {
id, sessionId: owner.sessionId, uid: owner.uid, userEmail: owner.userEmail,
command: cmd, cwd: owner.cwd, startedAt: Date.now(), status: 'running',
};
let fd: number;
try { fd = openSync(logFile(id), 'a'); }
catch (e) { throw new Error(`cannot open job log: ${(e as Error).message}`); }
try {
// stdin is /dev/null so a child can never block on (or steal) input; stdout+stderr go straight
// to the fd — nothing is buffered in this process, so a chatty job costs disk, not memory.
const proc = spawn('/bin/sh', ['-c', buildShell(id, cmd)], {
cwd: owner.cwd,
env: {
...childEnv(),
...owner.env,
// A long-running launcher can gate on these: only a REGISTERED job gets a completion wake,
// so one started in the foreground orphans silently when its 60s tool call is killed.
SHRAGA_JOB_ID: id,
SHRAGA_BG_JOB: '1',
// The owning session, so a launcher never has to be HANDED it. A model writing the command
// has to remember `export` for a var to survive into `sh script.sh`; a plain assignment
// silently doesn't, which failed every scheduled social run for a day. The runner already
// knows whose job this is — passing it removes the caller from the loop entirely.
SHRAGA_SESSION_ID: owner.sessionId,
},
detached: true, stdio: ['ignore', fd, fd],
});
j.pid = proc.pid;
proc.on('error', (err) => finish(id, 'error', undefined, `spawn error: ${err.message}`));
// The pipeline's own code is `head`'s; the command's real one is in the status file.
proc.on('close', (code) => finish(id, 'exited', readStatus(id) ?? code ?? undefined));
proc.unref(); // never hold the event loop open for a background job
} catch (e) {
j.status = 'error';
save(j);
closeSync(fd);
throw new Error(`failed to spawn: ${(e as Error).message}`);
}
closeSync(fd); // the child holds its own dup of the fd
save(j);
console.log(`${PREFIX} started ${id} pid=${j.pid} session=${j.sessionId.slice(0, 8)} cmd=${cmd.slice(0, 120)}`);
return id;
}
/** The little of a spawned child this module needs to adopt one (node's ChildProcess satisfies it). */
type AdoptableProcess = {
pid?: number | null;
stdout?: { on(ev: 'data', cb: (chunk: unknown) => void): void } | null;
stderr?: { on(ev: 'data', cb: (chunk: unknown) => void): void } | null;
on(ev: 'close' | 'error', cb: (arg: never) => void): void;
unref?: () => void;
};
/**
* Register an ALREADY-RUNNING child as a background job. The Shell tool hands a foreground command
* over at its deadline instead of killing it: the work continues, and from here it is an ordinary job
* — polled with ShellOutput, and reported back to the session when it ends. Without this the deadline
* stays a kill, which is what abandoned a scheduled run midway.
*/
export function adoptJob(owner: JobOwner, command: string, proc: AdoptableProcess, seed = ''): string {
const id = `job-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
const j: JobRecord = {
id, sessionId: owner.sessionId, uid: owner.uid, userEmail: owner.userEmail,
command: command.trim(), cwd: owner.cwd, startedAt: Date.now(), status: 'running',
pid: proc.pid ?? undefined,
};
// The child was spawned with pipes by the caller, so output is appended here rather than dup'd to
// an fd the way a job we spawn ourselves does it.
const append = (chunk: unknown) => {
const text = typeof chunk === 'string' ? chunk : String(chunk ?? '');
if (!text) return;
try { writeFileSync(logFile(id), text, { flag: 'a' }); } catch { /* the job outlives its log */ }
};
if (seed) append(seed);
proc.stdout?.on('data', append);
proc.stderr?.on('data', append);
proc.on('error', ((err: Error) => finish(id, 'error', undefined, `adopted job error: ${err?.message ?? err}`)) as never);
proc.on('close', ((code: number | null) => finish(id, 'exited', code ?? undefined)) as never);
proc.unref?.();
records.set(id, j);
save(j);
console.log(`${PREFIX} adopted ${id} pid=${j.pid} session=${j.sessionId.slice(0, 8)} cmd=${j.command.slice(0, 120)}`);
return id;
}
/** Record a terminal state exactly once, then schedule the follow-up. */
function finish(id: string, status: JobStatus, exitCode?: number, note?: string): void {
const j = records.get(id);
if (!j || j.status !== 'running') return; // already terminal (e.g. killed, then close fires)
j.status = status;
j.exitCode = exitCode;
j.endedAt = Date.now();
if (logSize(id) >= MAX_LOG_BYTES) {
// Say so IN the log: every path the model can read the result through goes via readTail, so the
// notice reaches it whether it polls ShellOutput or gets the wake report.
j.truncated = true;
try { writeFileSync(logFile(id), `\n[output limit: ${MAX_LOG_BYTES >> 20}MB reached — output was cut off here and the command was terminated]\n`, { flag: 'a' }); }
catch { /* nothing more we can do about the log */ }
console.warn(`${PREFIX} ${id} hit the ${MAX_LOG_BYTES >> 20}MB log cap — output truncated`);
}
save(j);
console.log(`${PREFIX} ${id} ${status}${exitCode != null ? ` exit=${exitCode}` : ''} in ${Math.round((j.endedAt - j.startedAt) / 1000)}s${note ? ` (${note})` : ''}`);
// Grace: a turn that is still alive gets first refusal on telling the user (see `observed`).
setTimer(id, GRACE_MS, () => { void report(id, Date.now()); });
}
export function killJob(id: string): boolean {
const j = records.get(id);
if (!j) return false;
if (j.status === 'running') {
// Group-kill: the child is its own group leader, so signalling `-pid` reaps the whole subtree —
// `kill(pid)` alone would hit /bin/sh and orphan whatever it forked.
if (j.pid) { try { process.kill(-j.pid, 'SIGTERM'); } catch { try { process.kill(j.pid, 'SIGTERM'); } catch { /* already gone */ } } }
finish(id, 'killed');
}
return true;
}
/** A job the model explicitly killed inside a turn needs no follow-up — it already knows. */
function markObservedIfTerminal(j: JobRecord): void {
if (j.status !== 'running' && !j.observed) { j.observed = true; save(j); }
}
// ── Follow-up delivery ─────────────────────────────────────────────────────────
function reportPrompt(j: JobRecord): string {
const dur = Math.round(((j.endedAt ?? Date.now()) - j.startedAt) / 1000);
const result = (j.orphaned
? 'ended while the server was restarting — exit code unknown'
: j.status === 'killed' ? 'killed'
: j.status === 'error' ? 'failed to run'
: `exit ${j.exitCode ?? 0}`)
+ (j.truncated ? `, cut off at the ${MAX_LOG_BYTES >> 20}MB output limit` : '');
const tail = readTail(j.id).slice(-REPORT_CHARS).trim();
return [
`[Background job finished] A command you started in an earlier turn has completed, after that turn ended.`,
`Command: \`${j.command}\``,
`Result: ${result} (after ${dur}s)`,
``,
`Output (tail):`,
tail || '(no output)',
``,
`Report this outcome to the user now — that turn promised a follow-up and this is it.`,
`Be brief and concrete. Do NOT re-run the command.`,
].join('\n');
}
/** Plain-text form, for when we cannot run a turn (no runner, or the session never went idle). */
function rawReport(j: JobRecord): string {
const dur = Math.round(((j.endedAt ?? Date.now()) - j.startedAt) / 1000);
const result = (j.orphaned ? 'exit code unknown (server restarted)' : j.status === 'exited' ? `exit ${j.exitCode ?? 0}` : j.status)
+ (j.truncated ? `, cut off at the ${MAX_LOG_BYTES >> 20}MB output limit` : '');
const tail = readTail(j.id).slice(-REPORT_CHARS).trim();
return `Background job finished — \`${j.command}\`\nResult: ${result} (after ${dur}s)\n\n\`\`\`\n${tail || '(no output)'}\n\`\`\``;
}
/**
* Deliver a finished job's outcome, once.
*
* Order matters: `observed` is checked AFTER the grace window, so a turn that was still alive when
* the job exited has had its chance to poll and own the telling. `firstAttemptAt` bounds how long
* we defer behind a busy session.
*/
const report = (id: string, firstAttemptAt: number): Promise<void> =>
tryReport(id, firstAttemptAt).catch((e) => console.error(`${PREFIX} ${id} report threw:`, (e as Error).message));
async function tryReport(id: string, firstAttemptAt: number): Promise<void> {
const j = records.get(id);
if (!j || j.reported) return;
if (j.observed) {
j.reported = 'skipped-observed';
save(j);
console.log(`${PREFIX} ${id} observed inside a turn — no follow-up needed`);
return;
}
if (!getSession(j.sessionId)) {
j.reported = 'no-session';
save(j);
console.warn(`${PREFIX} ${id} session ${j.sessionId} is gone — dropping follow-up`);
return;
}
// Never start a turn on top of a live one — they would interleave into one transcript.
if (isSessionLocked(j.sessionId)) {
if (Date.now() - firstAttemptAt < MAX_DEFER_MS) {
setTimer(id, POLL_MS, () => { void report(id, firstAttemptAt); });
return;
}
console.warn(`${PREFIX} ${id} session busy for ${Math.round(MAX_DEFER_MS / 60_000)}min — delivering raw`);
await deliverToSession({ sessionId: j.sessionId, uid: j.uid, text: rawReport(j), title: 'Background job' })
.catch((e) => console.error(`${PREFIX} raw deliver failed:`, (e as Error).message));
j.reported = 'raw';
save(j);
return;
}
try {
if (!wakeReady()) {
await deliverToSession({ sessionId: j.sessionId, uid: j.uid, text: rawReport(j), title: 'Background job' });
j.reported = 'raw';
} else {
const outcome = await wakeSession({
sessionId: j.sessionId, uid: j.uid, userEmail: j.userEmail,
prompt: reportPrompt(j), channel: 'job', title: 'Background job',
});
if (outcome === 'woke') j.reported = 'woke';
else {
// The wake ran but said nothing (or the session vanished mid-flight). Falling back to raw
// keeps the promise: the user sees the result rather than nothing at all.
await deliverToSession({ sessionId: j.sessionId, uid: j.uid, text: rawReport(j), title: 'Background job' });
j.reported = 'raw';
console.warn(`${PREFIX} ${id} wake returned '${outcome}' — delivered raw instead`);
}
}
console.log(`${PREFIX} ${id} follow-up delivered (${j.reported})`);
} catch (e) {
j.reported = 'failed';
console.error(`${PREFIX} ${id} follow-up FAILED:`, (e as Error).message);
}
save(j);
}
// ── Boot: adopt what survived, prune what is stale ─────────────────────────────
/**
* Load persisted records and reconcile them with reality.
*
* `running` records are jobs we lost the `close` listener for when the process ended. Their children
* are detached, so many are genuinely still running — re-adopt those by polling the pid. The rest
* finished while we were down: we know THAT they ended but not with what code, which is reported
* honestly rather than guessed.
*/
export function initBackgroundJobs(): void {
let files: string[] = [];
try { files = readdirSync(dir()).filter((f) => f.endsWith('.json')); } catch { return; }
let adopted = 0, orphaned = 0;
for (const f of files) {
let j: JobRecord;
try { j = JSON.parse(readFileSync(path.join(dir(), f), 'utf-8')) as JobRecord; } catch { continue; }
if (!j?.id) continue;
records.set(j.id, j);
if (j.status !== 'running') {
// A terminal job whose follow-up never went out (we died between finish and report) still owes one.
if (!j.reported) setTimer(j.id, GRACE_MS, () => { void report(j.id, Date.now()); });
continue;
}
// pid liveness, guarded against pid REUSE. Our process started when the job did, so a pid that
// is still ours is AT LEAST as old as the job. A recycled pid belongs to a process the OS started
// later — i.e. YOUNGER than the job — which is exactly what this rejects. (Getting the comparison
// backwards adopts the stranger: we would report its exit as the job's, and ShellKill would
// `kill(-pid)` an unrelated process tree.) The 5s slack absorbs `etime`'s 1s granularity.
const age = j.pid ? pidAgeSeconds(j.pid) : null;
const jobAgeSec = (Date.now() - j.startedAt) / 1000;
if (age != null && age >= jobAgeSec - 5) { adoptRunning(j.id); adopted++; }
else {
if (age != null) console.warn(`${PREFIX} ${j.id} pid ${j.pid} is ${Math.round(age)}s old but the job is ${Math.round(jobAgeSec)}s old — pid was reused, not adopting`);
// Its own shell may still have recorded the real code before we went down.
const st = readStatus(j.id);
if (st != null) j.exitCode = st; else j.orphaned = true;
j.status = 'exited';
j.endedAt = Date.now();
save(j);
orphaned++;
setTimer(j.id, GRACE_MS, () => { void report(j.id, Date.now()); });
}
}
pruneOld();
setInterval(pruneOld, 60 * 60_000).unref?.();
if (adopted || orphaned) console.log(`${PREFIX} boot: adopted ${adopted} running job(s), ${orphaned} ended while down`);
}
/** Watch a job we no longer have a child handle for; report when its pid disappears. */
function adoptRunning(id: string): void {
const j = records.get(id);
if (!j || j.status !== 'running' || !j.pid) return;
if (pidAgeSeconds(j.pid) == null) {
// `wait` only works for your own child, so the pid vanishing tells us THAT it ended, not how.
// Its own shell wrote the code down before exiting, though — prefer that over guessing.
const st = readStatus(id);
if (st == null) j.orphaned = true;
finish(id, 'exited', st ?? undefined);
return;
}
setTimer(id, POLL_MS, () => adoptRunning(id));
}
function pruneOld(): void {
const now = Date.now();
for (const j of [...records.values()]) {
if (j.status === 'running' || !j.endedAt || now - j.endedAt < PRUNE_AFTER_MS) continue;
clearTimer(j.id);
records.delete(j.id);
for (const p of [recFile(j.id), logFile(j.id), statusFile(j.id)]) try { rmSync(p); } catch { /* already gone */ }
}
}
// ── The per-turn view handed to the agent engine ───────────────────────────────
/**
* A session-scoped facade over the process-wide store, duck-compatible with the add-on engine's
* `ShellJobRegistry` — pass it as `makeRealShellTool({ registry })` and to `makeShellJobTools`.
*
* Cheap to build per turn (it holds no state); the jobs it starts belong to the server, so they
* outlive the turn that made it. Every lookup is session-scoped: one session's agent can neither
* read nor kill another's jobs.
*/
export function sessionJobRegistry(owner: JobOwner) {
const mine = (id: string): JobRecord | undefined => {
const j = records.get(id);
return j && j.sessionId === owner.sessionId ? j : undefined;
};
return {
start: (command: string) => startJob(owner, command),
adopt: (command: string, proc: AdoptableProcess, seed?: string) => adoptJob(owner, command, proc, seed),
output(id: string): string | null {
const j = mine(id);
if (!j) return null;
markObservedIfTerminal(j);
return readTail(id);
},
// NB: status() and list() deliberately do NOT mark the job observed — they return no output, so
// the model has not got the result and still needs the follow-up. Only output() may.
status(id: string): { status: JobStatus; exitCode?: number; bytes: number } | null {
const j = mine(id);
if (!j) return null;
return { status: j.status, exitCode: j.exitCode, bytes: logSize(id) };
},
list(): Array<{ id: string; command: string; status: JobStatus }> {
return [...records.values()]
.filter((j) => j.sessionId === owner.sessionId)
.map((j) => ({ id: j.id, command: j.command, status: j.status }));
},
kill(id: string): boolean {
const j = mine(id);
if (!j) return false;
const ok = killJob(id);
markObservedIfTerminal(j); // the model asked for this end — it does not need telling about it
return ok;
},
};
}
/** Test/introspection seam: the current record for a job (undefined once pruned). */
export function getJob(id: string): Readonly<JobRecord> | undefined { return records.get(id); }