Flaky CI at the cause: task exit verdict, race-free private-root removal, stalled-file teardown

scripts/test-file-teardown.mjs

@@ -0,0 +1,49 @@
+import {after} from 'node:test';
+import {spawnSync} from 'node:child_process';
+
+/**
+ * Preloaded into every test file's process by scripts/test.mjs. A test that stalls past its timeout is cancelled, but
+ * whatever it started keeps running: a live shell PTY or a piped child then keeps the file's process alive, the file
+ * never finishes, its failure is never printed, and a CI job waits for its own time limit (#415's Ubuntu shard).
+ * Once the file's tests are done, a timer that cannot itself keep the process alive gives the file's own cleanup time
+ * (NMSH_TEST_TEARDOWN_MS, default 15 s); if the process is still alive then, every process the file left running is
+ * ended and named, so the stall is reported instead of hanging the run.
+ */
+if (process.env.NODE_TEST_CONTEXT) {
+  after(() => {
+    const delay = Number(process.env.NMSH_TEST_TEARDOWN_MS) || 15_000;
+    setTimeout(endLeftovers, delay).unref();
+  });
+}
+
+export function endLeftovers() {
+  const rows = processTable();
+  const own = rows.find(row => row.pid === process.pid)?.group;
+  const left = descendants(rows, process.pid).filter(row => !/(?:^|\/)(?:esbuild|ps)$/u.test(row.command));
+  for (const row of left) {
+    // A leader of its own group (a PTY shell, a detached helper) takes its whole group with it; never our own group.
+    const target = row.group === row.pid && row.group !== own ? -row.group : row.pid;
+    try { process.kill(target, 'SIGKILL'); } catch { /* Already gone. */ }
+  }
+  if (left.length) process.stderr.write(`test teardown: ended ${left.length} process${left.length === 1 ? '' : 'es'} this file left running: ${left.map(row => `${row.pid} ${row.command}`).join(', ')}\n`);
+  return left;
+}
+
+function processTable() {
+  const result = spawnSync('ps', ['-A', '-o', 'pid=,ppid=,pgid=,comm='], {encoding: 'utf8', timeout: 5000, maxBuffer: 8 * 1024 * 1024});
+  if (result.status !== 0 || !result.stdout) return [];
+  return result.stdout.trim().split('\n').map(line => {
+    const [pid, parent, group, ...command] = line.trim().split(/\s+/u);
+    return {pid: Number(pid), parent: Number(parent), group: Number(group), command: command.join(' ')};
+  });
+}
+
+function descendants(rows, root) {
+  const found = [];
+  const queue = [root];
+  while (queue.length) {
+    const parent = queue.shift();
+    for (const row of rows) if (row.parent === parent && row.pid !== root) { found.push(row); queue.push(row.pid); }
+  }
+  return found;
+}

scripts/test.mjs

