fix(daemon): archive owned Codex threads after terminal completion (#8051)

* fix(daemon): archive owned Codex threads after terminal completion

* fix(daemon): archive owned Codex threads after interrupted close

---------

Co-authored-by: lefarcen <935902669@qq.com>
This commit is contained in:
open-design-crew[bot]
2026-09-14 03:52:01 +00:00
committed by GitHub
co-authored by lefarcen
parent 1424be9727
commit 5a3f0c8923
19 changed files with 2149 additions and 40 deletions
@@ -0,0 +1,78 @@
import type { ChildProcess } from 'node:child_process';
import type { CodexAppServerSession } from './session.js';
import { cleanupClosedCodexThread, type CodexThreadCleanupOptions, type CodexThreadCleanupResult } from './thread-cleanup.js';
export type CodexCleanupInvocation = Omit<CodexThreadCleanupOptions, 'receipt'>;
type Child = Pick<ChildProcess, 'once' | 'removeListener'>;
type Work = { child: Child; onClose: () => void; started: boolean; completion: Promise<void>; complete: () => void };
/** Own the original child-close boundary independently of mutable run attempts. */
export function createCodexThreadCleanupOwner() {
const pending = new Set<Work>();
const bound = new WeakSet<Child>();
let expired = false;
return {
bind(child: Child, session: CodexAppServerSession,
invocation: CodexCleanupInvocation, report: (result: CodexThreadCleanupResult) => void,
evidenceCollected?: Promise<void>) {
if (expired || bound.has(child)) return;
bound.add(child);
// Later config updates or retry generations cannot change this process's
// executable, wrapper arguments, environment or workspace authority.
const context = { ...invocation, args: [...invocation.args], env: { ...invocation.env } };
let complete: () => void = () => {};
const completion = new Promise<void>(resolve => { complete = resolve; });
const finish = () => { pending.delete(work); complete(); };
const work: Work = { child, started: false, completion, complete,
onClose: () => {
// bind() must run after attach(): the session's earlier close listener
// establishes the proof before the ordinary run close handler aborts.
const receipt = session.takeClosedThreadCleanup();
if (!receipt) { finish(); return; }
void (async () => {
let result: CodexThreadCleanupResult;
try {
// Archive moves rollout paths. Keep the close-time receipt, but
// let this attempt finish reading its evidence before moving it.
await evidenceCollected;
if (expired) return;
work.started = true;
result = await cleanupClosedCodexThread({ ...context, receipt });
}
catch (error) { result = { status: 'failed', message: String(error) }; }
try { report(result); }
catch (error) { console.warn('[codex] thread cleanup diagnostic failed', error); }
})().finally(finish);
},
};
pending.add(work);
child.once('close', work.onClose);
},
async drain(timeoutMs: number): Promise<{ pending: number }> {
if (pending.size === 0) return { pending: 0 };
const budget = Number.isFinite(timeoutMs) ? Math.max(0, Math.min(3000, timeoutMs)) : 3000;
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
Promise.all([...pending].map(work => work.completion)).then(() => ({ pending: pending.size })),
new Promise<{ pending: number }>(resolve => {
timer = setTimeout(() => {
expired = true;
const remaining = pending.size;
// No late close may launch a fresh cleanup after shutdown's
// deadline. Already started helpers retain their own bounded
// process cleanup and still report their independent result.
for (const work of pending) {
if (work.started) continue;
work.child.removeListener('close', work.onClose);
pending.delete(work);
work.complete();
}
resolve({ pending: remaining });
}, budget);
}),
]);
} finally { clearTimeout(timer); }
},
};
}
@@ -32,22 +32,30 @@
* including a JSON-RPC error response, is emitted as an `error` AGENT
* event so it flows through `sendAgentEvent`'s classifier.
*
* On `experimentalApi`: NOT enabled. Measured on codex-cli 0.149.1 across four
* configurations × two repetitions of one fixed prompt, `item/agentMessage/
* delta` arrived on every run regardless of the flag (52/54 vs 45/42 frames),
* and `item/reasoning/summaryTextDelta` was gated purely by the reasoning
* summary setting, not by the flag. The capability's own doc comment is "opt
* into receiving experimental API methods and fields" — declaring it would buy
* nothing measurable and widen the surface most likely to change.
* OD-owned threads opt into experimentalApi for persistent paginated history.
* Codex 0.146.0 only locks paginated histories across processes when archiving;
* its default legacy history can be archived while another process writes it.
* Streaming itself does not require the experimental capability.
*/
import { createCodexAppServerNormalizer } from './normalize.js';
import { codexHistoryCapabilities } from './thread-cleanup.js';
type JsonObject = Record<string, unknown>;
export type CodexSandboxMode = 'read-only' | 'workspace-write' | 'danger-full-access';
const closedThreadProof = Symbol('closed owned Codex thread');
/** Issued only by this session after its child closes; never reconstructed from a user ID. */
export interface CodexClosedThreadCleanup {
readonly threadId: string;
readonly historyMode: 'paginated' | 'legacy';
readonly [closedThreadProof]: true;
}
export interface CodexAppServerSessionOptions {
child: {
once?(event: 'close', listener: () => void): unknown;
stdout: { on(event: 'data', listener: (chunk: unknown) => void): unknown };
stdin: {
write(chunk: string, cb?: (err?: Error | null) => void): unknown;
@@ -63,6 +71,10 @@ export interface CodexAppServerSessionOptions {
serviceTier?: string | null;
sandboxMode: CodexSandboxMode;
resumeSessionId?: string | null;
/** Internal daemon policy; never infer ownership from a thread title. */
manageThreadVisibility?: boolean;
/** The daemon validated this resume handle against its captured session record. */
resumeSessionOwned?: boolean;
imagePaths?: string[];
clientVersion?: string;
onAgentEvent: (event: JsonObject) => void;
@@ -83,6 +95,8 @@ export interface CodexAppServerSession {
completedSuccessfully(): boolean;
getDurableSessionId(): string | null;
getLastSessionPath(): string | null;
/** Once-only cleanup authority, unavailable until this exact child has closed. */
takeClosedThreadCleanup(): CodexClosedThreadCleanup | null;
stats(): { unknownNotifications: number; unknownItems: number };
}
@@ -143,6 +157,14 @@ export function attachCodexAppServerSession(
let turnEnded = false;
let turnSucceeded = false;
let handleReported = false;
let terminalReceived = false;
let childClosed = false;
let cleanupThread: CodexClosedThreadCleanup | null = null;
let canArchiveSafely = false;
let supportsPaginatedHistory = false;
let protectsLegacyHistory = false;
let archiveTimer: ReturnType<typeof setTimeout> | undefined;
const ownsThread = !resumeSessionId || opts.resumeSessionOwned === true;
function write(frame: JsonObject, onWritten?: () => void): void {
const stdin = child.stdin;
@@ -160,11 +182,13 @@ export function attachCodexAppServerSession(
params: JsonObject,
onResult: (result: JsonObject) => void,
onWritten?: () => void,
onError?: (error: unknown) => void,
): void {
const id = nextId++;
pending.set(id, (frame) => {
if (frame.error !== undefined) {
reportFatal(rpcErrorText(method, frame.error));
if (onError) onError(frame.error);
else reportFatal(rpcErrorText(method, frame.error));
return;
}
onResult(isRecord(frame.result) ? frame.result : {});
@@ -240,6 +264,14 @@ export function attachCodexAppServerSession(
const id = typeof thread?.id === 'string' ? thread.id : null;
if (id) {
threadId = id;
canArchiveSafely = opts.manageThreadVisibility === true
&& (protectsLegacyHistory || (supportsPaginatedHistory && thread?.historyMode === 'paginated'));
if (canArchiveSafely && ownsThread && id.trim()
&& (!resumeSessionId || id === resumeSessionId)) {
cleanupThread = Object.freeze({ threadId: id,
historyMode: thread?.historyMode === 'paginated' ? 'paginated' : 'legacy',
[closedThreadProof]: true as const });
}
const path = typeof thread?.path === 'string' ? thread.path : '';
if (path) rolloutPath = path;
// `thread/start` is followed by a `thread/started` notification, but
@@ -255,10 +287,36 @@ export function attachCodexAppServerSession(
reportFatal('codex app-server returned no thread id');
};
if (resumeSessionId) {
request('thread/resume', { ...shared, threadId: resumeSessionId }, onThread);
request(
'thread/resume', { ...shared, threadId: resumeSessionId }, onThread,
undefined,
(error) => {
const message = rpcErrorText('thread/resume', error);
if (opts.manageThreadVisibility !== true || !ownsThread || !/\bis archived\b/u.test(message)) {
reportFatal(message);
return;
}
archiveTimer = setTimeout(() => {
reportFatal('codex app-server thread/unarchive timed out');
shutdown();
}, 1_500);
request('thread/unarchive', { threadId: resumeSessionId }, () => {
clearTimeout(archiveTimer);
archiveTimer = undefined;
request('thread/resume', { ...shared, threadId: resumeSessionId }, onThread);
}, undefined, (unarchiveError) => {
onAgentEvent({ type: 'diagnostic', name: 'codex_thread_unarchive_failed',
message: rpcErrorText('thread/unarchive', unarchiveError) });
// Do not feed a cleanup error's "no rollout" wording into the
// daemon's stale-resume detector, which would clear the cursor.
reportFatal('codex app-server thread/unarchive failed');
shutdown();
});
},
);
return;
}
request('thread/start', shared, onThread);
request('thread/start', supportsPaginatedHistory ? { ...shared, historyMode: 'paginated' } : shared, onThread);
}
/** Emit the thread id on the session-capture channel, at most once. */
@@ -269,6 +327,8 @@ export function attachCodexAppServerSession(
}
function handleFrame(frame: JsonObject): void {
if (turnEnded || childClosed) return;
if (aborted && !terminalReceived && frame.method !== 'turn/completed') return;
if (!cliReadySeen) {
cliReadySeen = true;
onCliReady?.();
@@ -303,13 +363,25 @@ export function attachCodexAppServerSession(
reportSessionHandle(startedId);
return;
}
if (frame.method === 'turn/completed' && terminalReceived) return;
normalizer.handleNotification(frame.method, frame.params);
if (frame.method === 'turn/completed') {
terminalReceived = true;
const params = isRecord(frame.params) ? frame.params : {};
const turn = isRecord(params.turn) ? params.turn : null;
turnSucceeded = turn?.status !== 'failed';
onTurnComplete?.();
shutdown();
if (canArchiveSafely && ownsThread && threadId
&& (!resumeSessionId || threadId === resumeSessionId)) {
// The upstream archive operation checks its cross-process writer lock.
// A competing resume therefore refuses cleanup instead of losing history.
archiveTimer = setTimeout(shutdown, 1_500);
request('thread/archive', { threadId }, shutdown, undefined, (error) => {
onAgentEvent({ type: 'diagnostic', name: 'codex_thread_archive_failed',
message: rpcErrorText('thread/archive', error) });
shutdown();
});
} else shutdown();
}
}
@@ -327,6 +399,7 @@ export function attachCodexAppServerSession(
function shutdown(): void {
if (turnEnded) return;
turnEnded = true;
if (archiveTimer) clearTimeout(archiveTimer);
const stdin = child.stdin;
if (!stdin || stdin.destroyed) return;
try {
@@ -356,6 +429,12 @@ export function attachCodexAppServerSession(
}
});
child.once?.('close', () => {
childClosed = true;
if (archiveTimer) clearTimeout(archiveTimer);
pending.clear();
});
request(
'initialize',
{
@@ -364,9 +443,16 @@ export function attachCodexAppServerSession(
title: 'Open Design',
version: opts.clientVersion ?? '0.0.0',
},
capabilities: { experimentalApi: false, requestAttestation: false },
capabilities: { experimentalApi: opts.manageThreadVisibility === true, requestAttestation: false },
},
() => {
(result) => {
// Verified with the official 0.146.0 and 0.153.4 binaries: paginated
// history has cross-process archive locks in both; legacy only in the
// latter. Never convert an existing legacy history or archive it on an
// older server. Unknown servers retain the existing resumable behavior.
const capabilities = codexHistoryCapabilities(result.userAgent);
supportsPaginatedHistory = opts.manageThreadVisibility === true && capabilities.paginated;
protectsLegacyHistory = opts.manageThreadVisibility === true && capabilities.legacy;
notify('initialized', {});
openThread();
},
@@ -390,6 +476,12 @@ export function attachCodexAppServerSession(
getLastSessionPath(): string | null {
return rolloutPath;
},
takeClosedThreadCleanup(): CodexClosedThreadCleanup | null {
if (!childClosed || terminalReceived) return null;
const receipt = cleanupThread;
cleanupThread = null;
return receipt;
},
stats() {
return normalizer.stats();
},
@@ -0,0 +1,249 @@
import { spawn } from 'node:child_process';
import { captureProcessSnapshot, collectProcessTreePids, isProcessAlive, selectOwnedProcessTree,
signalProcesses, stopProcesses, type ProcessSnapshot } from '@open-design/platform';
import type { CodexClosedThreadCleanup } from './session.js';
/** Same measured compatibility floor used by the live session's archive path. */
export function codexHistoryCapabilities(userAgent: unknown): { paginated: boolean; legacy: boolean } {
const version = typeof userAgent === 'string'
? /^[^/\s]+\/(\d+)\.(\d+)\.(\d+)(?:\s|$)/u.exec(userAgent) : null;
if (!version) return { paginated: false, legacy: false };
const [, major, minor, patch] = version;
return {
paginated: Number(major) > 0 || Number(minor) >= 146,
legacy: Number(major) > 0 || Number(minor) > 153
|| (Number(minor) === 153 && Number(patch) >= 4),
};
}
export interface CodexThreadCleanupOptions {
receipt: CodexClosedThreadCleanup | null;
/** The original resolved app-server invocation, including platform wrapper arguments. */
command: string;
args: readonly string[];
cwd: string;
env: NodeJS.ProcessEnv;
windowsVerbatimArguments?: boolean;
}
export type CodexThreadCleanupResult =
// RPC success only. false means late/unobserved descendants are not proven gone.
| { status: 'archived'; treeVerified?: false }
| { status: 'skipped'; reason: 'no-owned-closed-thread' | 'unverified-writer-lock' }
| { status: 'failed'; message: string; remainingPids?: number[]; treeVerified?: boolean };
const CLEANUP_BUDGET_MS = 1_500;
/**
* Best-effort cleanup after the original writer has closed. This process never
* loads/resumes a thread: archive itself arbitrates any other active writer.
* The caller reports failures separately and must not delay run finalization.
*/
export async function cleanupClosedCodexThread(opts: CodexThreadCleanupOptions): Promise<CodexThreadCleanupResult> {
const receipt = opts.receipt;
if (!receipt) return { status: 'skipped', reason: 'no-owned-closed-thread' };
return new Promise((resolve) => {
let child: ReturnType<typeof spawn>;
try {
child = spawn(opts.command, [...opts.args], {
cwd: opts.cwd, env: opts.env, stdio: ['pipe', 'pipe', 'pipe'], shell: false,
detached: process.platform !== 'win32',
windowsVerbatimArguments: opts.windowsVerbatimArguments,
});
} catch (error) {
resolve({ status: 'failed', message: String(error) });
return;
}
let result: CodexThreadCleanupResult | undefined;
let buffer = '';
let awaiting: 'initialize' | 'ownership' | 'archive' | null = 'initialize';
let knownProcesses: ProcessSnapshot[] = [];
let terminating = false;
let settled = false;
let closed = false;
let closeObserved!: () => void;
const closePromise = new Promise<void>(done => { closeObserved = done; });
function settle(): void {
if (settled) return;
settled = true;
clearTimeout(timer);
resolve(result ?? { status: 'failed', message: 'codex thread cleanup exited before acknowledgement' });
}
const originalAlive = () => child.exitCode === null && child.signalCode === null;
const timer = setTimeout(() => {
result = { status: 'failed', message: 'codex thread cleanup timed out' };
awaiting = null;
if (process.platform === 'win32') {
void terminateWindowsTree();
} else void terminatePosixGroup();
}, CLEANUP_BUDGET_MS);
async function terminatePosixGroup(): Promise<void> {
if (terminating) return;
terminating = true;
try {
// Only this spawned group is proven owned. A detached descendant can
// retain our pipes without belonging to that group.
if (child.pid) signalProcesses([-child.pid], 'SIGKILL');
else child.kill('SIGKILL');
} catch (error) {
result = { status: 'failed', message: `codex thread cleanup timed out; ${String(error)}` };
}
let closeTimer: ReturnType<typeof setTimeout> | undefined;
await Promise.race([closePromise, new Promise<void>(done => { closeTimer = setTimeout(done, 200); })]);
clearTimeout(closeTimer);
if (!closed) {
child.stdin?.destroy(); child.stdout?.destroy(); child.stderr?.destroy();
result = { status: 'failed',
message: 'codex thread cleanup timed out; inherited stdio remained open; descendant ownership is unverified',
treeVerified: false,
remainingPids: originalAlive() && child.pid && isProcessAlive(child.pid) ? [child.pid] : [] };
}
settle();
}
async function captureOwnedTree(): Promise<ProcessSnapshot[]> {
const snapshot = await captureProcessSnapshot({ timeoutMs: 500 });
// The ChildProcess exit state fences the root identity across the async
// query. A PID from a later snapshot alone is never ownership proof.
if (knownProcesses.length === 0 && originalAlive()) {
const root = snapshot.find(entry => entry.pid === child.pid
&& Number.isSafeInteger(entry.startedAtMs) && entry.startedAtMs! > 0);
if (root) knownProcesses = [root];
}
const selected = selectOwnedProcessTree(knownProcesses, snapshot);
const structural = collectProcessTreePids(snapshot, selected.map(entry => entry.pid));
if (structural.some(pid => !selected.some(entry => entry.pid === pid))) {
throw new Error('codex cleanup descendant generation is unverified');
}
if (!knownProcesses.length) throw new Error('codex cleanup process ownership is unverified');
knownProcesses = selected;
return selected;
}
async function terminateWindowsTree(archiveAcknowledged = false): Promise<void> {
if (terminating) return;
terminating = true;
// The RPC and owned wrapper have already completed. Remaining work has
// its own bounded snapshot/stop calls; do not replace the acknowledged
// archive with a late protocol timer.
if (archiveAcknowledged) clearTimeout(timer);
let treeVerified = false;
let failure = '';
let remainingPids: number[] = [];
try {
const owned = await captureOwnedTree();
treeVerified = true;
const stopped = await stopProcesses(owned.map(entry => entry.pid), { termGraceMs: 0, killGraceMs: 200 });
remainingPids = stopped.remainingPids;
} catch (error) {
failure = `; ${String(error)}`;
remainingPids = knownProcesses.map(entry => entry.pid).filter(isProcessAlive);
} finally {
// This actual child is owned even if enumeration failed. Do not let a
// missing snapshot turn into a leaked wrapper or an endless close wait.
if (originalAlive()) {
try { child.kill('SIGKILL'); } catch (error) { failure += `; ${String(error)}`; }
}
let closeTimer: ReturnType<typeof setTimeout> | undefined;
await Promise.race([closePromise, new Promise<void>(done => { closeTimer = setTimeout(done, 200); })]);
clearTimeout(closeTimer);
if (!closed) {
child.stdin?.destroy(); child.stdout?.destroy(); child.stderr?.destroy();
treeVerified = false;
}
if (originalAlive() && child.pid) remainingPids.push(child.pid);
remainingPids = [...new Set(remainingPids)].filter(isProcessAlive);
if (archiveAcknowledged && !failure && remainingPids.length === 0) {
// Reaping the previously observed set cannot prove that the exited
// wrapper created no later, detached stdio-ignore descendants.
result = { status: 'archived', treeVerified: false };
} else {
const message = archiveAcknowledged
? 'codex thread archive acknowledged; process cleanup failed'
: 'codex thread cleanup timed out';
result = { status: 'failed', message: `${message}${failure}`, remainingPids, treeVerified };
}
settle();
}
}
function finish(value: CodexThreadCleanupResult): void {
if (result) return;
result = value;
awaiting = null;
child.stdin?.end();
}
function write(frame: Record<string, unknown>): void {
try {
child.stdin?.write(`${JSON.stringify(frame)}\n`, (error) => {
if (error) finish({ status: 'failed', message: error.message });
});
} catch (error) { finish({ status: 'failed', message: String(error) }); }
}
child.on('error', (error) => finish({ status: 'failed', message: error.message }));
child.stdin?.on('error', (error) => finish({ status: 'failed', message: error.message }));
// Drain stderr so a chatty CLI cannot fill its pipe; do not expose its environment.
child.stderr?.resume();
child.once('close', (code, signal) => {
closed = true;
closeObserved();
result ??= { status: 'failed', message: `codex thread cleanup exited before acknowledgement (${code ?? signal})` };
if (!terminating && process.platform === 'win32' && result.status === 'archived') {
void terminateWindowsTree(true);
} else if (!terminating) settle();
});
child.stdout?.on('data', (chunk: Buffer) => {
if (!awaiting) return;
buffer += chunk.toString();
if (buffer.length > 1_048_576) {
finish({ status: 'failed', message: 'codex thread cleanup response exceeded budget' });
return;
}
let newline: number;
while ((newline = buffer.indexOf('\n')) !== -1) {
const line = buffer.slice(0, newline);
buffer = buffer.slice(newline + 1);
let frame: Record<string, unknown>;
try {
const value: unknown = JSON.parse(line);
if (!value || typeof value !== 'object' || Array.isArray(value)) continue;
frame = value as Record<string, unknown>;
} catch { continue; }
if (!awaiting || awaiting === 'ownership' || frame.id !== (awaiting === 'initialize' ? 1 : 2)) continue;
if (frame.error !== undefined) {
finish({ status: 'failed', message: `codex thread cleanup ${awaiting}: ${JSON.stringify(frame.error)}` });
continue;
}
if (!frame.result || typeof frame.result !== 'object' || Array.isArray(frame.result)) {
finish({ status: 'failed', message: `codex thread cleanup ${awaiting}: invalid response` });
continue;
}
if (awaiting === 'initialize') {
const capabilities = codexHistoryCapabilities((frame.result as Record<string, unknown>).userAgent);
if (!capabilities[receipt.historyMode]) {
finish({ status: 'skipped', reason: 'unverified-writer-lock' });
continue;
}
const archive = () => {
if (!awaiting || terminating || settled) return;
write({ jsonrpc: '2.0', method: 'initialized', params: {} });
awaiting = 'archive';
write({ jsonrpc: '2.0', id: 2, method: 'thread/archive', params: { threadId: receipt.threadId } });
};
if (process.platform === 'win32') {
awaiting = 'ownership';
void captureOwnedTree().then(archive, error => {
finish({ status: 'failed', message: String(error), treeVerified: false });
});
} else archive();
} else finish({ status: 'archived' });
}
});
write({ jsonrpc: '2.0', id: 1, method: 'initialize', params: {
clientInfo: { name: 'open-design', title: 'Open Design', version: '0.0.0' },
capabilities: { experimentalApi: true, requestAttestation: false },
} });
});
}
@@ -0,0 +1,55 @@
import { lstat, realpath } from 'node:fs/promises';
import type { Stats } from 'node:fs';
import path from 'node:path';
import Database from 'better-sqlite3';
function isOwnedAndPrivate(stat: Stats): boolean {
return !stat.isSymbolicLink()
&& (stat.mode & 0o022) === 0
&& (typeof process.getuid !== 'function' || stat.uid === process.getuid());
}
/**
* Codex 0.146+ keeps archived rollouts in a flat directory. Its state_5.sqlite
* primary-key index resolves only the known thread, independently of archive
* size. The index is optional: callers retain their bounded directory fallback
* for older, missing, busy, or incompatible native databases.
*/
export async function findIndexedCodexArchivedRolloutPath(
codexHome: string,
sessionId: string,
): Promise<string | null> {
if (!path.isAbsolute(codexHome)
|| !/^[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}$/iu.test(sessionId)) return null;
const archiveRoot = path.join(codexHome, 'archived_sessions');
const dbPath = path.join(codexHome, 'state_5.sqlite');
const [rootStat, dbStat] = await Promise.all([
lstat(archiveRoot).catch(() => null),
lstat(dbPath).catch(() => null),
]);
if (!rootStat?.isDirectory() || !isOwnedAndPrivate(rootStat)
|| !dbStat?.isFile() || !isOwnedAndPrivate(dbStat)) return null;
let db: Database.Database | undefined;
try {
db = new Database(dbPath, { readonly: true, fileMustExist: true, timeout: 25 });
// Requiring the native primary-key index also prevents an incompatible
// schema from turning this lookup into a scan of unrelated thread rows.
const row = db.prepare(`SELECT rollout_path FROM threads
INDEXED BY sqlite_autoindex_threads_1 WHERE id = ? AND archived = 1 LIMIT 1`)
.get(sessionId) as { rollout_path?: unknown } | undefined;
const filePath = row?.rollout_path;
if (typeof filePath !== 'string' || !path.isAbsolute(filePath)) return null;
const name = path.basename(filePath);
if (!name.startsWith('rollout-') || !name.endsWith(`-${sessionId}.jsonl`)) return null;
const archiveRealPath = await realpath(archiveRoot);
if (await realpath(path.dirname(filePath)) !== archiveRealPath) return null;
const fileStat = await lstat(filePath);
if (!fileStat.isFile() || !isOwnedAndPrivate(fileStat)) return null;
if (path.dirname(await realpath(filePath)) !== archiveRealPath) return null;
return path.join(archiveRoot, name);
} catch {
return null;
} finally {
db?.close();
}
}
+15 -2
View File
@@ -12,9 +12,11 @@
// at the call site (it needs CODEX_HOME + the captured session id).
import os from 'node:os';
import { readFile, readdir } from 'node:fs/promises';
import { lstat, readFile, readdir } from 'node:fs/promises';
import path from 'node:path';
import { findIndexedCodexArchivedRolloutPath } from './codex-archived-rollout-index.js';
interface CodexFirstCallUsage {
first_call_input_tokens: number;
first_call_cache_read_input_tokens: number;
@@ -168,7 +170,18 @@ export async function readCodexRolloutFirstCall(opts: {
const sessionId = opts.sessionId?.trim();
if (!sessionId) return null;
try {
const rolloutPath = await findCodexRolloutPath(codexHome, sessionId);
let rolloutPath = await findCodexRolloutPath(codexHome, sessionId);
if (!rolloutPath) rolloutPath = await findIndexedCodexArchivedRolloutPath(codexHome, sessionId);
if (!rolloutPath) {
const archiveRoot = path.join(codexHome, 'archived_sessions');
const stat = await lstat(archiveRoot).catch(() => null);
if (!stat?.isDirectory() || stat.isSymbolicLink()) return null;
const entries = await readdir(archiveRoot, { withFileTypes: true }).catch(() => []);
if (entries.length > 2_048) return null;
const matches = entries.filter((entry) => entry.isFile() && !entry.isSymbolicLink()
&& rolloutFileMatchesSession(entry.name, sessionId));
if (matches.length === 1) rolloutPath = path.join(archiveRoot, matches[0]!.name);
}
if (!rolloutPath) return null;
const contents = await readFile(rolloutPath, 'utf8');
return extractCodexLastTurnFirstCallUsage(contents);
@@ -12,6 +12,8 @@ import {
} from 'node:fs/promises';
import path from 'node:path';
import { findIndexedCodexArchivedRolloutPath } from '../codex-archived-rollout-index.js';
import {
normalizeAgentObservationV1,
type NormalizedAgentObservationV1,
@@ -452,14 +454,28 @@ async function resolveRolloutPath(input: {
}
const sessionsRoot = path.join(codexHome, 'sessions');
const rootEntries = await safeDirectoryEntries(sessionsRoot, input.maxDirectoryEntries);
if (!rootEntries) return { ok: false, reason: 'sessions_root_unavailable' };
const rootRealPath = await realpath(sessionsRoot).catch(() => null);
if (!rootRealPath) return { ok: false, reason: 'sessions_root_unavailable' };
const matches: string[] = [];
const rootRealPath = rootEntries ? await realpath(sessionsRoot).catch(() => null) : null;
// Archiving moves only the explicitly owned thread and its descendants.
// Match the requested id under either native storage root; never inspect
// unrelated transcript contents or relax the existing file safety checks.
const archiveRoot = path.join(codexHome, 'archived_sessions');
const indexedArchivePath = await findIndexedCodexArchivedRolloutPath(codexHome, input.sessionId);
const archiveEntries = indexedArchivePath
? null
: await safeDirectoryEntries(archiveRoot, input.maxDirectoryEntries);
const archiveRealPath = indexedArchivePath || archiveEntries
? await realpath(archiveRoot).catch(() => null)
: null;
if (!rootRealPath && !archiveRealPath) return { ok: false, reason: 'sessions_root_unavailable' };
const suffix = `-${input.sessionId}.jsonl`;
const matches: string[] = (archiveEntries ?? [])
.filter((entry) => entry.isFile() && !entry.isSymbolicLink()
&& entry.name.startsWith('rollout-') && entry.name.endsWith(suffix))
.map((entry) => path.join(archiveRoot, entry.name));
if (indexedArchivePath) matches.push(indexedArchivePath);
let scannedDays = 0;
let exhausted = false;
const years = rootEntries
const years = (rootEntries ?? [])
.filter((entry) => entry.isDirectory() && !entry.isSymbolicLink() && YEAR_PATTERN.test(entry.name))
.sort((left, right) => codePointCompare(right.name, left.name));
for (const year of years) {
@@ -513,7 +529,8 @@ async function resolveRolloutPath(input: {
if (!stat || isUnsafeFileStat(stat)) return { ok: false, reason: 'unsafe_rollout_file' };
if (stat.size > input.maxRolloutBytes) return { ok: false, reason: 'rollout_too_large' };
const resolved = await realpath(filePath).catch(() => null);
if (!resolved || !resolved.startsWith(`${rootRealPath}${path.sep}`)) {
const expectedRoot = path.dirname(filePath) === archiveRoot ? archiveRealPath : rootRealPath;
if (!resolved || !expectedRoot || !resolved.startsWith(`${expectedRoot}${path.sep}`)) {
return { ok: false, reason: 'unsafe_rollout_file' };
}
return { ok: true, filePath, stat };
+43 -2
View File
@@ -504,6 +504,7 @@ import {
codexResolvedSandboxMode,
} from './runtimes/defs/codex.js';
import { attachCodexAppServerSession } from './agent-protocol/codex-app-server/session.js';
import { createCodexThreadCleanupOwner, type CodexCleanupInvocation } from './agent-protocol/codex-app-server/cleanup-owner.js';
import {
ensureDetectedRuntimeVersions,
getDetectedRuntimeVersions,
@@ -7760,6 +7761,7 @@ export async function startServer({
readRunTelemetrySinkConfig(process.env, configuredAmrEnv()),
),
);
const codexThreadCleanupOwner = createCodexThreadCleanupOwner();
const design = {
runs: createChatRunService({
createSseResponse,
@@ -14121,6 +14123,11 @@ export async function startServer({
let acpSession = null;
let writePromptToChildStdin = false;
let spawnedAgentEnv = null;
let codexCleanupInvocation: CodexCleanupInvocation | null = null;
let completeCodexEvidenceCollection = () => {};
const codexEvidenceCollected = def.id === 'codex'
? new Promise<void>((resolve) => { completeCodexEvidenceCollection = resolve; })
: undefined;
// The stream handler is block-scoped to its parser branch, but the OpenCode
// post-run child export runs in the shared close handler below — after the
// stream that produced the candidates is gone.
@@ -14236,6 +14243,13 @@ export async function startServer({
args,
env,
});
if (def.streamFormat === CODEX_APP_SERVER_STREAM_FORMAT) {
codexCleanupInvocation = {
command: invocation.command, args: [...invocation.args], env: { ...env }, cwd: effectiveCwd,
...(invocation.windowsVerbatimArguments !== undefined
? { windowsVerbatimArguments: invocation.windowsVerbatimArguments } : {}),
};
}
lifecycle.mark('launch_preflight_end');
lifecycle.mark('process_spawn_start');
child = spawn(invocation.command, invocation.args, {
@@ -15700,7 +15714,7 @@ export async function startServer({
// always did. The transport's own additions are token-level text and
// reasoning deltas.
trackingSubstantiveOutput = true;
acpSession = attachCodexAppServerSession({
const codexSession = attachCodexAppServerSession({
child,
prompt: composed,
cwd: effectiveCwd,
@@ -15708,6 +15722,10 @@ export async function startServer({
reasoning: safeReasoning,
serviceTier: safeServiceTier,
sandboxMode: codexResolvedSandboxMode(),
manageThreadVisibility: true,
// This handle came from our captured agent_sessions record (or a
// same-run daemon continuation), never from the public chat payload.
resumeSessionOwned: agentResumeCtx.isResuming,
imagePaths: def.supportsImagePaths ? amrStagedImages : [],
clientVersion: design.getAppVersion?.() ?? '0.0.0',
// Capture-style resume, same contract as `exec resume <thread_id>`:
@@ -15730,6 +15748,18 @@ export async function startServer({
onPromptSendEnd: () => lifecycle.mark('stdin_write_end'),
onTurnComplete: () => clearFirstOutputWatchdog(),
});
acpSession = codexSession;
if (codexCleanupInvocation) {
const cleanupRunId = run.id;
// Attach installs the session's physical-close proof first. This
// listener is independent of the later retry-generation close guard.
codexThreadCleanupOwner.bind(child, codexSession, codexCleanupInvocation, (result) => {
const details = { runId: cleanupRunId, ...result };
if (result.status === 'archived' && result.treeVerified !== false) {
console.info('[codex] closed-thread cleanup completed', details);
} else console.warn('[codex] closed-thread cleanup incomplete', details);
}, codexEvidenceCollected);
}
} else if (def.streamFormat === 'json-event-stream') {
// Pipe through sendAgentEvent so the OpenCode `type:'error'` frame
// (now emitted as a real error event by json-event-stream.ts after
@@ -15946,6 +15976,7 @@ export async function startServer({
}
}
}
completeCodexEvidenceCollection();
// Native OpenCode filters child-session events out of the root JSON
// stream, so the live stream can only produce a terminal-only L1
// candidate — which fails the evidence graph on `child_started_missing`
@@ -16946,6 +16977,9 @@ export async function startServer({
);
}
} finally {
// Superseded attempts and early/error exits also release their own
// receipt. The cleanup owner separately fences the shutdown deadline.
completeCodexEvidenceCollection();
// Best-effort cleanup of the per-run agy log file on every close
// path — successful, failed, cancelled, or non-zero exit — so
// /tmp doesn't accumulate one file per Antigravity run. The log
@@ -17790,7 +17824,14 @@ export async function startServer({
daemonShuttingDown = true;
amrTerminalReportDelivery.stop();
clearTerminalTelemetryFallbackTimers();
await design.runs.shutdownActive({ graceMs: resolveChatRunShutdownGraceMs() });
const shutdownGraceMs = resolveChatRunShutdownGraceMs();
await design.runs.shutdownActive({ graceMs: shutdownGraceMs });
// Cleanup runs independently of user terminal classification. Give
// confirmed closed writers at most three additional seconds at shutdown.
const cleanupDrain = await codexThreadCleanupOwner.drain(shutdownGraceMs);
if (cleanupDrain.pending > 0) {
console.warn('[codex] closed-thread cleanup shutdown deadline reached', cleanupDrain);
}
await terminalService.shutdownActive();
await browserSessionService.shutdownActive();
await design.analytics.shutdown();
@@ -41,6 +41,8 @@ const CLIENT_REQUESTS = [
'initialize',
'thread/start',
'thread/resume',
'thread/archive',
'thread/unarchive',
'turn/start',
'turn/interrupt',
] as const;
@@ -66,15 +68,18 @@ const SERVER_NOTIFICATIONS = [
const TYPE_SURFACE: Record<string, string[]> = {
'InitializeCapabilities.ts': ['experimentalApi', 'requestAttestation'],
'ClientInfo.ts': ['name', 'title', 'version'],
'v2/ThreadStartParams.ts': ['cwd', 'sandbox', 'approvalPolicy'],
'v2/ThreadStartParams.ts': ['cwd', 'sandbox', 'approvalPolicy', 'historyMode'],
'v2/ThreadResumeParams.ts': ['threadId', 'cwd', 'sandbox', 'approvalPolicy'],
'v2/ThreadArchiveParams.ts': ['threadId'],
'v2/ThreadUnarchiveParams.ts': ['threadId'],
'v2/TurnStartParams.ts': ['threadId', 'input', 'model', 'effort', 'summary', 'serviceTier'],
'v2/TurnInterruptParams.ts': ['threadId'],
'v2/SandboxMode.ts': ['"workspace-write"', '"danger-full-access"'],
'ReasoningSummary.ts': ['"detailed"'],
'v2/UserInput.ts': ['"text"', 'text_elements', '"localImage"', 'path'],
'v2/ThreadStartedNotification.ts': ['thread'],
'v2/Thread.ts': ['id', 'path'],
'v2/Thread.ts': ['id', 'path', 'historyMode'],
'v2/ThreadHistoryMode.ts': ['"legacy"', '"paginated"'],
'v2/TurnCompletedNotification.ts': ['turn'],
'v2/Turn.ts': ['id', 'status', 'error'],
'v2/TurnStatus.ts': ['"completed"', '"failed"', '"interrupted"'],
@@ -132,7 +137,7 @@ describeWithCodex('codex app-server protocol contract', () => {
codexVersion = String(
execFileSync('codex', ['--version'], { encoding: 'utf8', timeout: 10_000 }),
).trim();
execFileSync('codex', ['app-server', 'generate-ts', '--out', outDir], {
execFileSync('codex', ['app-server', 'generate-ts', '--experimental', '--out', outDir], {
stdio: 'ignore',
timeout: 120_000,
});
@@ -516,3 +516,222 @@ describe('codex app-server session', () => {
});
});
});
describe('OD-owned Codex thread visibility', () => {
function ready(overrides: Record<string, unknown> = {}, version = '0.153.4') {
const h = harness({ manageThreadVisibility: true, ...overrides });
const init = h.child.sent('initialize')!;
h.child.say({ id: init.id, result: { userAgent: `open-design/${version}` } });
const open = h.child.sent('thread/start') ?? h.child.sent('thread/resume');
return { ...h, open: open! };
}
function opened(h: ReturnType<typeof ready>) {
h.child.say({ id: h.open.id, result: { thread: { id: 'owned-1' } } });
}
function completed(child: FakeChild, status = 'completed') {
child.say({ method: 'turn/completed', params: { threadId: 'owned-1', turn: { status } } });
}
it.each(['completed', 'failed', 'interrupted'])('archives only its owned thread after %s', (status) => {
const h = ready();
opened(h);
if (status === 'interrupted') h.session.abort();
completed(h.child, status);
expect(h.child.sent('thread/archive')?.params).toEqual({ threadId: 'owned-1' });
expect(h.child.stdinEnded).toBe(0);
h.child.say({ id: h.child.sent('thread/archive')!.id, result: {} });
expect(h.child.stdinEnded).toBe(1);
expect(h.session.completedSuccessfully()).toBe(status !== 'failed');
expect(h.session.getDurableSessionId()).toBe('owned-1');
});
it('does not overwrite a successful result when another process owns the writer lock', () => {
const h = ready(); opened(h); completed(h.child);
h.child.say({ id: h.child.sent('thread/archive')!.id,
error: { code: -32600, message: 'thread owned-1 already has an active writer' } });
expect(h.session.completedSuccessfully()).toBe(true);
expect(h.child.stdinEnded).toBe(1);
expect(h.agentEvents.filter(e => e.type === 'error')).toEqual([]);
expect(h.agentEvents).toContainEqual(expect.objectContaining({ name: 'codex_thread_archive_failed' }));
});
it('keeps unknown resume handles outside its archive/unarchive policy', () => {
const h = ready({ resumeSessionId: 'owned-1' }); opened(h); completed(h.child);
expect(h.child.sent('thread/archive')).toBeUndefined();
expect(h.child.sent('thread/unarchive')).toBeUndefined();
expect(h.child.stdinEnded).toBe(1);
});
it('restores a positively owned archived handle before continuing the same thread', () => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true });
h.child.say({ id: h.open.id, error: { code: -32600, message: 'session owned-1 is archived' } });
expect(h.child.sent('turn/start')).toBeUndefined();
const unarchive = h.child.sent('thread/unarchive')!;
expect(unarchive.params).toEqual({ threadId: 'owned-1' });
h.child.say({ id: unarchive.id, result: { thread: { id: 'owned-1' } } });
const resumes = h.child.frames().filter(f => f.method === 'thread/resume');
expect(resumes).toHaveLength(2);
h.child.say({ id: resumes[1]!.id, result: { thread: { id: 'owned-1' } } });
expect(h.child.sent('turn/start')?.params.threadId).toBe('owned-1');
completed(h.child);
h.child.say({ id: h.child.sent('thread/archive')!.id, result: {} });
expect(h.session.getDurableSessionId()).toBe('owned-1');
});
it('can resume an already archived owned thread after a CLI downgrade', () => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true }, '0.146.0');
h.child.say({ id: h.open.id, error: { message: 'session owned-1 is archived' } });
const restore = h.child.sent('thread/unarchive')!;
h.child.say({ id: restore.id, result: {} });
const resume = h.child.frames().filter(f => f.method === 'thread/resume').at(-1)!;
h.child.say({ id: resume.id, result: { thread: { id: 'owned-1' } } });
expect(h.child.sent('turn/start')?.params.threadId).toBe('owned-1');
completed(h.child);
expect(h.child.sent('thread/archive')).toBeUndefined();
});
it('does not archive a different id returned by an owned resume request', () => {
const h = ready({ resumeSessionId: 'previous-id', resumeSessionOwned: true });
opened(h); completed(h.child);
expect(h.child.sent('thread/archive')).toBeUndefined();
});
it('keeps unarchive failure retryable without forwarding a stale-session signal', () => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true });
h.child.say({ id: h.open.id, error: { message: 'session owned-1 is archived' } });
h.child.say({ id: h.child.sent('thread/unarchive')!.id,
error: { message: 'no rollout found while restoring archived session' } });
expect(h.child.sent('turn/start')).toBeUndefined();
expect(h.child.sent('thread/start')).toBeUndefined();
expect(h.child.stdinEnded).toBe(1);
expect(h.agentEvents.filter(e => e.type === 'error')).toEqual([
{ type: 'error', message: 'codex app-server thread/unarchive failed' },
]);
});
it('archives once even when a completion notification is duplicated', () => {
const h = ready(); opened(h); completed(h.child); completed(h.child);
expect(h.child.frames().filter(f => f.method === 'thread/archive')).toHaveLength(1);
h.child.say({ id: h.child.sent('thread/archive')!.id, result: {} });
completed(h.child);
expect(h.child.stdinEnded).toBe(1);
});
it('bounds archive cleanup without changing the result', () => {
vi.useFakeTimers();
try {
const h = ready(); opened(h); completed(h.child);
vi.advanceTimersByTime(1499);
expect(h.child.stdinEnded).toBe(0);
vi.advanceTimersByTime(1);
expect(h.child.stdinEnded).toBe(1);
expect(h.session.completedSuccessfully()).toBe(true);
} finally { vi.useRealTimers(); }
});
it('bounds unarchive and ignores its late response after cancellation', () => {
vi.useFakeTimers();
try {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true });
h.child.say({ id: h.open.id, error: { message: 'session owned-1 is archived' } });
const restore = h.child.sent('thread/unarchive')!;
h.session.abort();
vi.advanceTimersByTime(1500);
h.child.say({ id: restore.id, result: {} });
expect(h.child.sent('turn/start')).toBeUndefined();
expect(h.child.stdinEnded).toBe(1);
} finally { vi.useRealTimers(); }
});
it.each(['completed', 'failed', 'interrupted'])('uses protected persistent history on the reported 0.146.0 version after %s', (status) => {
const h = ready({}, '0.146.0');
expect(h.child.sent('initialize')?.params.capabilities.experimentalApi).toBe(true);
expect(h.open.params.historyMode).toBe('paginated');
h.child.say({ id: h.open.id, result: { thread: { id: 'owned-1', historyMode: 'paginated' } } });
h.child.say({ method: 'item/agentMessage/delta', params: { itemId: 'm1', delta: 'visible progress' } });
expect(h.agentEvents).toContainEqual({ type: 'text_delta', delta: 'visible progress' });
if (status === 'interrupted') h.session.abort();
completed(h.child, status);
expect(h.child.sent('thread/archive')?.params).toEqual({ threadId: 'owned-1' });
h.child.say({ id: h.child.sent('thread/archive')!.id, result: {} });
expect(h.child.stdinEnded).toBe(1);
expect(h.session.completedSuccessfully()).toBe(status !== 'failed');
});
it.each([0, 1])('offers an owned interrupted thread for cleanup only after actual close (exit %s)', (code) => {
const h = ready(); opened(h);
// Optional lookup keeps the baseline executable: no receipt means the
// close owner cannot initiate cleanup, rather than a missing-module error.
const takeReceipt = () => (h.session as typeof h.session & {
takeClosedThreadCleanup?: () => { threadId: string; historyMode: string } | null;
}).takeClosedThreadCleanup?.() ?? null;
expect(takeReceipt()).toBeNull();
h.session.abort();
h.child.killed = true;
expect(takeReceipt()).toBeNull();
h.child.emit('close', code, null);
expect(takeReceipt()).toMatchObject({ threadId: 'owned-1', historyMode: 'legacy' });
expect(takeReceipt()).toBeNull();
expect(h.child.sent('thread/archive')).toBeUndefined();
expect(h.session.completedSuccessfully()).toBe(false);
});
it.each([
{ label: 'user resume', options: { resumeSessionId: 'owned-1' }, version: '0.153.4' },
{ label: 'mismatched owned resume', options: { resumeSessionId: 'other', resumeSessionOwned: true }, version: '0.153.4' },
{ label: 'unmanaged', options: { manageThreadVisibility: false }, version: '0.153.4' },
{ label: 'unknown protocol', options: {}, version: 'unknown' },
{ label: 'unprotected legacy', options: {}, version: '0.146.0' },
])('does not offer cleanup for $label', ({ options, version }) => {
const h = ready(options, version); opened(h); h.child.emit('close', 1, null);
expect(h.session.takeClosedThreadCleanup()).toBeNull();
});
it('requires a matching request response, not a thread/started notification alone', () => {
const h = ready();
h.child.say({ method: 'thread/started', params: { thread: { id: 'notification-only', historyMode: 'paginated' } } });
h.child.emit('close', 1, null);
h.child.say({ id: h.open.id, result: { thread: { id: 'late-response', historyMode: 'paginated' } } });
expect(h.session.takeClosedThreadCleanup()).toBeNull();
});
it.each(['completed', 'failed', 'interrupted'])('leaves terminal %s cleanup with the existing live session', (status) => {
const h = ready(); opened(h); completed(h.child, status);
h.child.say({ id: h.child.sent('thread/archive')!.id, error: { message: 'active writer' } });
h.child.emit('close', 0, null);
expect(h.session.takeClosedThreadCleanup()).toBeNull();
expect(h.child.frames().filter(f => f.method === 'thread/archive')).toHaveLength(1);
});
it('retains paginated proof for an interrupted owned resume on 0.146.0', () => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true }, '0.146.0');
h.child.say({ id: h.open.id, result: { thread: { id: 'owned-1', historyMode: 'paginated' } } });
h.child.emit('close', null, 'SIGKILL');
expect(h.session.takeClosedThreadCleanup()).toMatchObject({ threadId: 'owned-1', historyMode: 'paginated' });
});
it('resumes and archives the same protected 0.146.0 history without converting it', () => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true }, '0.146.0');
expect(h.open.params.historyMode).toBeUndefined();
h.child.say({ id: h.open.id, result: { thread: { id: 'owned-1', historyMode: 'paginated' } } });
completed(h.child);
expect(h.child.sent('thread/archive')?.params).toEqual({ threadId: 'owned-1' });
h.child.say({ id: h.child.sent('thread/archive')!.id, result: {} });
expect(h.session.getDurableSessionId()).toBe('owned-1');
});
it.each(['legacy', undefined])('does not trust an unprotected 0.146.0 returned history mode: %s', (historyMode) => {
const h = ready({ resumeSessionId: 'owned-1', resumeSessionOwned: true }, '0.146.0');
h.child.say({ id: h.open.id, result: { thread: { id: 'owned-1', historyMode } } });
completed(h.child);
expect(h.child.sent('thread/archive')).toBeUndefined();
expect(h.session.getDurableSessionId()).toBe('owned-1');
expect(h.child.stdinEnded).toBe(1);
});
it.each(['0.145.0', 'unknown'])('retains existing history on unverified writer-lock version %s', (version) => {
const h = ready({}, version); opened(h); completed(h.child);
expect(h.child.sent('thread/archive')).toBeUndefined();
expect(h.session.completedSuccessfully()).toBe(true);
});
});
+68 -1
View File
@@ -1,7 +1,8 @@
import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs';
import { chmodSync, mkdtempSync, mkdirSync, renameSync, rmSync, symlinkSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { describe, expect, it } from 'vitest';
import Database from 'better-sqlite3';
import {
codexSessionIdFromRunEvents,
@@ -154,6 +155,72 @@ describe('readCodexRolloutFirstCall', () => {
expect(result?.first_call_cache_hit_ratio).toBeCloseTo(12672 / 13342);
});
it('preserves first-call usage after an owned thread has been archived', async () => {
const sid = '019eef4f-7409-7c82-bebe-30504eed3959';
const home = plantRollout(sid, rollout(sid));
try {
const before = await readCodexRolloutFirstCall({ codexHome: home, sessionId: sid });
const name = `rollout-2026-06-24T12-00-00-${sid}.jsonl`;
mkdirSync(path.join(home, 'archived_sessions'));
renameSync(path.join(home, 'sessions/2026/06/24', name), path.join(home, 'archived_sessions', name));
expect(await readCodexRolloutFirstCall({ codexHome: home, sessionId: sid })).toEqual(before);
} finally { rmSync(home, { recursive: true, force: true }); }
});
it('retains indexed archived usage after the flat archive exceeds the scan limit', async () => {
const sid = '019eef4f-7409-7c82-bebe-30504eed3959';
const home = plantRollout(sid, rollout(sid));
try {
const archive = path.join(home, 'archived_sessions');
mkdirSync(archive);
const name = `rollout-2026-06-24T12-00-00-${sid}.jsonl`;
const archivedPath = path.join(archive, name);
renameSync(path.join(home, 'sessions/2026/06/24', name), archivedPath);
// Codex archives into one flat directory, while its native primary-key
// index resolves a known thread without scanning other conversations.
for (let i = 0; i < 2_049; i += 1) writeFileSync(path.join(archive, `unrelated-${i}`), '');
const db = new Database(path.join(home, 'state_5.sqlite'));
try {
db.exec('CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, archived INTEGER NOT NULL)');
db.prepare('INSERT INTO threads VALUES (?, ?, 1)').run(sid, archivedPath);
} finally { db.close(); }
expect((await readCodexRolloutFirstCall({ codexHome: home, sessionId: sid }))?.first_call_input_tokens).toBe(13342);
} finally { rmSync(home, { recursive: true, force: true }); }
});
it.each(['outside archive', 'other thread', 'symlink', 'writable file'])(
'does not trust an indexed rollout pointing to %s',
async (scenario) => {
const sid = '019eef4f-7409-7c82-bebe-30504eed3959';
const home = plantRollout(sid, rollout(sid));
try {
const archive = path.join(home, 'archived_sessions');
mkdirSync(archive);
const name = `rollout-2026-06-24T12-00-00-${sid}.jsonl`;
const activePath = path.join(home, 'sessions/2026/06/24', name);
const target = path.join(home, name);
renameSync(activePath, target);
let indexedPath = path.join(archive, name);
if (scenario === 'outside archive') indexedPath = target;
else if (scenario === 'other thread') {
indexedPath = path.join(archive, 'rollout-2026-06-24T12-00-00-20000000-0000-4000-8000-000000000002.jsonl');
renameSync(target, indexedPath);
} else if (scenario === 'symlink') symlinkSync(target, indexedPath);
else {
renameSync(target, indexedPath);
chmodSync(indexedPath, 0o666);
}
for (let i = 0; i < 2_049; i += 1) writeFileSync(path.join(archive, `unrelated-${i}`), '');
const db = new Database(path.join(home, 'state_5.sqlite'));
try {
db.exec('CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, archived INTEGER NOT NULL)');
db.prepare('INSERT INTO threads VALUES (?, ?, 1)').run(sid, indexedPath);
} finally { db.close(); }
expect(await readCodexRolloutFirstCall({ codexHome: home, sessionId: sid })).toBeNull();
} finally { rmSync(home, { recursive: true, force: true }); }
},
);
it('returns null (never throws) when the session id has no matching rollout', async () => {
const home = plantRollout('some-other-session', rollout('some-other-session'));
await expect(
@@ -0,0 +1,375 @@
import { spawn } from 'node:child_process';
import { once } from 'node:events';
import { mkdir, mkdtemp, readFile, realpath, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { describe, expect, it, vi } from 'vitest';
import { attachCodexAppServerSession } from '../src/agent-protocol/codex-app-server/session.js';
import { createCodexThreadCleanupOwner } from '../src/agent-protocol/codex-app-server/cleanup-owner.js';
import { collectCodexChildEvidence, type CodexChildEvidenceCollection } from '../src/runtimes/codex-child-evidence.js';
import { createChatRunService } from '../src/runtimes/runs.js';
import type { CodexThreadCleanupResult } from '../src/agent-protocol/codex-app-server/thread-cleanup.js';
// Keep real filesystem/process operations; control only the interleaving that
// can move a selected active rollout before the collector opens it.
const collectorRace = vi.hoisted(() => ({
home: '', armed: false, helperStarted: false, activePath: '', archiveFinished: Promise.resolve(),
}));
vi.mock('node:child_process', async importOriginal => {
const actual = await importOriginal<typeof import('node:child_process')>();
return { ...actual, spawn: (...args: Parameters<typeof actual.spawn>) => {
if (collectorRace.armed && args[2]?.env?.RACE_HOME === collectorRace.home) {
collectorRace.helperStarted = true;
}
return actual.spawn(...args);
} };
});
vi.mock('node:fs/promises', async importOriginal => {
const actual = await importOriginal<typeof import('node:fs/promises')>();
return { ...actual, open: async (...args: Parameters<typeof actual.open>) => {
if (collectorRace.armed && args[0] === collectorRace.activePath && collectorRace.helperStarted) {
// The old close owner has already launched its archiver. Let that real
// child perform rename before opening the now-stale path.
await collectorRace.archiveFinished;
}
return actual.open(...args);
} };
});
const fixture = String.raw`
import { appendFileSync, existsSync, renameSync, watch } from 'node:fs';
import { createInterface } from 'node:readline';
const send = frame => process.stdout.write(JSON.stringify(frame) + '\n');
createInterface({ input: process.stdin }).on('line', line => {
const frame = JSON.parse(line);
appendFileSync(process.env.TRANSCRIPT, JSON.stringify({ pid: process.pid, marker: process.env.CASE_MARKER, cwd: process.cwd(), ...frame }) + '\n');
if (frame.method === 'initialize') send({ id: frame.id, result: { userAgent: 'codex/0.154.0' } });
else if (frame.method === 'thread/start') send({ id: frame.id, result: { thread: { id: process.env.THREAD_ID || 'owned-close-thread', historyMode: 'paginated' } } });
else if (frame.method === 'turn/start') {
if (process.env.MODE === 'normal') send({ method: 'turn/completed', params: { threadId: process.env.THREAD_ID || 'owned-close-thread', turn: { id: 'turn-1', status: 'completed', items: [] } } });
else if (process.env.MODE !== 'cancel') process.exit(1);
} else if (frame.method === 'turn/interrupt') process.exit(130);
else if (frame.method === 'thread/archive') {
if (process.env.ROLL_FILES) for (const [source, target] of JSON.parse(process.env.ROLL_FILES)) renameSync(source, target);
if (process.env.ARCHIVE === 'refuse') send({ id: frame.id, error: { code: -32600, message: 'thread already has an active writer' } });
else if (process.env.ARCHIVE_GATE) {
const watcher = watch(process.cwd(), () => {
if (existsSync(process.env.ARCHIVE_GATE)) { watcher.close(); send({ id: frame.id, result: {} }); }
});
if (existsSync(process.env.ARCHIVE_GATE)) { watcher.close(); send({ id: frame.id, result: {} }); }
} else send({ id: frame.id, result: {} });
} else if (frame.method !== 'initialized') process.exit(3);
});
`;
type Frame = { pid: number; marker: string; cwd: string; method: string; params?: { threadId?: string } };
async function scenario(mode: 'abnormal' | 'cancel' | 'normal', refuse = false, control: 'ordinary' | 'shutdown' | 'deadline' | 'unowned' = 'ordinary') {
const root = await mkdtemp(path.join(tmpdir(), 'codex-close-owner-'));
const script = path.join(root, 'fixture.ts');
const transcript = path.join(root, 'frames.jsonl');
await writeFile(script, fixture);
const invocation = { command: process.execPath, args: [script], cwd: root,
env: { PATH: process.env.PATH, TRANSCRIPT: transcript, CASE_MARKER: 'original-context', MODE: mode, ARCHIVE: refuse ? 'refuse' : 'success',
ARCHIVE_GATE: control === 'shutdown' ? path.join(root, 'release-archive') : '' },
windowsVerbatimArguments: false };
const child = spawn(invocation.command, invocation.args, { cwd: root, env: invocation.env, stdio: 'pipe', detached: process.platform !== 'win32' });
child.stderr.resume();
const closed = once(child, 'close');
const runs = createChatRunService({ createSseResponse: () => ({ send: vi.fn(), end: vi.fn(), cleanup: vi.fn() }),
createSseErrorPayload: (code: string, message: string) => ({ error: { code, message } }) });
const run = runs.create({ agentId: 'codex' });
run.status = 'running';
// The legacy service infers initial null slots; fill them with actual typed handles.
Object.assign(run, { child, childPid: child.pid ?? null,
processGroupId: process.platform !== 'win32' ? child.pid ?? null : null,
});
const reports: CodexThreadCleanupResult[] = [];
let reportComplete: () => void = () => {};
const reported = new Promise<void>(resolve => { reportComplete = resolve; });
const owner = createCodexThreadCleanupOwner();
const session = attachCodexAppServerSession({ child, prompt: 'Local owned cleanup fixture', cwd: root,
sandboxMode: 'workspace-write', manageThreadVisibility: control !== 'unowned', onAgentEvent: () => {} });
Object.assign(run, { acpSession: session });
// This is the production attach order: session installs its proof listener,
// then owner binds close, then the ordinary run finalizer may abort it.
owner.bind(child, session, invocation, result => { reports.push(result); reportComplete(); });
owner.bind(child, session, invocation, () => { throw new Error('Duplicate binding must never report'); });
invocation.args[0] = path.join(root, 'later-config-command.ts');
invocation.env.CASE_MARKER = 'later-config-marker';
invocation.cwd = path.join(root, 'later-config-cwd');
child.once('close', (code) => {
session.abort();
runs.finish(run, run.cancelRequested ? 'canceled' : code === 0 ? 'succeeded' : 'failed', code, null);
});
try {
await vi.waitFor(async () => {
const frames = (await readFile(transcript, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as Frame);
expect(frames.some(frame => frame.method === 'turn/start')).toBe(true);
});
if (control === 'deadline') {
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
try {
let drained = false;
const drain = owner.drain(999_999).then(result => { drained = true; return result; });
await vi.advanceTimersByTimeAsync(2999);
expect(drained).toBe(false);
await vi.advanceTimersByTimeAsync(1);
expect(await drain).toEqual({ pending: 1 });
} finally { vi.useRealTimers(); }
}
if (control === 'shutdown') {
await runs.shutdownActive({ graceMs: 10 });
let drained = false;
const drain = owner.drain(3000).then(result => { drained = true; return result; });
await vi.waitFor(async () => {
const requests = (await readFile(transcript, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as Frame);
expect(requests.some(frame => frame.pid !== child.pid && frame.method === 'thread/archive')).toBe(true);
});
expect(drained).toBe(false);
expect(run.status).toBe('canceled');
expect(run.cancelOrigin).toBe('daemon_shutdown');
await writeFile(path.join(root, 'release-archive'), 'release');
expect(await drain).toEqual({ pending: 0 });
} else if (mode === 'cancel') await runs.cancel(run, 'user_stop');
await closed;
const terminal = { status: run.status, error: run.error, errorCode: run.errorCode, cancelOrigin: run.cancelOrigin };
await owner.drain(3000);
expect({ status: run.status, error: run.error, errorCode: run.errorCode, cancelOrigin: run.cancelOrigin }).toEqual(terminal);
const frames = (await readFile(transcript, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as Frame);
const cleanup = frames.filter(frame => frame.pid !== child.pid);
expect(run.status).toBe(mode === 'cancel' ? 'canceled' : mode === 'normal' ? 'succeeded' : 'failed');
if (control === 'deadline' || control === 'unowned') {
expect(reports).toEqual([]);
expect(cleanup).toEqual([]);
} else if (mode === 'normal') {
expect(reports).toEqual([]);
expect(cleanup).toEqual([]);
expect(frames.filter(frame => frame.method === 'thread/archive')).toHaveLength(1);
} else {
expect(cleanup.map(frame => frame.method)).toEqual(['initialize', 'initialized', 'thread/archive']);
expect(reports).toHaveLength(1);
expect(reports[0]?.status).toBe(refuse ? 'failed' : 'archived');
expect(cleanup.at(-1)?.params).toEqual({ threadId: 'owned-close-thread' });
const originalCwd = await realpath(root);
expect(cleanup.every(frame => frame.cwd === originalCwd && frame.marker === 'original-context')).toBe(true);
expect(session.takeClosedThreadCleanup()).toBeNull();
for (const pid of new Set(cleanup.map(frame => frame.pid))) {
if (!Number.isSafeInteger(pid) || pid <= 0) throw new Error('Invalid fixture process identity');
expect(() => process.kill(pid, 0)).toThrow();
}
}
} finally {
if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await closed; }
try {
if (control === 'shutdown') await writeFile(path.join(root, 'release-archive'), 'final cleanup release');
const remainder = await owner.drain(3000);
// The source is closed above. If a deliberately withdrawn drain returns
// early, still observe the bounded owned helper before removing its cwd.
if (remainder.pending > 0) await reported;
} finally { await rm(root, { recursive: true, force: true }); }
}
}
describe('Codex close ownership at the real session attachment boundary', () => {
it('archives abnormal exit after session proves physical close, before a later abort discards the attempt', async () => scenario('abnormal'));
it('retains user cancellation while archiving only the closed owned writer', async () => scenario('cancel'));
it('does not duplicate normal terminal archive', async () => scenario('normal'));
it('drains the owned archive ACK after real shutdownActive while preserving daemon cancellation', async () => scenario('cancel', false, 'shutdown'));
it('caps shutdown drain at three seconds and never launches cleanup on a later close', async () => scenario('cancel', false, 'deadline'));
it('does not use a normal thread response as cleanup authority when visibility ownership was not granted', async () => scenario('abnormal', false, 'unowned'));
it('reports another writer refusal without changing the original failed run', async () => scenario('abnormal', true));
});
function deferred() {
let resolve: () => void = () => {};
const promise = new Promise<void>(done => { resolve = done; });
return { promise, resolve };
}
async function collectorScenario(mode: 'abnormal' | 'normal') {
const root = await mkdtemp(path.join(tmpdir(), 'codex-close-owner-evidence-'));
const active = path.join(root, 'sessions', '2026', '08', '14');
const archived = path.join(root, 'archived_sessions');
const parentId = '10000000-0000-4000-8000-000000000001';
const childId = '20000000-0000-4000-8000-000000000002';
const timestamp = (offset: number) => new Date(1786665600000 + offset).toISOString();
const event = (offset: number, payload: Record<string, unknown>) => ({ timestamp: timestamp(offset), type: 'event_msg', payload });
const parentPath = path.join(active, `rollout-2026-08-14-${parentId}.jsonl`);
const childPath = path.join(active, `rollout-2026-08-14-${childId}.jsonl`);
await mkdir(active, { recursive: true, mode: 0o700 });
await mkdir(archived, { mode: 0o700 });
const logs = [
{ file: parentPath, lines: [
{ timestamp: timestamp(0), type: 'session_meta', payload: { id: parentId } },
event(100, { type: 'task_started', turn_id: 'parent-turn' }),
event(1000, { type: 'sub_agent_activity', agent_thread_id: childId, kind: 'started', occurred_at_ms: 1786665601000 }),
event(6000, { type: 'sub_agent_activity', agent_thread_id: childId, kind: 'completed', occurred_at_ms: 1786665606000 }),
event(7000, { type: 'task_complete', turn_id: 'parent-turn' }),
] },
{ file: childPath, lines: [
{ timestamp: timestamp(0), type: 'session_meta', payload: { id: childId, parent_thread_id: parentId } },
event(2000, { type: 'task_started', turn_id: 'child-turn' }),
event(2010, { type: 'user_message', message: 'Read the scoped child evidence' }),
event(2900, { type: 'task_complete', turn_id: 'child-turn' }),
] },
];
for (const log of logs) await writeFile(log.file, log.lines.map(line => JSON.stringify(line)).join('\n') + '\n', { mode: 0o600 });
const script = path.join(root, 'fixture.ts');
const transcript = path.join(root, 'frames.jsonl');
await writeFile(script, fixture);
const invocation = { command: process.execPath, args: [script], cwd: root,
env: { PATH: process.env.PATH, MODE: mode, TRANSCRIPT: transcript, THREAD_ID: parentId,
RACE_HOME: root, ROLL_FILES: JSON.stringify(logs.map(log => [log.file, path.join(archived, path.basename(log.file))])) },
windowsVerbatimArguments: false };
const evidenceDone = deferred();
const archiveDone = deferred();
const owner = createCodexThreadCleanupOwner();
const reports: CodexThreadCleanupResult[] = [];
const child = spawn(invocation.command, invocation.args, { cwd: root, env: invocation.env, stdio: 'pipe', detached: process.platform !== 'win32' });
child.stderr.resume();
const closed = once(child, 'close');
const session = attachCodexAppServerSession({ child, prompt: 'Scoped evidence', cwd: root,
sandboxMode: 'workspace-write', manageThreadVisibility: true, onAgentEvent: () => {} });
Object.assign(collectorRace, { home: root, armed: true, helperStarted: false, activePath: parentPath, archiveFinished: archiveDone.promise });
owner.bind(child, session, invocation, result => { reports.push(result); archiveDone.resolve(); }, evidenceDone.promise);
const collection = new Promise<CodexChildEvidenceCollection>((resolve, reject) => {
child.once('close', () => {
void collectCodexChildEvidence({ codexHome: root, parentSessionId: parentId, parentTurnId: 'parent-turn',
taskExecutionId: 'task-1', runId: 'run-1', taskRunIndex: 0, stage: 'production', parentObservationId: 'root',
runStartedAtMs: 1786665600000, runEndedAtMs: 1786665610000,
}).then(resolve, reject).finally(evidenceDone.resolve);
});
});
try {
const evidence = await collection;
await closed;
await owner.drain(3000);
expect(evidence.availability).toBe('complete');
expect(evidence.knownChildCount).toBe(1);
expect(evidence.observations.filter(observation => observation.kind === 'child_agent' && observation.status === 'completed')).toHaveLength(1);
expect(reports).toHaveLength(mode === 'normal' ? 0 : 1);
const frames = (await readFile(transcript, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as Frame);
expect(frames.filter(frame => frame.method === 'thread/archive')).toHaveLength(1);
for (const log of logs) expect(await readFile(path.join(archived, path.basename(log.file)), 'utf8')).toBe(log.lines.map(line => JSON.stringify(line)).join('\n') + '\n');
} finally {
evidenceDone.resolve();
archiveDone.resolve();
collectorRace.armed = false;
if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await closed; }
await owner.drain(3000);
await rm(root, { recursive: true, force: true });
}
}
describe('Codex close archive and actual rollout collection', () => {
it('keeps completed child facts when abnormal close would move the selected active rollout before open', async () => collectorScenario('abnormal'));
it('reads the normal terminal archive that was already moved before physical close', async () => collectorScenario('normal'));
});
async function evidenceExitScenario(boundary: 'superseded' | 'early-return' | 'collector-failure' | 'deadline') {
const root = await mkdtemp(path.join(tmpdir(), 'codex-close-owner-boundary-'));
const script = path.join(root, 'fixture.ts');
const transcript = path.join(root, 'frames.jsonl');
await writeFile(script, fixture);
const invocation = { command: process.execPath, args: [script], cwd: root,
env: { PATH: process.env.PATH, MODE: 'cancel', TRANSCRIPT: transcript, RACE_HOME: root },
windowsVerbatimArguments: false };
const child = spawn(invocation.command, invocation.args, { cwd: root, env: invocation.env, stdio: 'pipe', detached: process.platform !== 'win32' });
child.stderr.resume();
const closed = once(child, 'close');
const successor = boundary === 'superseded'
? spawn(process.execPath, ['-e', 'process.stdin.resume()'], { stdio: 'pipe' })
: null;
successor?.stderr.resume();
const successorClosed = successor ? once(successor, 'close') : null;
const runs = createChatRunService({ createSseResponse: () => ({ send: vi.fn(), end: vi.fn(), cleanup: vi.fn() }),
createSseErrorPayload: (code: string, message: string) => ({ error: { code, message } }) });
const run = runs.create({ agentId: 'codex' });
run.status = 'running';
Object.assign(run, { child, childPid: child.pid ?? null,
processGroupId: process.platform !== 'win32' ? child.pid ?? null : null });
const evidenceDone = deferred();
const readRelease = deferred();
const handlerDone = deferred();
const owner = createCodexThreadCleanupOwner();
const reports: CodexThreadCleanupResult[] = [];
const errors: string[] = [];
const session = attachCodexAppServerSession({ child, prompt: 'Scoped closure', cwd: root,
sandboxMode: 'workspace-write', manageThreadVisibility: true, onAgentEvent: () => {} });
Object.assign(run, { acpSession: session });
Object.assign(collectorRace, { home: root, armed: true, helperStarted: false, activePath: '' });
owner.bind(child, session, invocation, result => { reports.push(result); }, evidenceDone.promise);
child.once('close', () => {
void (async () => {
try {
// Mirror the server's two early returns before its evidence collector.
if (run.child !== child || boundary === 'early-return') return;
if (boundary === 'collector-failure') throw new Error('Scoped collector failed');
await readRelease.promise;
} catch (error) {
errors.push(String(error));
} finally {
evidenceDone.resolve();
handlerDone.resolve();
}
})();
});
try {
await vi.waitFor(async () => {
const frames = (await readFile(transcript, 'utf8')).trim().split('\n').map(line => JSON.parse(line) as Frame);
expect(frames.some(frame => frame.method === 'turn/start')).toBe(true);
});
if (successor) {
Object.assign(run, { child: successor, childPid: successor.pid ?? null, acpSession: null });
child.kill('SIGTERM');
} else await runs.cancel(run, 'user_stop');
await closed;
const terminal = { status: run.status, error: run.error, errorCode: run.errorCode, cancelOrigin: run.cancelOrigin, child: run.child };
if (boundary === 'deadline') {
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] });
try {
let drained = false;
const drain = owner.drain(999_999).then(result => { drained = true; return result; });
await vi.advanceTimersByTimeAsync(2999);
expect(drained).toBe(false);
await vi.advanceTimersByTimeAsync(1);
expect(await drain).toEqual({ pending: 1 });
} finally { vi.useRealTimers(); }
readRelease.resolve();
await handlerDone.promise;
await owner.drain(3000);
expect(collectorRace.helperStarted).toBe(false);
expect(reports).toEqual([]);
} else {
await handlerDone.promise;
expect(await owner.drain(3000)).toEqual({ pending: 0 });
expect(reports).toHaveLength(1);
expect(reports[0]?.status).toBe('archived');
expect(errors).toEqual(boundary === 'collector-failure' ? ['Error: Scoped collector failed'] : []);
}
expect({ status: run.status, error: run.error, errorCode: run.errorCode, cancelOrigin: run.cancelOrigin, child: run.child }).toEqual(terminal);
expect(run.status).toBe(successor ? 'running' : 'canceled');
if (successor) {
expect(successor.exitCode).toBeNull();
expect(successor.signalCode).toBeNull();
}
} finally {
readRelease.resolve();
evidenceDone.resolve();
collectorRace.armed = false;
if (child.exitCode === null && child.signalCode === null) { child.kill('SIGKILL'); await closed; }
if (successor && successorClosed) {
if (successor.exitCode === null && successor.signalCode === null) successor.kill('SIGKILL');
await successorClosed;
}
await owner.drain(3000);
await rm(root, { recursive: true, force: true });
}
}
describe('Codex evidence completion and attempt ownership', () => {
it('releases old-generation cleanup while the replacement child and run stay active', async () => evidenceExitScenario('superseded'));
it('releases cleanup after an early terminal close return', async () => evidenceExitScenario('early-return'));
it('preserves a reported collector failure and user cancellation while releasing cleanup', async () => evidenceExitScenario('collector-failure'));
it('does not launch an archive when a held collector completes after the shutdown deadline', async () => evidenceExitScenario('deadline'));
});
@@ -0,0 +1,211 @@
import { EventEmitter, once } from 'node:events';
import { spawn } from 'node:child_process';
import { mkdtemp, writeFile, readFile, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { describe, expect, it, vi } from 'vitest';
const state = vi.hoisted(() => ({ metadata: '', calls: 0, failRefresh: false, failInitial: false, stopped: [] as number[][] }));
vi.mock('@open-design/platform', async (load) => {
const actual = await load<typeof import('@open-design/platform')>();
return { ...actual,
captureProcessSnapshot: async (options: { timeoutMs?: number }) => {
expect(options.timeoutMs).toBe(500);
state.calls++;
if (state.failInitial || (state.calls > 1 && state.failRefresh)) throw new Error('controlled CIM timeout');
const pids = JSON.parse(await readFile(state.metadata, 'utf8')) as { wrapper: number; descendant: number };
return [
{ pid: pids.wrapper, ppid: process.pid, command: 'controlled wrapper', startedAtMs: 100 },
{ pid: pids.descendant, ppid: pids.wrapper, command: 'controlled child', startedAtMs: 101 },
].filter(entry => Number.isSafeInteger(entry.pid) && entry.pid > 0 && actual.isProcessAlive(entry.pid));
},
stopProcesses: async (pids: number[], options: { termGraceMs: number; killGraceMs: number }) => {
if (!pids.every(pid => Number.isSafeInteger(pid) && pid > 0)) throw new Error('unsafe PID in controlled fixture');
state.stopped.push(pids);
return actual.stopProcesses(pids, options);
},
};
});
import { attachCodexAppServerSession } from '../src/agent-protocol/codex-app-server/session.js';
import { cleanupClosedCodexThread } from '../src/agent-protocol/codex-app-server/thread-cleanup.js';
import { captureProcessSnapshot, isProcessAlive, stopProcesses, waitForProcessExit } from '@open-design/platform';
async function receipt() {
let ready!: () => void;
const readyPromise = new Promise<void>(resolve => { ready = resolve; });
const source = Object.assign(new EventEmitter(), { stdout: new EventEmitter(), stdin: {
write: (line: string) => {
const request = JSON.parse(line) as { id: number; method: string };
if (request.method === 'initialize' || request.method === 'thread/start') queueMicrotask(() => {
source.stdout.emit('data', JSON.stringify({ id: request.id, result: request.method === 'initialize'
? { userAgent: 'codex/0.154.0' } : { thread: { id: 'owned-thread', historyMode: 'paginated' } } }) + '\n');
});
}, end: () => {},
} });
const session = attachCodexAppServerSession({ child: source, cwd: '/', prompt: 'owned', sandboxMode: 'workspace-write',
manageThreadVisibility: true, onAgentEvent: () => {}, onSessionReady: ready });
await readyPromise; source.emit('close');
return session.takeClosedThreadCleanup();
}
// Exercise the actual Windows lifecycle branch with real wrapper/descendant
// stdio processes. Only Win32/CIM process records are controlled: this is NOT
// native Windows .cmd or CIM acceptance, which requires a Windows runner.
describe('controlled process fixture PID safety', () => {
it.each([-1, 0, 0.5, Number.MAX_SAFE_INTEGER + 1, '1', null])(
'rejects invalid PID %s before any OS process operation', async (pid) => {
const root = await mkdtemp(path.join(tmpdir(), 'codex-fixture-pid-'));
state.metadata = path.join(root, 'owned-pids.json'); state.calls = 0; state.stopped = [];
state.failInitial = false; state.failRefresh = false;
await writeFile(state.metadata, JSON.stringify({ wrapper: pid, descendant: pid }));
// Even if this guard regresses, this test cannot deliver any OS signal.
const processCalls = vi.spyOn(process, 'kill').mockReturnValue(true);
try {
expect(await captureProcessSnapshot({ timeoutMs: 500 })).toEqual([]);
await expect(stopProcesses([pid as number], { termGraceMs: 0, killGraceMs: 0 }))
.rejects.toThrow('unsafe PID in controlled fixture');
expect(state.stopped).toEqual([]);
expect(processCalls).not.toHaveBeenCalled();
} finally {
processCalls.mockRestore();
await rm(root, { recursive: true, force: true });
}
},
);
});
describe('Windows owned cleanup lifecycle with controlled generation records', () => {
it.each(['wrapper-alive', 'wrapper-exited', 'snapshot-failure', 'initial-snapshot-failure'])(
'bounds cleanup and preserves ownership when %s', async (mode) => {
const root = await mkdtemp(path.join(tmpdir(), 'codex-windows-tree-'));
const wrapper = path.join(root, 'wrapper.ts');
const descendant = path.join(root, 'descendant.ts');
state.metadata = path.join(root, 'owned-pids.json'); state.calls = 0; state.stopped = [];
state.failRefresh = mode === 'snapshot-failure'; state.failInitial = mode === 'initial-snapshot-failure';
await writeFile(descendant, `import { createInterface } from 'node:readline';
setInterval(() => {}, 1000);
createInterface({ input: process.stdin }).on('line', line => {
const frame = JSON.parse(line);
if (frame.method === 'initialize') process.stdout.write(JSON.stringify({ id: frame.id, result: { userAgent: 'codex/0.154.0' } }) + '\\n');
if (frame.method === 'thread/archive') {
setInterval(() => {}, 1000);
if (process.env.EXIT_WRAPPER === 'yes') process.send('exit-wrapper');
}
});`);
await writeFile(wrapper, `import { spawn } from 'node:child_process'; import { writeFileSync } from 'node:fs';
const child = spawn(process.execPath, [process.argv[2]], { stdio: ['inherit', 'inherit', 'inherit', 'ipc'] });
writeFileSync(process.env.OWNED_PIDS, JSON.stringify({ wrapper: process.pid, descendant: child.pid }));
child.on('message', () => process.exit(0)); child.on('exit', code => process.exit(code ?? 1));`);
const authority = await receipt();
const sibling = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' });
const siblingClosed = once(sibling, 'close');
const platform = Object.getOwnPropertyDescriptor(process, 'platform')!;
let cleanup: ReturnType<typeof cleanupClosedCodexThread> | undefined;
let watchdog: ReturnType<typeof setTimeout> | undefined;
try {
Object.defineProperty(process, 'platform', { value: 'win32', configurable: true });
cleanup = cleanupClosedCodexThread({ receipt: authority, command: process.execPath, args: [wrapper, descendant], cwd: root,
env: { PATH: process.env.PATH, OWNED_PIDS: state.metadata, EXIT_WRAPPER: mode === 'wrapper-exited' ? 'yes' : 'no' } });
const result = await Promise.race([cleanup, new Promise<'still-pending'>(resolve => {
watchdog = setTimeout(() => resolve('still-pending'), 4000);
})]);
expect(result).not.toBe('still-pending');
expect(result).toMatchObject({ status: 'failed' });
const owned = JSON.parse(await readFile(state.metadata, 'utf8')) as { wrapper: number; descendant: number };
expect(await waitForProcessExit(owned.wrapper, 500)).toBe(true);
if (mode === 'initial-snapshot-failure') {
expect(result).toMatchObject({ treeVerified: false, remainingPids: [] });
expect(state.stopped).toEqual([]);
expect(isProcessAlive(owned.descendant)).toBe(true);
} else if (mode === 'snapshot-failure') {
expect(result).toMatchObject({ treeVerified: false, remainingPids: expect.arrayContaining([owned.descendant]) });
expect(state.stopped).toEqual([]);
expect(isProcessAlive(owned.descendant)).toBe(true);
} else {
expect(result).toMatchObject({ treeVerified: true, remainingPids: [] });
expect(await waitForProcessExit(owned.descendant, 500)).toBe(true);
expect(state.stopped.flat()).toContain(owned.descendant);
}
expect(isProcessAlive(sibling.pid)).toBe(true);
} finally {
clearTimeout(watchdog);
Object.defineProperty(process, 'platform', platform);
const owned = JSON.parse(await readFile(state.metadata, 'utf8')) as { wrapper: number; descendant: number };
for (const pid of [owned.descendant, owned.wrapper].filter(pid => Number.isSafeInteger(pid) && pid > 0)) {
try { process.kill(pid, 'SIGKILL'); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error; }
}
await cleanup;
sibling.kill('SIGKILL'); await siblingClosed;
await rm(root, { recursive: true, force: true });
}
},
);
it.each(['known-child', 'late-unknown-child', 'known-child-snapshot-failure'])(
'distinguishes archive acknowledgement from process cleanup for %s', async (mode) => {
const root = await mkdtemp(path.join(tmpdir(), 'codex-windows-ack-'));
const wrapper = path.join(root, 'wrapper.ts');
state.metadata = path.join(root, 'owned-pids.json'); state.calls = 0; state.stopped = [];
state.failRefresh = mode === 'known-child-snapshot-failure'; state.failInitial = false;
await writeFile(wrapper, `import { spawn } from 'node:child_process';
import { writeFileSync } from 'node:fs'; import { createInterface } from 'node:readline';
let child;
const save = () => writeFileSync(process.env.OWNED_PIDS, JSON.stringify({ wrapper: process.pid, descendant: child?.pid ?? -1 }));
const createChild = () => { child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' }); save(); };
if (process.env.CHILD_MODE !== 'late-unknown-child') createChild(); else save();
createInterface({ input: process.stdin }).on('line', line => {
const frame = JSON.parse(line);
if (frame.method === 'initialize') process.stdout.write(JSON.stringify({ id: frame.id, result: { userAgent: 'codex/0.154.0' } }) + '\\n');
if (frame.method === 'thread/archive') {
if (!child) createChild();
process.stdout.write(JSON.stringify({ id: frame.id, result: {} }) + '\\n', () => process.exit(0));
}
});`);
const authority = await receipt();
const sibling = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' });
const siblingClosed = once(sibling, 'close');
const platform = Object.getOwnPropertyDescriptor(process, 'platform')!;
let cleanup: ReturnType<typeof cleanupClosedCodexThread> | undefined;
try {
Object.defineProperty(process, 'platform', { value: 'win32', configurable: true });
cleanup = cleanupClosedCodexThread({ receipt: authority, command: process.execPath, args: [wrapper], cwd: root,
env: { PATH: process.env.PATH, OWNED_PIDS: state.metadata, CHILD_MODE: mode } });
const result = await cleanup;
const owned = JSON.parse(await readFile(state.metadata, 'utf8')) as { wrapper: number; descendant: number };
expect(await waitForProcessExit(owned.wrapper, 500)).toBe(true);
if (mode === 'known-child-snapshot-failure') {
expect(result).toMatchObject({ status: 'failed', treeVerified: false,
message: expect.stringContaining('archive acknowledged; process cleanup failed'),
remainingPids: expect.arrayContaining([owned.descendant]) });
expect(isProcessAlive(owned.descendant)).toBe(true);
expect(state.stopped).toEqual([]);
} else if (mode === 'known-child') {
expect(result).toMatchObject({ status: 'archived', treeVerified: false });
expect(await waitForProcessExit(owned.descendant, 500)).toBe(true);
expect(state.stopped.flat()).toContain(owned.descendant);
} else {
// It appeared only after the ownership snapshot. The dead wrapper
// cannot prove this new process belongs to us, even with its old ppid.
expect(result).toMatchObject({ status: 'archived', treeVerified: false });
expect(isProcessAlive(owned.descendant)).toBe(true);
expect(state.stopped.flat()).not.toContain(owned.descendant);
}
expect(isProcessAlive(sibling.pid)).toBe(true);
} finally {
Object.defineProperty(process, 'platform', platform);
const stored = await readFile(state.metadata, 'utf8').catch((error: NodeJS.ErrnoException) => {
if (error.code !== 'ENOENT') throw error;
return null;
});
const owned = stored ? JSON.parse(stored) as { wrapper: number; descendant: number } : null;
for (const pid of (owned ? [owned.descendant, owned.wrapper] : []).filter(pid => Number.isSafeInteger(pid) && pid > 0)) {
try { process.kill(pid, 'SIGKILL'); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error; }
}
await cleanup;
sibling.kill('SIGKILL'); await siblingClosed;
for (const pid of (owned ? [owned.descendant, owned.wrapper] : []).filter(pid => Number.isSafeInteger(pid) && pid > 0)) expect(await waitForProcessExit(pid, 500)).toBe(true);
await rm(root, { recursive: true, force: true });
}
},
);
});
@@ -0,0 +1,224 @@
import { spawn } from 'node:child_process';
import { EventEmitter, once } from 'node:events';
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { describe, expect, it } from 'vitest';
import { waitForProcessExit } from '@open-design/platform';
import { attachCodexAppServerSession } from '../src/agent-protocol/codex-app-server/session.js';
import { cleanupClosedCodexThread } from '../src/agent-protocol/codex-app-server/thread-cleanup.js';
// A real stdio child with controlled protocol responses, no provider or service.
// It records every request before responding, so prohibited resume/list/start
// calls from the cleanup process cannot pass unnoticed.
const fixture = `
import { appendFileSync } from 'node:fs';
import { createInterface } from 'node:readline';
const send = (id, result) => process.stdout.write(JSON.stringify({ id, result }) + '\\n');
const lines = createInterface({ input: process.stdin });
lines.on('line', line => {
const frame = JSON.parse(line);
appendFileSync(process.env.TRANSCRIPT, JSON.stringify({ pid: process.pid, cwd: process.cwd(), marker: process.env.CASE_MARKER, ...frame }) + '\\n');
if (frame.method === 'initialize') {
if (process.env.INIT === 'hang') return;
if (process.env.INIT === 'invalid') { send(frame.id, []); return; }
if (process.env.INIT === 'refuse') {
process.stdout.write(JSON.stringify({ id: frame.id, error: { message: 'initialization refused' } }) + '\\n');
return;
}
send(frame.id, { userAgent: 'codex/' + (process.env.VERSION || '0.154.0') });
} else if (frame.method === 'thread/start') {
send(frame.id, { thread: { id: 'owned-thread', historyMode: process.env.HISTORY_MODE || 'paginated' } });
} else if (frame.method === 'turn/start') {
if (process.env.SOURCE_EXIT === 'kill') process.kill(process.pid, 'SIGKILL');
else process.exit(Number(process.env.SOURCE_EXIT || 0));
} else if (frame.method === 'thread/archive') {
if (process.env.ARCHIVE === 'hang') { setInterval(() => {}, 1000); return; }
if (process.env.ARCHIVE === 'exit') process.exit(2);
if (process.env.ARCHIVE === 'refuse') {
process.stdout.write(JSON.stringify({ id: frame.id, error: { code: -32600, message: 'thread already has an active writer' } }) + '\\n');
} else send(frame.id, {});
} else if (frame.method !== 'initialized') process.exit(3);
});
`;
async function scenario(options: { sourceExit?: string; archive?: string; cleanupVersion?: string; missingCommand?: boolean; init?: string; historyMode?: 'legacy' | 'paginated' } = {}) {
const root = await mkdtemp(path.join(tmpdir(), 'codex-close-cleanup-'));
const transcript = path.join(root, 'requests.jsonl');
const script = path.join(root, 'controlled-cli.ts');
await writeFile(script, fixture);
const invocation = {
command: process.execPath,
args: [script],
cwd: root,
env: { PATH: process.env.PATH, TRANSCRIPT: transcript, CASE_MARKER: 'same-original-context',
SOURCE_EXIT: options.sourceExit ?? '0', ARCHIVE: options.archive ?? 'success', HISTORY_MODE: options.historyMode ?? 'paginated' },
};
const original = spawn(invocation.command, invocation.args, { cwd: root, env: invocation.env, stdio: 'pipe' });
original.stderr.resume();
const session = attachCodexAppServerSession({
child: original, prompt: 'controlled owned prompt', cwd: root,
sandboxMode: 'workspace-write', manageThreadVisibility: true, onAgentEvent: () => {},
});
try {
expect(session.takeClosedThreadCleanup()).toBeNull();
await once(original, 'close');
const receipt = session.takeClosedThreadCleanup();
expect(receipt).toMatchObject({ threadId: 'owned-thread', historyMode: options.historyMode ?? 'paginated' });
expect(session.takeClosedThreadCleanup()).toBeNull();
// A version change explicitly models the original executable being upgraded
// or downgraded between exits; normal cases pass the exact original context.
const cleanup = await cleanupClosedCodexThread({ ...invocation, receipt,
env: { ...invocation.env, ...(options.cleanupVersion ? { VERSION: options.cleanupVersion } : {}),
...(options.init ? { INIT: options.init } : {}) },
...(options.missingCommand ? { command: path.join(root, 'missing-cli') } : {}),
});
const requests = (await readFile(transcript, 'utf8')).trim().split('\n')
.map(line => JSON.parse(line) as { pid: number; cwd: string; marker: string; method: string; params?: { threadId?: string } });
const cleanupRequests = requests.filter(request => request.pid !== original.pid);
for (const pid of new Set(cleanupRequests.map(request => request.pid))) {
expect(() => process.kill(pid, 0)).toThrow();
}
expect(cleanupRequests.every(request => request.cwd === requests[0]?.cwd)).toBe(true);
expect(session.completedSuccessfully()).toBe(false);
expect(cleanupRequests.every(request => request.marker === 'same-original-context')).toBe(true);
expect(cleanupRequests.map(request => request.method)).not.toContain('thread/resume');
expect(cleanupRequests.map(request => request.method)).not.toContain('thread/list');
expect(cleanupRequests.map(request => request.method)).not.toContain('turn/start');
return { cleanup, cleanupRequests };
} finally {
if (original.exitCode === null && original.signalCode === null) {
original.kill('SIGKILL');
await once(original, 'close');
}
await rm(root, { recursive: true, force: true });
}
}
describe('closed owned Codex thread cleanup over real stdio', () => {
it.each(['0', '1', 'kill'])('archives only the captured owned ID after physical exit %s', async (sourceExit) => {
const { cleanup, cleanupRequests } = await scenario({ sourceExit });
expect(cleanup).toEqual({ status: 'archived' });
expect(cleanupRequests.map(request => request.method)).toEqual(['initialize', 'initialized', 'thread/archive']);
expect(cleanupRequests.at(-1)?.params).toEqual({ threadId: 'owned-thread' });
});
it('preserves another writer and reports the archive refusal without retrying', async () => {
const { cleanup, cleanupRequests } = await scenario({ archive: 'refuse' });
expect(cleanup).toMatchObject({ status: 'failed', message: expect.stringContaining('active writer') });
expect(cleanupRequests.filter(request => request.method === 'thread/archive')).toHaveLength(1);
});
it.each(['unknown', '0.145.0'])('does not send archive when the fresh handshake cannot verify locks: %s', async (cleanupVersion) => {
const { cleanup, cleanupRequests } = await scenario({ cleanupVersion });
expect(cleanup).toEqual({ status: 'skipped', reason: 'unverified-writer-lock' });
expect(cleanupRequests.map(request => request.method)).toEqual(['initialize']);
});
it('returns a cleanup failure when the new process exits without an archive acknowledgement', async () => {
expect((await scenario({ archive: 'exit' })).cleanup).toMatchObject({ status: 'failed', message: expect.stringContaining('before acknowledgement') });
});
it('reaps its own hung cleanup process within the existing archive budget', async () => {
// Real child I/O and actual process closure are the assertion boundary;
// fake timers would not establish that the precise spawned PID was reaped.
expect((await scenario({ archive: 'hang' })).cleanup).toEqual({ status: 'failed', message: 'codex thread cleanup timed out' });
});
it('does not archive legacy history after a downgrade to paginated-only writer protection', async () => {
const { cleanup, cleanupRequests } = await scenario({ historyMode: 'legacy', cleanupVersion: '0.146.0' });
expect(cleanup).toEqual({ status: 'skipped', reason: 'unverified-writer-lock' });
expect(cleanupRequests.map(request => request.method)).toEqual(['initialize']);
});
it.each(['refuse', 'invalid', 'hang'])('handles initialization %s without touching the thread', async (init) => {
const { cleanup, cleanupRequests } = await scenario({ init });
expect(cleanup.status).toBe('failed');
expect(cleanupRequests.map(request => request.method)).toEqual(['initialize']);
});
it('handles an executable disappearing after the original child exited', async () => {
expect((await scenario({ missingCommand: true })).cleanup).toMatchObject({ status: 'failed', message: expect.stringContaining('ENOENT') });
});
it.skipIf(process.platform === 'win32').each([false, true])('bounds POSIX wrapper cleanup with detached descendant=%s', async (detached) => {
// Ownership/session behavior is covered above with physical source children.
// This test isolates the new cleanup process tree with a protocol-issued receipt.
const source = Object.assign(new EventEmitter(), { stdout: new EventEmitter(),
stdin: { write: (line: string) => {
const request = JSON.parse(line) as { id: number; method: string };
if (request.method === 'initialize' || request.method === 'thread/start') {
queueMicrotask(() => source.stdout.emit('data', JSON.stringify({ id: request.id,
result: request.method === 'initialize' ? { userAgent: 'codex/0.154.0' }
: { thread: { id: 'owned-thread', historyMode: 'paginated' } } }) + '\n'));
}
}, end: () => {} } });
let ready!: () => void;
const readyPromise = new Promise<void>(resolve => { ready = resolve; });
const session = attachCodexAppServerSession({ child: source, prompt: 'owned', cwd: '/',
sandboxMode: 'workspace-write', manageThreadVisibility: true, onAgentEvent: () => {}, onSessionReady: ready });
await readyPromise;
source.emit('close');
const receipt = session.takeClosedThreadCleanup();
expect(receipt).not.toBeNull();
const root = await mkdtemp(path.join(tmpdir(), 'codex-cleanup-wrapper-'));
const script = path.join(root, 'controlled-cli.ts');
const wrapper = path.join(root, 'wrapper.ts');
const metadata = path.join(root, 'owned-pids.json');
const transcript = path.join(root, 'requests.jsonl');
await writeFile(script, fixture);
await writeFile(wrapper, `import { spawn } from 'node:child_process';
import { writeFileSync } from 'node:fs';
const child = spawn(process.execPath, [process.argv[2]], { stdio: 'inherit', detached: ${detached} });
writeFileSync(process.env.OWNED_PIDS, JSON.stringify({ wrapper: process.pid, descendant: child.pid }));
child.on('exit', code => process.exit(code ?? 1));`);
const unrelated = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], { stdio: 'ignore' });
const unrelatedClosed = once(unrelated, 'close');
const cleanup = cleanupClosedCodexThread({ receipt, command: process.execPath, args: [wrapper, script], cwd: root,
env: { PATH: process.env.PATH, OWNED_PIDS: metadata, TRANSCRIPT: transcript, ARCHIVE: 'hang' } });
let watchdog: ReturnType<typeof setTimeout> | undefined;
try {
// Real timeout is the resource behavior under test. A second bounded
// watchdog lets the red case reach finally and reclaim the known PIDs.
const outcome = await Promise.race([cleanup, new Promise<'still-pending'>(resolve => {
watchdog = setTimeout(() => resolve('still-pending'), 2500);
})]);
expect(await readFile(transcript, 'utf8')).toContain('thread/archive');
const owned = JSON.parse(await readFile(metadata, 'utf8')) as { wrapper: number; descendant: number };
if (detached) {
// The fixture alone knows this escaped process. Production must not
// infer its identity from a PID/name scan or claim it was reaped.
expect(outcome).toMatchObject({ status: 'failed', treeVerified: false,
message: expect.stringContaining('inherited stdio remained open') });
expect(() => process.kill(owned.descendant, 0)).not.toThrow();
expect(await waitForProcessExit(owned.wrapper, 500)).toBe(true);
} else {
expect(outcome).toEqual({ status: 'failed', message: 'codex thread cleanup timed out' });
expect(await waitForProcessExit(owned.descendant, 500)).toBe(true);
}
expect(unrelated.exitCode).toBeNull();
expect(unrelated.signalCode).toBeNull();
expect(() => process.kill(unrelated.pid!, 0)).not.toThrow();
} finally {
clearTimeout(watchdog);
const owned = JSON.parse(await readFile(metadata, 'utf8')) as { wrapper: number; descendant: number };
// Never scan processes or target a name: these exact PIDs come from this
// test-owned wrapper before any termination, including the baseline leak.
for (const pid of [owned.descendant, owned.wrapper]) {
try { process.kill(pid, 'SIGKILL'); } catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ESRCH') throw error;
}
}
await cleanup;
unrelated.kill('SIGKILL');
await unrelatedClosed;
for (const pid of [owned.descendant, owned.wrapper]) expect(await waitForProcessExit(pid, 500)).toBe(true);
await rm(root, { recursive: true, force: true });
}
});
it('does not spawn anything without a closed ownership receipt', async () => {
expect(await cleanupClosedCodexThread({ receipt: null, command: '/must-not-run', args: [], cwd: '/', env: {} }))
.toEqual({ status: 'skipped', reason: 'no-owned-closed-thread' });
});
});
@@ -1,3 +1,5 @@
import type { CollectCodexChildEvidenceInput, CodexChildEvidenceCollection } from '../src/runtimes/codex-child-evidence.js';
import type { CodexThreadCleanupResult } from '../src/agent-protocol/codex-app-server/thread-cleanup.js';
import type { Server } from 'node:http';
import { execFile } from 'node:child_process';
import { chmod, cp, mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises';
@@ -6,6 +8,7 @@ import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { Agent, fetch as undiciFetch } from 'undici';
import type {
AppliedStrategyBindingV2,
OdNextRuntimeCapabilitySnapshotV1,
@@ -18,6 +21,56 @@ import {
parseOdNextPromptBundleV2,
} from '@open-design/contracts';
const codexArchiveBoundary = vi.hoisted(() => ({
home: '', activeParentPath: '', cleanupStarted: false,
cleanupFinished: Promise.resolve(), onCleanupFinished: () => {},
order: [] as string[],
collections: [] as Array<{ input: CollectCodexChildEvidenceInput; result: CodexChildEvidenceCollection }>,
cleanupResults: [] as CodexThreadCleanupResult[],
}));
vi.mock('../src/agent-protocol/codex-app-server/thread-cleanup.js', async importOriginal => {
const actual = await importOriginal<typeof import('../src/agent-protocol/codex-app-server/thread-cleanup.js')>();
return { ...actual, cleanupClosedCodexThread: async (...args: Parameters<typeof actual.cleanupClosedCodexThread>) => {
if (!codexArchiveBoundary.home || args[0].env.CODEX_HOME !== codexArchiveBoundary.home) {
return actual.cleanupClosedCodexThread(...args);
}
codexArchiveBoundary.cleanupStarted = true;
codexArchiveBoundary.order.push('cleanup:start');
try {
const result = await actual.cleanupClosedCodexThread(...args);
codexArchiveBoundary.cleanupResults.push(result);
return result;
} finally {
codexArchiveBoundary.order.push('cleanup:end');
codexArchiveBoundary.onCleanupFinished();
}
} };
});
vi.mock('../src/runtimes/codex-child-evidence.js', async importOriginal => {
const actual = await importOriginal<typeof import('../src/runtimes/codex-child-evidence.js')>();
return { ...actual, collectCodexChildEvidence: async (...args: Parameters<typeof actual.collectCodexChildEvidence>) => {
if (!codexArchiveBoundary.home || args[0].codexHome !== codexArchiveBoundary.home) {
return actual.collectCodexChildEvidence(...args);
}
codexArchiveBoundary.order.push('collector:start');
const result = await actual.collectCodexChildEvidence(...args);
codexArchiveBoundary.collections.push({ input: args[0], result });
codexArchiveBoundary.order.push('collector:end');
return result;
} };
});
vi.mock('node:fs/promises', async importOriginal => {
const actual = await importOriginal<typeof import('node:fs/promises')>();
return { ...actual, open: async (...args: Parameters<typeof actual.open>) => {
// Only an already-started real cleanup may delay this actual file open.
// On the repaired server cleanup cannot start until collection finishes,
// so the collector never waits for an archive that cannot yet be launched.
if (codexArchiveBoundary.home && args[0] === codexArchiveBoundary.activeParentPath
&& codexArchiveBoundary.cleanupStarted) await codexArchiveBoundary.cleanupFinished;
return actual.open(...args);
} };
});
const uuidControl = vi.hoisted(() => ({ forced: [] as string[] }));
let pendingAutomaticFixtureIdentity: {
initialRunId: string;
@@ -58,6 +111,16 @@ type StartedServer = {
shutdown?: () => Promise<void> | void;
};
// Each fixture owns its HTTP transport; daemon-internal fetch remains untouched.
const fixtureHttpClients = new Map<string, { owner: StartedServer; dispatcher: Agent }>();
const fixtureShutdowns = new WeakMap<StartedServer, Promise<void>>();
function fetch(input: Parameters<typeof undiciFetch>[0], init?: Parameters<typeof undiciFetch>[1]) {
const origin = new URL(typeof input === 'string' ? input : input instanceof URL ? input.href : input.url).origin;
const client = fixtureHttpClients.get(origin);
if (!client) throw new Error(`No fixture HTTP client owns ${origin}`);
return undiciFetch(input, { ...init, dispatcher: client.dispatcher });
}
type RunStatus = {
id: string;
status: string;
@@ -2536,6 +2599,108 @@ describe('OD Next automatic production through the real server', () => {
expect(persistedEvents).not.toContain('INTERNAL_CHILD_TOOL_OUTPUT');
});
it.each(['cancel', 'normal'] as const)(
'OPEND-2960 real server preserves child evidence through %s app-server close and archives once',
async (mode) => {
const fixture = await createFixture('repair');
if (!started || !binDir) throw new Error('Expected the owned server fixture');
const ownedServer = started;
const dataDir = process.env.OD_DATA_DIR;
if (!dataDir) throw new Error('Expected the fixture data directory');
// Reuse the exact selected CLI path installed by createFixture('repair').
// Its version remains the same verified 0.147.0 tuple; only its local
// protocol implementation changes from exec-json to app-server.
const history = await writeCodexArchiveServerFixture(path.join(binDir, 'codex-repair'), binDir, mode);
process.env.OD_CODEX_TRANSPORT = 'app-server';
let releaseArchive: () => void = () => {};
const archiveFinished = new Promise<void>(resolve => { releaseArchive = resolve; });
Object.assign(codexArchiveBoundary, { home: binDir, activeParentPath: history.parentPath,
cleanupStarted: false, cleanupFinished: archiveFinished, onCleanupFinished: releaseArchive,
order: [], collections: [], cleanupResults: [] });
let runId: string | undefined;
let stopped = false;
try {
queueFixtureIds(fixture);
const created = await postRun(ownedServer.url, createRunRequest(fixture, 'Ask a question before making files.'));
if (typeof created.runId !== 'string') throw new Error('Expected the actual physical run id');
runId = created.runId;
await vi.waitFor(async () => {
expect((await readCodexArchiveFrames(history.framesPath)).some(frame => frame.method === 'fixture:ready')).toBe(true);
});
if (mode === 'cancel') {
const response = await fetch(`${ownedServer.url}/api/runs/${runId}/cancel`, { method: 'POST' });
expect(response.status, await response.text()).toBe(200);
}
const terminal = await waitForRunTerminal(ownedServer.url, runId);
expect(terminal.status).toBe(mode === 'cancel' ? 'canceled' : 'succeeded');
// The production server must actually invoke the collector. A canceled
// state or a collection made manually by this test is not sufficient.
await vi.waitFor(() => expect(codexArchiveBoundary.collections).toHaveLength(1));
const collected = codexArchiveBoundary.collections[0];
if (!collected) throw new Error('Expected the production collector result');
const task = getStrategyTaskExecution(database(), fixture.taskExecutionId);
const mapping = task?.runs.find(run => run.runId === runId);
if (!mapping) throw new Error('Expected the physical run mapping');
expect(collected.input).toMatchObject({ codexHome: binDir,
parentSessionId: history.parentId, taskExecutionId: fixture.taskExecutionId,
runId, taskRunIndex: mapping.taskRunIndex, stage: mapping.inputStage });
expect.soft(collected.result.availability).toBe('complete');
expect.soft(collected.result.knownChildCount).toBe(1);
expect.soft(collected.result.observations.filter(row => row.kind === 'child_agent' && row.status === 'completed')).toHaveLength(1);
if (mode === 'cancel') {
await vi.waitFor(() => expect(codexArchiveBoundary.cleanupResults).toHaveLength(1));
expect(codexArchiveBoundary.cleanupResults[0]).toMatchObject({ status: 'archived' });
expect.soft(codexArchiveBoundary.order.indexOf('collector:end'))
.toBeLessThan(codexArchiveBoundary.order.indexOf('cleanup:start'));
} else {
expect(codexArchiveBoundary.cleanupStarted).toBe(false);
expect(codexArchiveBoundary.cleanupResults).toEqual([]);
expect(task?.outcome).toBe('clarification_required');
}
// Drain the actual server lifecycle before checking durable external
// effects. Collector/ordering assertions above remain independent.
await stopServer(ownedServer);
stopped = true;
if (started === ownedServer) started = null;
const persisted = JSON.parse(await readFile(path.join(dataDir, 'runs', runId, 'state.json'), 'utf8')) as { id?: string; status?: string };
expect(persisted).toMatchObject({ id: runId, status: terminal.status });
const frames = await readCodexArchiveFrames(history.framesPath);
const original = frames.find(frame => frame.method === 'turn/start');
const archives = frames.filter(frame => frame.method === 'thread/archive');
if (!original) throw new Error('Expected the original app-server turn');
expect(archives).toHaveLength(1);
const archiveFrame = archives[0];
if (!archiveFrame) throw new Error('Expected one real archive request');
expect(archives[0]?.threadId).toBe(history.parentId);
expect(archives[0]?.pid === original?.pid).toBe(mode === 'normal');
if (mode === 'cancel') {
expect(frames.filter(frame => frame.pid === archiveFrame.pid).map(frame => frame.method))
.toEqual(['initialize', 'initialized', 'thread/archive']);
}
expect(await readFile(history.childPath, 'utf8')).toBe(await readFile(history.expectedChildPath, 'utf8'));
expect(await readFile(history.archivedParentPath, 'utf8')).toBe(await readFile(history.expectedParentPath, 'utf8'));
await expect(readFile(history.parentPath, 'utf8')).rejects.toMatchObject({ code: 'ENOENT' });
} finally {
// Release only the test's interleaving waiter. Real server shutdown
// owns the original child and its bounded cleanup helper lifecycle.
releaseArchive();
try {
if (!stopped) {
try {
if (runId) await fetch(`${ownedServer.url}/api/runs/${runId}/cancel`, { method: 'POST' });
} finally {
await stopServer(ownedServer);
if (started === ownedServer) started = null;
}
}
} finally {
Object.assign(codexArchiveBoundary, { home: '', activeParentPath: '', cleanupStarted: false,
cleanupFinished: Promise.resolve(), onCleanupFinished: () => {}, order: [], collections: [], cleanupResults: [] });
}
}
},
);
async function createFixture(
mode: 'repair' | 'direct' | 'complex',
{
@@ -2689,6 +2854,84 @@ describe('OD Next automatic production through the real server', () => {
}
});
type CodexArchiveFrame = { pid: number; method: string; threadId?: string };
async function readCodexArchiveFrames(file: string): Promise<CodexArchiveFrame[]> {
return (await readFile(file, 'utf8')).trim().split('\n').filter(Boolean)
.map(line => JSON.parse(line) as CodexArchiveFrame);
}
async function writeCodexArchiveServerFixture(bin: string, home: string, mode: 'cancel' | 'normal') {
const parentId = '10000000-0000-4000-8000-000000000001';
const childId = '20000000-0000-4000-8000-000000000002';
const active = path.join(home, 'sessions', '2026', '09', '12');
const archive = path.join(home, 'archived_sessions');
const parentName = `rollout-2026-09-12-${parentId}.jsonl`;
const childName = `rollout-2026-09-12-${childId}.jsonl`;
const framesPath = path.join(home, 'archive-server-frames.jsonl');
const childPath = path.join(active, childName);
const parentPath = path.join(active, parentName);
const expectedChildPath = path.join(home, 'expected-child.jsonl');
const expectedParentPath = path.join(home, 'expected-parent.jsonl');
await writeFile(bin, `#!/usr/bin/env node
const fs = require('node:fs');
const path = require('node:path');
const readline = require('node:readline');
const argv = process.argv.slice(2);
if (argv.includes('--version')) { console.log('codex-cli 0.147.0'); process.exit(0); }
if (argv.includes('--help')) { console.log('Usage: codex app-server'); process.exit(0); }
if (!argv.includes('app-server')) process.exit(9);
const home = ${JSON.stringify(home)};
if (process.env.CODEX_HOME !== home) process.exit(10);
const active = ${JSON.stringify(active)}, archive = ${JSON.stringify(archive)};
const parentId = ${JSON.stringify(parentId)}, childId = ${JSON.stringify(childId)};
const parentPath = ${JSON.stringify(parentPath)}, childPath = ${JSON.stringify(childPath)};
const send = frame => process.stdout.write(JSON.stringify(frame) + '\\n');
const record = (method, threadId) => fs.appendFileSync(${JSON.stringify(framesPath)}, JSON.stringify({ pid: process.pid, method, ...(threadId ? {threadId} : {}) }) + '\\n');
const input = readline.createInterface({ input: process.stdin });
input.on('line', line => {
const frame = JSON.parse(line);
record(frame.method, frame.params?.threadId);
if (frame.method === 'initialize') send({ id: frame.id, result: { userAgent: 'codex/0.147.0' } });
else if (frame.method === 'thread/start') send({ id: frame.id, result: { thread: { id: parentId, historyMode: 'paginated', path: parentPath } } });
else if (frame.method === 'turn/start') {
const now = Date.now();
const row = (offset, payload) => ({ timestamp: new Date(now + offset).toISOString(), type: 'event_msg', payload });
const parent = [
{ timestamp: new Date(now).toISOString(), type: 'session_meta', payload: { id: parentId } },
row(-6, { type: 'task_started', turn_id: 'parent-turn' }),
row(-5, { type: 'sub_agent_activity', agent_thread_id: childId, kind: 'started', occurred_at_ms: now - 5 }),
row(-1, { type: 'sub_agent_activity', agent_thread_id: childId, kind: 'completed', occurred_at_ms: now - 1 }),
row(0, { type: 'task_complete', turn_id: 'parent-turn' }),
];
fs.mkdirSync(active, { recursive: true, mode: 0o700 });
fs.mkdirSync(archive, { recursive: true, mode: 0o700 });
fs.writeFileSync(parentPath, parent.map(row => JSON.stringify(row)).join('\\n') + '\\n', { mode: 0o600 });
fs.writeFileSync(${JSON.stringify(expectedParentPath)}, fs.readFileSync(parentPath), { mode: 0o600 });
const child = [
{ timestamp: new Date(now - 4).toISOString(), type: 'session_meta', payload: { id: childId, parent_thread_id: parentId } },
row(-4, { type: 'task_started', turn_id: 'child-turn' }),
row(-3, { type: 'user_message', message: 'Read the scoped child evidence' }),
row(-2, { type: 'task_complete', turn_id: 'child-turn' }),
].map(row => JSON.stringify(row)).join('\\n') + '\\n';
fs.writeFileSync(childPath, child, { mode: 0o600 });
fs.writeFileSync(${JSON.stringify(expectedChildPath)}, child, { mode: 0o600 });
send({ id: frame.id, result: { turn: { id: 'parent-turn' } } });
send({ method: 'turn/started', params: { threadId: parentId, turn: { id: 'parent-turn', status: 'inProgress' } } });
send({ method: 'item/agentMessage/delta', params: { threadId: parentId, turnId: 'parent-turn', itemId: 'answer', delta: '<question-form id="archive-check">{"questions":[{"id":"audience","type":"text","label":"Audience?","required":true}]}</question-form>' } });
record('fixture:ready', parentId);
if (${JSON.stringify(mode)} === 'normal') send({ method: 'turn/completed', params: { threadId: parentId, turn: { id: 'parent-turn', status: 'completed', items: [] } } });
} else if (frame.method === 'turn/interrupt') process.exit(130);
else if (frame.method === 'thread/archive') {
if (frame.params?.threadId !== parentId) process.exit(11);
fs.renameSync(parentPath, path.join(archive, path.basename(parentPath)));
send({ id: frame.id, result: {} });
}
});
`, 'utf8');
await chmod(bin, 0o755);
return { parentId, framesPath, parentPath, childPath, expectedChildPath, expectedParentPath, archivedParentPath: path.join(archive, parentName) };
}
async function createPublicRolloutFixture(
label: string,
conversationMode: 'design' | 'chat' | 'plan' = 'chat',
@@ -2898,17 +3141,35 @@ async function startDaemon(
() => EXECUTION_PREFLIGHT,
complexResolver: StartServerOptions['odNextComplexProductionResolver'] = null,
): Promise<StartedServer> {
return await startServer({
const started = await startServer({
port: 0,
returnServer: true,
odNextExecutionPreflightResolver: resolver,
odNextComplexProductionResolver: complexResolver,
}) as StartedServer;
fixtureHttpClients.set(new URL(started.url).origin, { owner: started, dispatcher: new Agent() });
return started;
}
async function stopServer(server: StartedServer | null): Promise<void> {
if (!server) return;
function stopServer(server: StartedServer | null): Promise<void> {
if (!server) return Promise.resolve();
const existing = fixtureShutdowns.get(server);
if (existing) return existing;
const pending = stopOwnedServer(server);
fixtureShutdowns.set(server, pending);
return pending;
}
async function stopOwnedServer(server: StartedServer): Promise<void> {
await Promise.resolve(server.shutdown?.());
const origin = new URL(server.url).origin;
const client = fixtureHttpClients.get(origin);
if (client?.owner === server) {
fixtureHttpClients.delete(origin);
// Runs have settled first. Release only this fixture's keep-alive clients,
// including a connection whose idle transition races server.close().
await client.dispatcher.destroy();
}
if (server.server.listening) {
await new Promise<void>((resolve) => server.server.close(() => resolve()));
}
@@ -1,6 +1,7 @@
import { createHash } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { chmod, mkdir, mkdtemp, rm, symlink, writeFile } from 'node:fs/promises';
import { readFileSync, writeFileSync } from 'node:fs';
import Database from 'better-sqlite3';
import { chmod, mkdir, mkdtemp, rename, rm, symlink, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
@@ -175,6 +176,85 @@ afterEach(async () => {
});
describe('collectCodexChildEvidence', () => {
it('retains parent and declared child evidence after terminal archiving', async () => {
const home = await codexHome();
const parentPath = await writeRollout(home, PARENT, [
metadata(PARENT),
...turn({ id: 'parent-turn', startedAtMs: 100, terminal: 'none', childActivities: [
{ sessionId: CHILD, kind: 'started', atMs: 1_000 },
{ sessionId: CHILD, kind: 'completed', atMs: 6_000 },
] }),
event(7_000, { type: 'task_complete', turn_id: 'parent-turn' }),
]);
const childPath = await writeRollout(home, CHILD, [
metadata(CHILD, PARENT),
...turn({ id: 'child-turn', startedAtMs: 2_000, prompt: 'Audit child evidence' }),
]);
const before = await collectCodexChildEvidence(collectInput(home));
expect(before.availability).toBe('complete');
expect(before.observations.filter(o => o.kind === 'child_agent' && o.status === 'completed')).toHaveLength(1);
// thread/archive moves both logs before server collects evidence on close.
const archive = path.join(home, 'archived_sessions');
await mkdir(archive);
for (const filePath of [parentPath, childPath]) {
await rename(filePath, path.join(archive, path.basename(filePath)));
}
const after = await collectCodexChildEvidence(collectInput(home));
expect(after.availability).toBe(before.availability);
expect(after.observations).toEqual(before.observations);
});
it('retains indexed parent and child evidence in an archive larger than the scan limit', async () => {
const home = await codexHome();
const parentPath = await writeRollout(home, PARENT, [
metadata(PARENT),
...turn({ id: 'parent-turn', startedAtMs: 100, childActivities: [
{ sessionId: CHILD, kind: 'started', atMs: 200 },
] }),
]);
const childPath = await writeRollout(home, CHILD, [
metadata(CHILD, PARENT),
...turn({ id: 'child-turn', startedAtMs: 300, prompt: 'Retain this child' }),
]);
const before = await collectCodexChildEvidence(collectInput(home));
expect(before.availability).toBe('complete');
const archive = path.join(home, 'archived_sessions');
await mkdir(archive);
const db = new Database(path.join(home, 'state_5.sqlite'));
try {
db.exec('CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, archived INTEGER NOT NULL)');
for (const [id, filePath] of [[PARENT, parentPath], [CHILD, childPath]] as const) {
const archivedPath = path.join(archive, path.basename(filePath));
await rename(filePath, archivedPath);
db.prepare('INSERT INTO threads VALUES (?, ?, 1)').run(id, archivedPath);
}
} finally { db.close(); }
for (let i = 0; i < 4_097; i += 1) writeFileSync(path.join(archive, `unrelated-${i}`), '');
const after = await collectCodexChildEvidence(collectInput(home));
expect(after.availability).toBe(before.availability);
expect(after.observations).toEqual(before.observations);
});
it('rejects ambiguous active/archive copies and ignores unrelated archived threads', async () => {
const home = await codexHome();
const records = [metadata(PARENT), ...turn({ id: 'parent-turn', startedAtMs: 100 })];
const parentPath = await writeRollout(home, PARENT, records);
const archive = path.join(home, 'archived_sessions');
await mkdir(archive);
const archivedParent = path.join(archive, path.basename(parentPath));
await writeFile(archivedParent, readFileSync(parentPath));
const db = new Database(path.join(home, 'state_5.sqlite'));
try {
db.exec('CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, archived INTEGER NOT NULL)');
db.prepare('INSERT INTO threads VALUES (?, ?, 1)').run(PARENT, archivedParent);
} finally { db.close(); }
const ambiguous = await collectCodexChildEvidence(collectInput(home));
expect(ambiguous.availability).toBe('unavailable');
expect(ambiguous.diagnostics).toContainEqual({ code: 'rollout_ambiguous', count: 1 });
await rename(archivedParent, path.join(archive, `rollout-unrelated-${CHILD}.jsonl`));
expect((await collectCodexChildEvidence(collectInput(home))).availability).toBe('complete');
});
it('replays the exact Codex 0.147.0 best-effort success, started failure, and cancel records', async () => {
const seed = JSON.parse(readFileSync(sanitizedRealSeedPath, 'utf8')) as {
fixtureKind: string;
+2
View File
@@ -28,6 +28,7 @@ export {
export type {
ProcessSnapshot,
ProcessSnapshotOptions,
ProcessStampContract,
ProcessStampField,
ProcessStampShape,
@@ -54,6 +55,7 @@ export {
readProcessStamp,
readProcessStampFromCommand,
signalProcesses,
selectOwnedProcessTree,
spawnBackgroundProcess,
spawnLoggedProcess,
stopProcesses,
+42 -7
View File
@@ -16,6 +16,17 @@ import { setTimeout as sleep } from "node:timers/promises";
import { createCommandInvocation, type CommandInvocationRequest } from "./command.js";
/** Opt-in query budget; existing snapshot callers retain their current defaults. */
export type ProcessSnapshotOptions = { timeoutMs?: number };
function snapshotBudget(options: ProcessSnapshotOptions): { timeout?: number; killSignal?: "SIGKILL" } {
if (options.timeoutMs === undefined) return {};
if (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs <= 0) {
throw new RangeError("Process snapshot timeout must be a positive integer");
}
return { timeout: options.timeoutMs, killSignal: "SIGKILL" };
}
export type ProcessStampShape = object;
export type ProcessStampField<TStamp extends ProcessStampShape> = Extract<keyof TStamp, string>;
@@ -348,9 +359,9 @@ function parsePsOutput(stdout: string): ProcessSnapshot[] {
}
/** @internal Enumerate process snapshots on POSIX via `ps`. */
async function listPosixProcessSnapshots(): Promise<ProcessSnapshot[]> {
async function listPosixProcessSnapshots(options: ProcessSnapshotOptions = {}): Promise<ProcessSnapshot[]> {
const stdout = await new Promise<string>((resolveList, rejectList) => {
execFile("ps", ["-axo", "pid=,ppid=,command="], { encoding: "utf8", maxBuffer: 8 * 1024 * 1024 }, (error, out) => {
execFile("ps", ["-axo", "pid=,ppid=,command="], { encoding: "utf8", maxBuffer: 8 * 1024 * 1024, ...snapshotBudget(options) }, (error, out) => {
if (error) rejectList(error);
else resolveList(out);
});
@@ -359,13 +370,13 @@ async function listPosixProcessSnapshots(): Promise<ProcessSnapshot[]> {
}
/** @internal Enumerate process snapshots on Windows via `Get-CimInstance Win32_Process` JSON. */
async function listWindowsProcessSnapshots(): Promise<ProcessSnapshot[]> {
async function listWindowsProcessSnapshots(options: ProcessSnapshotOptions = {}): Promise<ProcessSnapshot[]> {
const command = [
"$ErrorActionPreference = 'Stop'",
"Get-CimInstance Win32_Process | Select-Object ProcessId, ParentProcessId, CommandLine, @{Name='StartedAtMs';Expression={([DateTimeOffset]$_.CreationDate).ToUnixTimeMilliseconds()}} | ConvertTo-Json -Compress",
].join("; ");
const stdout = await new Promise<string>((resolveList, rejectList) => {
execFile("powershell.exe", ["-NoProfile", "-NonInteractive", "-Command", command], { encoding: "utf8", maxBuffer: 8 * 1024 * 1024 }, (error, out) => {
execFile("powershell.exe", ["-NoProfile", "-NonInteractive", "-Command", command], { encoding: "utf8", maxBuffer: 8 * 1024 * 1024, ...snapshotBudget(options) }, (error, out) => {
if (error) rejectList(error);
else resolveList(out);
});
@@ -422,10 +433,10 @@ export function parseWindowsProcessSnapshots(stdout: string): ProcessSnapshot[]
* empty snapshot. Mutation paths use this strict form so discovery failure can
* never be mistaken for an already-stopped process set.
*/
export async function captureProcessSnapshot(): Promise<ProcessSnapshot[]> {
export async function captureProcessSnapshot(options: ProcessSnapshotOptions = {}): Promise<ProcessSnapshot[]> {
return process.platform === "win32"
? await listWindowsProcessSnapshots()
: await listPosixProcessSnapshots();
? await listWindowsProcessSnapshots(options)
: await listPosixProcessSnapshots(options);
}
/**
@@ -610,6 +621,30 @@ export function collectProcessTreePids(
return [...visited].sort((left, right) => right - left);
}
/**
* Revalidate known Windows process generations before extending their tree.
* The caller supplies identities captured while its owned child was alive;
* this function never upgrades a bare root PID into an ownership proof.
*/
export function selectOwnedProcessTree(known: ProcessSnapshot[], current: ProcessSnapshot[]): ProcessSnapshot[] {
const validTime = (value: number | undefined): value is number =>
typeof value === "number" && Number.isSafeInteger(value) && value > 0;
const currentByPid = new Map(current.map(entry => [entry.pid, entry]));
const selected = new Map<number, ProcessSnapshot>();
const queue = known.flatMap(entry => {
const live = currentByPid.get(entry.pid);
return live && validTime(entry.startedAtMs) && live.startedAtMs === entry.startedAtMs ? [live] : [];
});
for (let index = 0; index < queue.length; index++) {
const parent = queue[index]!;
if (selected.has(parent.pid)) continue;
selected.set(parent.pid, parent);
queue.push(...current.filter(entry => entry.ppid === parent.pid
&& validTime(entry.startedAtMs) && entry.startedAtMs >= parent.startedAtMs!));
}
return [...selected.values()];
}
/** Send a signal to each PID, ignoring `ESRCH` (already-dead) but rethrowing other errors. */
export function signalProcesses(pids: number[], signal: NodeJS.Signals): void {
for (const pid of pids) {
@@ -0,0 +1,55 @@
import { once } from "node:events";
import type { ChildProcess, ExecFileOptions, ExecFileOptionsWithStringEncoding } from "node:child_process";
import { afterEach, describe, expect, it, vi } from "vitest";
const state = vi.hoisted(() => ({ mode: "hang", child: undefined as ChildProcess | undefined,
calls: [] as Array<{ command: string; options: ExecFileOptions }> }));
vi.mock("node:child_process", async (load) => {
const actual = await load<typeof import("node:child_process")>();
return { ...actual, execFile: ((command: string, _args: string[], options: ExecFileOptionsWithStringEncoding,
callback: (error: Error | null, stdout: string, stderr: string) => void) => {
state.calls.push({ command, options });
// Real bounded child, with the Windows query command intercepted only at
// the OS launch boundary. No CIM claim is made by this cross-platform test.
state.child = actual.execFile(process.execPath, ["-e", state.mode === "hang"
? "setInterval(() => {}, 1000)"
: 'process.stdout.write(JSON.stringify([{ ProcessId: 80, ParentProcessId: 1, CommandLine: "owned", StartedAtMs: 1234 }]))'],
options, callback);
return state.child;
}) as typeof actual.execFile };
});
import { captureProcessSnapshot } from "../src/process.js";
const platformDescriptor = Object.getOwnPropertyDescriptor(process, "platform")!;
afterEach(() => { Object.defineProperty(process, "platform", platformDescriptor); state.calls = []; });
describe("explicit Windows snapshot budget", () => {
it("kills the actual query child and rejects within a caller-supplied budget", async () => {
Object.defineProperty(process, "platform", { value: "win32", configurable: true });
state.mode = "hang";
const capture = captureProcessSnapshot as (options?: { timeoutMs: number }) => Promise<unknown>;
const result = capture({ timeoutMs: 50 }).then(() => "resolved", () => "rejected");
let watchdog: ReturnType<typeof setTimeout> | undefined;
try {
const observed = await Promise.race([result, new Promise<string>(resolve => {
watchdog = setTimeout(() => resolve("still-pending"), 500);
})]);
expect(observed).toBe("rejected");
expect(state.calls[0]?.command).toBe("powershell.exe");
expect(state.child?.exitCode !== null || state.child?.signalCode !== null).toBe(true);
} finally {
clearTimeout(watchdog);
if (state.child && state.child.exitCode === null && state.child.signalCode === null) {
const closed = once(state.child, "close"); state.child.kill("SIGKILL"); await closed;
}
await result;
}
});
it("keeps existing default callers and Windows parsing unchanged", async () => {
Object.defineProperty(process, "platform", { value: "win32", configurable: true });
state.mode = "success";
expect(await captureProcessSnapshot()).toEqual([{ pid: 80, ppid: 1, command: "owned", startedAtMs: 1234 }]);
expect(state.calls[0]?.options.timeout).toBeUndefined();
});
});
@@ -4,6 +4,7 @@ import { describe, expect, it } from "vitest";
import {
collectProcessTreePids,
selectOwnedProcessTree,
processCommandExactlyRunsExecutable,
stopProcesses,
type ProcessStampContract,
@@ -202,3 +203,32 @@ describe("stopProcesses", () => {
5_000,
);
});
describe("generation-fenced owned process trees", () => {
const processAt = (pid: number, ppid: number, startedAtMs?: number): ProcessSnapshot => ({
pid, ppid, command: "same executable", ...(startedAtMs === undefined ? {} : { startedAtMs }),
});
const root = processAt(10, 1, 100);
const child = processAt(11, 10, 101);
it("extends only a proven live ancestor and ignores same-name siblings", () => {
expect(selectOwnedProcessTree([root], [root, child, processAt(12, 11, 102), processAt(20, 1, 101)])
.map(entry => entry.pid)).toEqual([10, 11, 12]);
});
it("retains a known child after the wrapper exits, without trusting a reused wrapper PID", () => {
const replacement = processAt(10, 1, 200);
expect(selectOwnedProcessTree([root, child], [replacement, child, processAt(12, 11, 102), processAt(21, 10, 201)])
.map(entry => entry.pid)).toEqual([11, 12]);
});
it("rejects a reused descendant PID and that replacement's children", () => {
expect(selectOwnedProcessTree([root, child], [processAt(11, 1, 200), processAt(12, 11, 201)])).toEqual([]);
});
it("rejects unknown creation times and children older than their claimed parent", () => {
expect(selectOwnedProcessTree([root], [root, processAt(11, 10), processAt(12, 10, 99)])).toEqual([root]);
expect(selectOwnedProcessTree([processAt(10, 1)], [root, child])).toEqual([]);
});
it("does not discover an orphan from a root PID that was never captured", () => {
expect(selectOwnedProcessTree([], [child])).toEqual([]);
expect(selectOwnedProcessTree([root], [child])).toEqual([]);
});
});