fix(test): wait for writers before removing temp homes to end ENOTEMPTY flakes (lr-c3e80a)

test/access-harness.js

@@ -15,6 +15,7 @@ var path = require("path");
 var crypto = require("crypto");
 var http = require("http");
 var WebSocket = require("ws");
+var { removeTempHome } = require("./helpers/remove-temp-home");
 
 var COOKIE_NAME = "relay_auth_user";
 var TOKEN_TTL_MS = 30 * 24 * 60 * 60 * 1000;
@@ -202,19 +203,12 @@ async function start(spec) {
     }
     await new Promise(function (resolve) { relay.server.close(function () { resolve(); }); });
     if (typeof relay.server.closeAllConnections === "function") relay.server.closeAllConnections();
-    // An adapter may still be writing under HOME as the server shuts down:
-    // give those writes time to land, and retry the removal while one races it.
-    await settle(300);
-    var removed = false;
-    for (var attempt = 0; attempt < 5 && !removed; attempt++) {
-      try {
-        fs.rmSync(home, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 });
-        removed = true;
-      } catch (e) {
-        if (e.code !== "ENOTEMPTY" || attempt === 4) failures.push(e);
-        if (e.code !== "ENOTEMPTY") break;
-        await settle(300);
-      }
+    // An adapter may still be writing under HOME as the server shuts down: the
+    // helper lets its in-flight writes finish before it removes the home.
+    try {
+      await removeTempHome(home);
+    } catch (e) {
+      failures.push(e);
     }
     if (failures.length) throw failures[0];
   }

test/automation-sender.test.js

@@ -11,6 +11,7 @@ var os = require("os");
 var path = require("path");
 var execFileSync = require("child_process").execFileSync;
 
+var { removeTempHome } = require("./helpers/remove-temp-home");
 var turnInitiator = require("../lib/turn-initiator");
 var sender = require("../lib/injected-sender");
 var { RATE_LIMIT_RESUME_NOTE } = require("../lib/scheduled-input/rate-limit-resume");
@@ -305,11 +306,14 @@ test("a loop iteration is recorded as the loop, with its name and iteration, and
     assert.deepEqual(turnInitiator.claudeOrigin(entry.initiator), { kind: "task-notification", subkind: "scheduled-trigger", fireReason: "loop" });
   } finally {
     global.setTimeout = origSetTimeout;
-    armed.forEach(function (t) { clearTimeout(t); });
     if (origHome === undefined) delete process.env.CLAGENTIC_CONSOLE_HOME;
     else process.env.CLAGENTIC_CONSOLE_HOME = origHome;
-    fs.rmSync(cwd, { recursive: true, force: true });
-    fs.rmSync(home, { recursive: true, force: true });
+    // The loop persists through lib/store in several async steps, which the
+    // helper lets finish. Its retry and watchdog timers (lib/project-loop) can
+    // still start a save later, so every timer it armed is cancelled first.
+    var cancelLoopTimers = function () { armed.forEach(function (t) { clearTimeout(t); }); };
+    await removeTempHome(cwd, { quiesce: cancelLoopTimers });
+    await removeTempHome(home, { quiesce: cancelLoopTimers });
   }
 });
 

test/fixtures/daemon-harness.js

@@ -13,6 +13,7 @@ var http = require("http");
 var { spawn } = require("child_process");
 var WebSocket = require("ws");
 var { TOKEN_ENV, newHarnessToken, verifyOwnedPid } = require("./process-ownership");
+var { removeTempHome } = require("../helpers/remove-temp-home");
 
 var DAEMON_SCRIPT = path.resolve(__dirname, "..", "..", "lib", "daemon.js");
 var SETUP_CODE = "HARNESSCODE";
@@ -61,7 +62,7 @@ async function waitForServer(port, ms) {
 
 function waitForExit(proc, ms) {
   return new Promise(function (resolve, reject) {
-    if (proc.exitCode !== null) return resolve(proc.exitCode);
+    if (proc.exitCode !== null || proc.signalCode !== null) return resolve(proc.exitCode);
     var timer = setTimeout(function () { reject(new Error("daemon did not exit within " + ms + " ms")); }, ms);
     proc.once("exit", function (code) { clearTimeout(timer); resolve(code); });
   });
@@ -217,12 +218,21 @@ async function bootDaemon(o) {
           successor = null;
         }
       }
+      var children = [proc];
       if (successor) {
+        children.push(successor);
         signalPid(successor, "SIGTERM");
         await new Promise(function (r) { setTimeout(r, 1500); });
         signalPid(successor, "SIGKILL");
       }
-      fs.rmSync(home, { recursive: true, force: true, maxRetries: 5, retryDelay: 100 });
+      // A claude process the daemon started can outlive it and keep writing
+      // under the home; the helper stops what carries the token and waits for
+      // every child before removing.
+      try {
+        await removeTempHome(home, { children: children, token: token });
+      } catch (e) {
+        problems.push(e.message);
+      }
       if (problems.length > 0) throw new Error(problems.join("; "));
     },
   };