@@ -27,7 +27,9 @@ export async function runTestFiles(files, args = [], {cwd = process.cwd(), stdio
     // PTY workers also own shell/helper processes. Bound fan-out on larger
     // development hosts; explicit Node --test-concurrency options still win.
     const concurrency = testConcurrency(process.env.NMSH_TEST_CONCURRENCY, availableParallelism());
-    const child = spawn(process.execPath, ['--import=tsx', '--test', `--test-concurrency=${concurrency}`, ...args, ...files], {
+    // Each file's process ends what it left running once its tests are done (a stalled test cannot hang the run).
+    const teardown = new URL('./test-file-teardown.mjs', import.meta.url).href;
+    const child = spawn(process.execPath, ['--import=tsx', `--import=${teardown}`, '--test', `--test-concurrency=${concurrency}`, ...args, ...files], {
       cwd, stdio, env,
     });
     let interrupted = false;

src/backgroundTasks/supervisor.ts

@@ -79,12 +79,16 @@ interface LiveTask {
   /** The live guard owns signalling from inside its group; no numeric group id is retained. */
   group?: TaskGroup;
   leaderExited?: boolean;
+  /** Since when the leader's exit waits for the guard's verdict on the rest of the group (performance.now()). */
+  verdictSince?: number;
   pending?: ReturnType<typeof setTimeout>;
   /** Observes the command leader separately from the guard's eventual exit. */
   watch?: ReturnType<typeof setInterval>;
 }
 
 const ORDER: readonly StopLevel[] = ['interrupt', 'terminate', 'kill'];
+/** How long a finished leader waits for the guard's verdict when it cannot get one (its `ps` keeps failing). */
+const VERDICT_MS = 3000;
 
 export class BackgroundTaskSupervisor {
   private readonly tasks = new Map<string, LiveTask>();
@@ -124,11 +128,20 @@ export class BackgroundTaskSupervisor {
     this.tasks.set(id, task);
     pty.onData(data => { task.output.append(data); this.changed(task); });
     const observeLeader = () => {
-      if (task.leaderExited) return;
-      const exit = task.group?.leaderExit();
-      if (exit) {
+      if (!task.leaderExited) {
+        const exit = task.group?.leaderExit();
+        if (!exit) return;
+        task.leaderExited = true;
+        this.exited(task, exit.exitCode, exit.signal);
+      }
+      // The command has exited. The guard decides how it ended: it exits when nothing else of the group runs (onExit
+      // below), or says that something does. Only if it cannot tell in time is something reported as left.
+      if (task.verdictSince === undefined) { clearInterval(task.watch); task.watch = undefined; return; }
+      if (task.group?.leftBehind() || performance.now() - task.verdictSince > VERDICT_MS) {
+        task.verdictSince = undefined;
         clearInterval(task.watch); task.watch = undefined;
-        task.leaderExited = true; this.exited(task, exit.exitCode, exit.signal);
+        if (task.record.stopRequested) { task.record.leftovers = true; this.changed(task, true); }
+        else this.finish(task);
       }
     };
     task.watch = setInterval(observeLeader, 25);
@@ -140,7 +153,7 @@ export class BackgroundTaskSupervisor {
       clearInterval(task.watch);
       task.watch = undefined;
       if (!task.leaderExited) this.exited(task, exitCode, signal);
-      else if (task.record.state === 'stopping') this.finish(task);
+      else if (task.record.state === 'stopping' || task.verdictSince !== undefined) { task.verdictSince = undefined; this.finish(task); }
       else { task.record.leftovers = false; this.changed(task, true); }
     });
     this.changed(task, true);
@@ -236,6 +249,8 @@ export class BackgroundTaskSupervisor {
       this.changed(task, true);
       return;
     }
+    // Whether anything else of the group still runs is not known yet: wait for the guard's verdict, never guess.
+    if (task.group?.alive === true) { task.verdictSince = performance.now(); return; }
     this.finish(task);
   }
 

src/backgroundTasks/taskGroup.ts

@@ -53,6 +53,11 @@ export class TaskGroup {
     return undefined;
   }
 
+  /** The guard found members of the group still running after the command's leader exited. */
+  leftBehind(): boolean {
+    try { return readFileSync(join(this.directory, 'left'), 'utf8') === '1'; } catch { return false; }
+  }
+
   close(): void {
     this.alive = false;
     rmSync(this.directory, {recursive: true, force: true});

src/backgroundTasks/taskGroupGuard.ts

@@ -35,6 +35,7 @@ const saveExit = (exitCode: number, signal?: number) => {
 child.on('error', () => saveExit(127));
 child.on('exit', (code, signal) => saveExit(code ?? 0, signal ? constants.signals[signal] : undefined));
 let lastProbe = -Infinity;
+let reported = false;
 const poll = setInterval(() => {
   // An orphaned guard owns its group, never the former parent's possibly recycled PID.
   if (process.ppid !== parent) { rmSync(directory, {recursive: true, force: true}); process.kill(0, 'SIGKILL'); }
@@ -49,4 +50,9 @@ const poll = setInterval(() => {
     return Number(pgid) === process.pid && Number(pid) !== process.pid && Number(ppid) !== process.pid && !state?.startsWith('Z');
   });
   if (!members) { clearInterval(poll); process.exit(0); }
+  // Say once that members remain, so the supervisor reports them now instead of waiting for this guard to exit.
+  if (!reported) {
+    reported = true;
+    try { writeFileSync(`${directory}/left-next`, '1', {mode: 0o600}); renameSync(`${directory}/left-next`, `${directory}/left`); } catch { /* The supervisor's bound applies. */ }
+  }
 }, 25);

src/shell/ConfiguredCompletion.ts

@@ -4,6 +4,7 @@ import {mkdtempSync, writeFileSync, readFileSync, rmSync, existsSync, statSync}
 import {tmpdir} from 'node:os';
 import {dirname, join} from 'node:path';
 import {fileURLToPath} from 'node:url';
+import {removeDirectoryNow} from '../util/removeDirectory.js';
 import {commandIdentity, completionLabel, type CompletionCandidate, type CompletionContext, type CompletionSource} from './completion.js';
 
 const MAX_BYTES = 1024 * 1024;
@@ -124,7 +125,8 @@ export class ConfiguredCompletionSource implements CompletionSource {
     }
     this.wait?.resolve(false);
     this.wait = undefined;
-    if (this.root) rmSync(this.root, {recursive: true, force: true});
+    // The helper's own cleanup may be removing it, and a shell may still write its pid file, at this moment.
+    if (this.root) removeDirectoryNow(this.root);
     this.root = undefined;
     this.output = '';
   }

src/util/removeDirectory.ts

@@ -0,0 +1,50 @@
+import {mkdtempSync, readdirSync, renameSync, rmdirSync, rmSync} from 'node:fs';
+import {basename, dirname, join} from 'node:path';
+
+/**
+ * Remove a private directory that helper processes may still be writing into or removing themselves. `rmSync` alone is
+ * not enough then: a file created during its walk made it throw ENOTEMPTY, and on Linux a concurrent `rm -rf` made it
+ * return with the directory still there. Renaming the directory away first (atomic, same filesystem) leaves no path to
+ * it for anyone to write into, so the path is gone when this returns; the renamed copy is then removed. Never throws.
+ */
+export function removeDirectoryNow(path: string): void {
+  let grave: string | undefined;
+  try {
+    grave = mkdtempSync(join(dirname(path), `${basename(path)}-gone-`));
+    renameSync(path, join(grave, 'directory'));
+  } catch { /* Already gone, or not movable: removed where it is below. */ }
+  for (const target of grave ? [grave, path] : [path]) {
+    if (!removeTree(target)) {
+      setTimeout(() => { try { rmSync(target, {recursive: true, force: true}); } catch { /* Best effort. */ } }, 100).unref();
+    }
+  }
+}
+
+/**
+ * Remove a directory tree while another remover may be deleting inside it. A remover that already holds the directory
+ * open (`rm -rf` works through directory descriptors, so it follows a rename) makes entries vanish mid-walk, and one
+ * rmSync then stops early without an error. Each pass removes what is listed, tolerating what vanished, and retries
+ * the directory itself while it is not empty; nothing can add entries (its path is gone), so every pass shrinks it.
+ */
+function removeTree(target: string): boolean {
+  for (let pass = 0; pass < 50; pass += 1) {
+    let names: string[];
+    try { names = readdirSync(target); }
+    catch (error) {
+      const code = (error as NodeJS.ErrnoException).code;
+      if (code === 'ENOENT') return true;
+      if (code !== 'ENOTDIR') return false;
+      try { rmSync(target, {force: true}); return true; } catch { return false; }
+    }
+    for (const name of names) {
+      try { rmSync(join(target, name), {recursive: true, force: true}); } catch { /* Vanished or retried next pass. */ }
+    }
+    try { rmdirSync(target); return true; }
+    catch (error) {
+      const code = (error as NodeJS.ErrnoException).code;
+      if (code === 'ENOENT') return true;
+      if (code !== 'ENOTEMPTY' && code !== 'EEXIST' && code !== 'EBUSY') return false;
+    }
+  }
+  return false;
+}

tests/backgroundSupervisor.test.ts

@@ -109,6 +109,18 @@ test('a pipeline stops as a whole; a detached helper it started (like the adb se
   });
 });
 
+test('a task that leaves nothing behind is never reported as having processes left, even for a moment; it is removable once exited', async () => {
+  await withSupervisor(async (supervisor, cwd, changes) => {
+    for (let run = 0; run < 5; run += 1) {
+      const {task} = supervisor.start({command: `printf 'done ${run}\\n'`, cwd, shell: SHELL, env: env()}) as {ok: true; task: TaskRecord};
+      await waitFor(() => supervisor.get(task.id)!.state === 'exited', 'exited');
+      assert.deepEqual(changes.filter(change => change.id === task.id && change.leftovers).map(change => change.state), [],
+        'the guard still being alive is not a process left behind');
+      assert.equal(supervisor.remove(task.id), true, 'exited with nothing left: removable at once');
+    }
+  });
+});
+
 test('processes left in the group after the leader exits are reported, and killed only on request', async () => {
   await withSupervisor(async (supervisor, cwd) => {
     const {task} = supervisor.start({command: afterSleeperReady(cwd, '60000', 'sleep 1; exit 0'), cwd, shell: SHELL, env: env()}) as {ok: true; task: TaskRecord};

tests/codexAdapter.test.ts

@@ -33,7 +33,13 @@ test('messages: items, approvals and turn results normalize from structured fiel
   }
 });
 
-async function session(scenario: string, act: (session: CodexSession, events: AgentEvent[], wait: (kind: AgentEvent['kind'], count?: number) => Promise<void>) => Promise<void>) {
+/**
+ * Runs one scenario and returns what the client sent, as the fake server recorded it. The server records a message when
+ * it reads it, which can be after the turn settles (a refusal written after the person's answer), so `recorded` says
+ * what the test needs on record; it is waited for (bounded) rather than read at whatever moment the turn settled.
+ */
+async function session(scenario: string, act: (session: CodexSession, events: AgentEvent[], wait: (kind: AgentEvent['kind'], count?: number) => Promise<void>) => Promise<void>,
+  recorded: (client: Array<Record<string, unknown>>) => boolean = () => true) {
   const directory = await mkdtemp(join(tmpdir(), 'nmsh-codex-'));
   const log = join(directory, 'client.log');
   const events: AgentEvent[] = [];
@@ -50,7 +56,9 @@ async function session(scenario: string, act: (session: CodexSession, events: Ag
   try {
     assert.deepEqual(codex.start(), {ok: true});
     await act(codex, events, wait);
-    const client = (await readFile(log, 'utf8').catch(() => '')).trim().split('\n').filter(Boolean).map(line => JSON.parse(line) as Record<string, unknown>);
+    const read = async () => (await readFile(log, 'utf8').catch(() => '')).trim().split('\n').filter(Boolean).map(line => JSON.parse(line) as Record<string, unknown>);
+    let client = await read();
+    for (const deadline = Date.now() + 5000; !recorded(client) && Date.now() < deadline; client = await read()) await new Promise(resolve => setTimeout(resolve, 20));
     return client;
   } finally {
     codex.close();
@@ -78,7 +86,7 @@ test('a session: handshake, thread, a message, one approval answered by the pers
     ]);
     assert.deepEqual(events.filter(event => event.kind === 'assistant'), [{kind: 'assistant', text: 'Cleaned.'}]);
     assert.deepEqual(events.at(-1), {kind: 'settled', ok: true});
-  });
+  }, client => client.some(message => message.id === 100) && client.some(message => message.id === 101));
   // The two responses race (the refusal is immediate, the approval waits for the person), so only the first four are ordered.
   assert.deepEqual(client.slice(0, 4).map(message => message.method), ['initialize', 'initialized', 'thread/start', 'turn/start']);
   assert.equal(client.length, 6);

tests/configuredCompletion.test.ts

@@ -10,7 +10,9 @@ import {CompletionService} from '../src/shell/CompletionService.js';
 test('real configured candidate capture is bounded to 4096 records', async () => {
   const home = mkdtempSync(join(tmpdir(), 'nmsh-large-completion-'));
   writeFileSync(join(home, '.zshrc'), 'autoload -Uz compinit\ncompinit -D\n_large() { compadd -J large -- value{1..5000}; }\ncompdef _large large\n');
-  const source = new ConfiguredCompletionSource({env: {...process.env, HOME: home}, startupMs: 4000, queryMs: 2000});
+  // This bounds the records, not startup: the helper's own startup cap (5 s) and a long query window keep a loaded
+  // machine's compinit from turning into an empty result.
+  const source = new ConfiguredCompletionSource({env: {...process.env, HOME: home}, startupMs: 5000, queryMs: 2000});
   try {
     const values = await source.query({buffer: 'large v', cwd: home}, new AbortController().signal);
     assert.equal(values.length, 4096);

tests/directoryNavigation.test.ts

@@ -46,7 +46,9 @@ test('zoxide query uses a temporary copy, caches results and falls back on missi
   try {
     await writeFile(join(directory, 'db.zo'), 'original', {mode: 0o600});
     const binary = join(directory, 'zoxide');
-    await writeFile(binary, `#!${process.execPath}\nconst fs=require('node:fs');if(process.env._ZO_DATA_DIR===process.env.ORIGINAL_DB)process.exit(2);fs.writeFileSync(process.env._ZO_DATA_DIR+'/db.zo','sorted');fs.appendFileSync(process.env.CALLS,JSON.stringify(process.argv.slice(2))+'\\n');process.stdout.write('20.5 /work/one\\n 4.0 /work/two spaces\\n');\n`, {mode: 0o700});
+    // A shell script, like the real (native) zoxide quick to start: a Node stand-in's cold start could exceed the query's
+    // 1.5 s limit on a loaded machine, and the service then (rightly) fell back to Native.
+    await writeFile(binary, `#!/bin/sh\n[ "$_ZO_DATA_DIR" = "$ORIGINAL_DB" ] && exit 2\nprintf sorted > "$_ZO_DATA_DIR/db.zo"\nprintf '["%s","%s","%s","%s"]\\n' "$1" "$2" "$3" "$4" >> "$CALLS"\nprintf '20.5 /work/one\\n 4.0 /work/two spaces\\n'\n`, {mode: 0o700});
     const calls = join(directory, 'calls');
     const service = new DirectoryService({...process.env, PATH: directory, _ZO_DATA_DIR: directory, ORIGINAL_DB: directory, CALLS: calls});
     assert.equal((await service.query([], '', 'zoxide'))[0]?.path, '/work/one');

tests/fixtures/fake-codex-app-server.mjs

@@ -37,7 +37,9 @@ createInterface({input: process.stdin}).on('line', line => {
     if (scenario === 'fail') return send({method: 'turn/completed', params: {threadId: 'thr_1', turn: {id: 'turn_1', status: 'failed', error: {message: 'usage limit reached\nsecond line'}, items: []}}});
     send({method: 'item/started', params: {threadId: 'thr_1', turnId: 'turn_1', startedAtMs: 1, item: {type: 'commandExecution', id: 'item_1', command: 'rm -rf build', commandActions: [], cwd: '/tmp', status: 'inProgress'}}});
     send({id: 100, method: 'item/commandExecution/requestApproval', params: {threadId: 'thr_1', turnId: 'turn_1', itemId: 'item_1', startedAtMs: 1, command: 'rm -rf build', cwd: '/tmp', reason: 'cleans the build'}});
-    send({id: 101, method: 'item/tool/requestUserInput', params: {threadId: 'thr_1', turnId: 'turn_1', itemId: 'item_9', questions: []}});
+    // A moment later, as a separate read: the person may already have answered 100, so the client's refusal of 101 can
+    // reach this server after that answer (and after the turn completes). Tests must wait for what they assert.
+    setTimeout(() => send({id: 101, method: 'item/tool/requestUserInput', params: {threadId: 'thr_1', turnId: 'turn_1', itemId: 'item_9', questions: []}}), 30);
     return;
   }
   if (id === 100) {