test/helpers/remove-temp-home.js

@@ -0,0 +1,210 @@
+"use strict";
+
+// The one way a test removes a temp home that a process or an in-process async
+// writer may still be writing into. A bare synchronous rm races those writers:
+// a file or directory created while the removal runs leaves the parent
+// non-empty (ENOTEMPTY), and a write queued behind the removal recreates the
+// tree afterwards.
+//
+// Order matters and is the fix, not a retry: (0) run the caller's quiesce step,
+// which stops in-process writers that schedule work on a timer; (1) wait,
+// bounded, for every child the test was told about to exit, failing with its
+// pid when one does not; (2) let in-process promise-based fs work finish, so
+// the event loop is free to run the chain that is still writing; (3) remove;
+// (4) confirm the home stayed gone. The retry options only cover a transient
+// directory listing race after (1) and (2).
+//
+// Limit: the drain in (2) sees fs requests that already exist. It cannot see a
+// timer or debounce that will start a write later, so a writer that schedules
+// on a timer must be stopped by the caller through its own close, flush or stop
+// API, passed as `quiesce`.
+
+var fs = require("fs");
+var path = require("path");
+var util = require("util");
+var { verifyOwnedPid } = require("../fixtures/process-ownership");
+
+var REMOVE_OPTIONS = { recursive: true, force: true, maxRetries: 10, retryDelay: 100 };
+var DEFAULT_WAIT_MS = 10000;
+var POLL_MS = 25;
+var debugLog = util.debuglog("remove-temp-home");
+
+function sleep(ms) {
+  return new Promise(function (resolve) { setTimeout(resolve, ms); });
+}
+
+// A zombie has exited and only waits for its parent to reap it; it writes nothing.
+function pidAlive(pid) {
+  try { process.kill(pid, 0); } catch (e) { return e.code === "EPERM"; }
+  try {
+    var stat = fs.readFileSync("/proc/" + pid + "/stat", "utf8");
+    return stat.slice(stat.lastIndexOf(") ") + 2, stat.lastIndexOf(") ") + 3) !== "Z";
+  } catch (e) {
+    return true;
+  }
+}
+
+// Every live process that carries the harness token: the daemon's own children
+// and grandchildren inherit its environment, and one outlives the daemon when
+// the daemon exits first (it is re-parented, so no handle to it exists).
+// scanned:false means /proc cannot be listed (macOS), so nothing was checked.
+function ownedProcesses(token) {
+  var out = [];
+  var entries;
+  try { entries = fs.readdirSync("/proc"); } catch (e) { return { scanned: false, procs: out }; }
+  entries.forEach(function (name) {
+    if (!/^\d+$/.test(name)) return;
+    var pid = Number(name);
+    if (pid === process.pid || !pidAlive(pid)) return;
+    if (!verifyOwnedPid(pid, token).ok) return;
+    out.push({ pid: pid, start: startTimeOf(pid) });
+  });
+  return { scanned: true, procs: out };
+}
+
+// Field 22 of /proc/<pid>/stat, the process start time in clock ticks since
+// boot: with the pid it identifies one process, so a recycled pid differs.
+function startTimeOf(pid) {
+  try {
+    var stat = fs.readFileSync("/proc/" + pid + "/stat", "utf8");
+    return stat.slice(stat.lastIndexOf(") ") + 2).split(" ")[19] || null;
+  } catch (e) {
+    return null;
+  }
+}
+
+// Still the process that was found: alive, started at the same time, and still
+// carrying the token. A pid recycled for an unrelated process fails this.
+function stillOwned(proc, token) {
+  return pidAlive(proc.pid) && proc.start !== null &&
+    startTimeOf(proc.pid) === proc.start && verifyOwnedPid(proc.pid, token).ok;
+}
+
+function signalAll(pids, signal) {
+  pids.forEach(function (pid) {
+    try { process.kill(pid, signal); } catch (e) { if (e.code !== "ESRCH") throw e; }
+  });
+}
+
+// Stops what the harness started and the test holds no handle to. Only a
+// process that carries this instance's token is signalled, and it is checked
+// again right before SIGKILL. Returns the pids that survived, so the caller
+// fails loudly rather than skipping them.
+async function stopOwned(token, ms, log, io) {
+  // `io` lets a test observe the signals and fake the identity check; callers
+  // outside this file's tests pass nothing.
+  var deps = Object.assign({ scan: ownedProcesses, owned: stillOwned, signal: signalAll }, io);
+  var found = deps.scan(token);
+  if (!found.scanned) {
+    log("token scan skipped: /proc cannot be listed on this platform");
+    return [];
+  }
+  var procs = found.procs;
+  if (procs.length === 0) return [];
+  function owned(p) { return deps.owned(p, token); }
+  function pidOf(p) { return p.pid; }
+  deps.signal(procs.map(pidOf), "SIGTERM");
+  var end = Date.now() + ms;
+  var killedAt = Date.now() + ms / 2;
+  var killed = false;
+  while (Date.now() < end) {
+    procs = procs.filter(owned);
+    if (procs.length === 0) return [];
+    if (!killed && Date.now() >= killedAt) { deps.signal(procs.map(pidOf), "SIGKILL"); killed = true; }
+    await sleep(POLL_MS);
+  }
+  return procs.filter(owned).map(pidOf);
+}
+
+// A child is a ChildProcess the test spawned, or a bare pid recorded elsewhere
+// (a self-spawned successor). Resolves null once it has exited, or a message
+// naming it when the bound expires.
+function waitForChild(child, ms) {
+  if (typeof child === "number") {
+    return (async function () {
+      var end = Date.now() + ms;
+      while (Date.now() < end) {
+        if (!pidAlive(child)) return null;
+        await sleep(POLL_MS);
+      }
+      return pidAlive(child) ? "pid " + child + " was still running after " + ms + " ms" : null;
+    })();
+  }
+  if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve(null);
+  return new Promise(function (resolve) {
+    var timer = setTimeout(function () {
+      resolve("pid " + child.pid + " was still running after " + ms + " ms");
+    }, ms);
+    child.once("exit", function () { clearTimeout(timer); resolve(null); });
+  });
+}
+
+function fsPromiseRequestsInFlight() {
+  return process.getActiveResourcesInfo().indexOf("FSReqPromise") !== -1;
+}
+
+// Every step of a promise-based fs chain (mkdir, then writeFile, then rename)
+// is queued from a microtask after the previous step settles, so a check that
+// runs after a macrotask turn sees the chain's next request if there is one.
+async function drainPromiseFs(ms) {
+  var end = Date.now() + ms;
+  do {
+    await new Promise(function (resolve) { setImmediate(resolve); });
+    if (!fsPromiseRequestsInFlight()) return null;
+    await sleep(POLL_MS);
+  } while (Date.now() < end);
+  return "promise-based fs requests were still in flight after " + ms + " ms";
+}
+
+/**
+ * @param {string} home the directory to remove
+ * @param {object} [o]
+ * @param {Array<import("child_process").ChildProcess|number>} [o.children] processes that may write into home
+ * @param {string} [o.token] the harness token the spawned processes carry; every live process with it is stopped (an orphaned grandchild included)
+ * @param {number} [o.waitMs] bound for each wait (default 10000)
+ * @param {function(): (void|Promise<void>)} [o.quiesce] stops in-process writers
+ *   that schedule a write on a timer or debounce, through the component's own
+ *   close, flush or stop API. Runs first. The helper cannot detect such a
+ *   writer: a timer that has not fired yet has no fs request to wait for.
+ * @param {function(string): void} [o.log] receives the one-line notice that the
+ *   token scan was skipped (default: util.debuglog("remove-temp-home"))
+ * @throws when a child outlives the bound or the home comes back; the message names each pid
+ */
+async function removeTempHome(home, o) {
+  var opts = o || {};
+  var waitMs = opts.waitMs || DEFAULT_WAIT_MS;
+  var problems = [];
+  var log = opts.log || debugLog;
+
+  if (opts.quiesce) {
+    try { await opts.quiesce(); } catch (e) { problems.push("quiesce failed: " + (e && e.message ? e.message : e)); }
+  }
+
+  var verdicts = await Promise.all((opts.children || []).map(function (c) { return waitForChild(c, waitMs); }));
+  verdicts.forEach(function (v) { if (v) problems.push(v + " (it may still write into " + home + ")"); });
+
+  if (opts.token) {
+    var survivors = await stopOwned(opts.token, waitMs, log);
+    survivors.forEach(function (pid) {
+      problems.push("pid " + pid + " carrying the harness token survived SIGKILL (it may still write into " + home + ")");
+    });
+  }
+
+  var drained = await drainPromiseFs(waitMs);
+  if (drained) problems.push(drained);
+
+  // Bounded passes: a chain that began before the first pass and wrote after it
+  // shows up as the home existing again on the confirming check.
+  for (var pass = 0; pass < 3; pass++) {
+    await fs.promises.rm(home, REMOVE_OPTIONS);
+    await new Promise(function (resolve) { setImmediate(resolve); });
+    if (!fsPromiseRequestsInFlight() && !fs.existsSync(home)) break;
+    var again = await drainPromiseFs(waitMs);
+    if (again) { problems.push(again); break; }
+    if (pass === 2 && fs.existsSync(home)) problems.push(path.basename(home) + " was recreated after removal");
+  }
+
+  if (problems.length > 0) throw new Error(problems.join("; "));
+}
+
+module.exports = { removeTempHome, pidAlive, startTimeOf, stillOwned, stopOwned, REMOVE_OPTIONS };

test/message-sender-record.test.js

@@ -14,6 +14,7 @@ var os = require("os");
 var path = require("path");
 var execFileSync = require("child_process").execFileSync;
 
+var { removeTempHome } = require("./helpers/remove-temp-home");
 var { INITIATORS } = require("../lib/turn-initiator");
 var { RATE_LIMIT_RESUME_NOTE } = require("../lib/scheduled-input/rate-limit-resume");
 var { createScheduledInputService } = require("../lib/scheduled-input/service");
@@ -96,6 +97,7 @@ function makeWorld(vendor, opts) {
   });
   var world = {
     vendor: vendor, home: home, recordsDir: recordsDir, sm: sm, sdk: sdk, calls: calls, project: project, processor: processor,
+    services: [],
     newSession: function (extra) {
       var s = sm.createSessionRaw({ vendor: vendor });
       return Object.assign(s, extra || {});
@@ -110,21 +112,30 @@ function makeWorld(vendor, opts) {
         agentCatalog: function () { return []; },
       }, extra || {}));
     },
-    cleanup: function () {
+    cleanup: async function () {
       if (origHome === undefined) delete process.env.CLAGENTIC_CONSOLE_HOME; else process.env.CLAGENTIC_CONSOLE_HOME = origHome;
       if (origLegacyHome === undefined) delete process.env.CLAGENTIC_HOME; else process.env.CLAGENTIC_HOME = origLegacyHome;
       ["../lib/config", "../lib/sessions", "../lib/utils"].forEach(function (m) { delete require.cache[require.resolve(m)]; });
-      // A loop saves its state asynchronously into this directory, so removal
-      // can meet a file written meanwhile: retry.
-      fs.rmSync(home, { recursive: true, force: true, maxRetries: 10, retryDelay: 50 });
+      // A loop saves its state asynchronously into this directory (lib/store
+      // writes in several fs.promises steps); the helper lets those finish
+      // before removing, which a synchronous removal cannot. Two writers start
+      // work from a timer instead: the session manager batches appends behind a
+      // 50 ms timer, and the scheduled-input service re-arms its own. Each is
+      // stopped through its existing API before the helper drains.
+      await removeTempHome(home, {
+        quiesce: function () {
+          world.services.forEach(function (service) { service.stop(); });
+          sm.sessions.forEach(function (session) { sm.flushSessionBuffer(session); });
+        },
+      });
     },
   };
   return world;
 }
 
 async function inWorld(vendor, fn, opts) {
   var w = makeWorld(vendor, opts);
-  try { await fn(w); } finally { w.cleanup(); }
+  try { await fn(w); } finally { await w.cleanup(); }
 }
 
 // The browser composer's message handler over the real session manager.
@@ -165,6 +176,7 @@ function makeScheduledInputService(w) {
     sessionInput: w.sessionInput(),
   });
   service.attachProject("p");
+  w.services.push(service);
   return service;
 }