fix(terminal): recover worker records after reader contention

Retain the latest record until a transient sharing failure clears. Deliver exits while publication is pending, await publication before confirming creation, and cancel retries before retirement.

packages/server/src/terminal/terminal-worker-process.ts

@@ -22,6 +22,7 @@ import {
   writeTerminalWorkerRecord,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
+import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js";
 import { resolveTerminalWorkerSocketPath } from "./terminal-worker-socket-path.js";
 import {
   TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV,
@@ -217,19 +218,17 @@ function runWorker(argv: string[]): void {
     controller?.send(message);
   }
 
-  function persistRecord(): void {
+  const recordPublisher = createTerminalWorkerRecordPublisher((state) => {
+    writeTerminalWorkerRecord(args.paseoHome, state);
+  });
+
+  async function persistRecord(): Promise<void> {
     if (!record || exiting) return;
     record = {
       ...record,
       terminals: record.terminals.filter((entry) => sessionsById.has(entry.id)),
     };
-    console.error("CI terminal record write attempted");
-    try {
-      writeTerminalWorkerRecord(args.paseoHome, record);
-    } catch (error) {
-      console.error("CI terminal record write failed:", (error as NodeJS.ErrnoException).code);
-      throw error;
-    }
+    await recordPublisher.publish(record);
   }
 
   function removeOwnFiles(): void {
@@ -242,6 +241,7 @@ function runWorker(argv: string[]): void {
   function exitWorker(code: number): void {
     if (exiting) return;
     exiting = true;
+    recordPublisher.stop();
     removeOwnFiles();
     server?.close();
     process.exit(code);
@@ -370,7 +370,9 @@ function runWorker(argv: string[]): void {
       outputCoalescer.flush();
       clearTerminalSubscriptions(session.id);
       sessionsById.delete(session.id);
-      persistRecord();
+      void persistRecord().catch((error) => {
+        if (!exiting) console.error("Cannot publish terminal exit:", error);
+      });
       if (exitsHeldDuringCreation.has(session.id)) {
         exitsHeldDuringCreation.set(session.id, info);
         return;
@@ -445,7 +447,7 @@ function runWorker(argv: string[]): void {
         { id: session.id, pid, startedAt: inspection.snapshot.startedAt },
       ],
     };
-    persistRecord();
+    await persistRecord();
   }
 
   function enqueueCreateTerminalRequest(message: TerminalCreateRequest): Promise<void> {

packages/server/src/terminal/terminal-worker-record-publisher.test.ts

@@ -0,0 +1,70 @@
+import { describe, expect, it } from "vitest";
+import { setTimeout as delay } from "node:timers/promises";
+import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js";
+import type { TerminalWorkerRecord } from "./terminal-worker-record.js";
+
+const record: TerminalWorkerRecord = {
+  recordId: "worker",
+  pid: 123,
+  bootId: null,
+  createdAt: "now",
+  identity: { startedAt: "now", commandLine: "worker" },
+  protocolVersion: 1,
+  paseoVersion: "test",
+  socketPath: "socket",
+  secret: "secret",
+  terminals: [{ id: "terminal", pid: 124, startedAt: "now" }],
+};
+
+describe("live worker record publication", () => {
+  it.each(["EPERM", "EBUSY", "EACCES"])(
+    "publishes the latest state after %s contention clears",
+    async (code) => {
+      let locked = true;
+      const written: TerminalWorkerRecord[] = [];
+      const publisher = createTerminalWorkerRecordPublisher((state) => {
+        if (locked) throw Object.assign(new Error("reader holds record"), { code });
+        written.push(state);
+      });
+      try {
+        const creation = publisher.publish(record);
+        const exited = { ...record, terminals: [] };
+        const exit = publisher.publish(exited);
+        expect(written).toEqual([]);
+        locked = false;
+        await Promise.all([creation, exit]);
+        expect(written).toEqual([exited]);
+      } finally {
+        publisher.stop();
+      }
+    },
+  );
+
+  it("does not recreate a retired record when a reader releases it", async () => {
+    let locked = true;
+    const written: TerminalWorkerRecord[] = [];
+    const publisher = createTerminalWorkerRecordPublisher((state) => {
+      if (locked)
+        throw Object.assign(new Error("reader holds record"), {
+          code: "EPERM",
+        });
+      written.push(state);
+    });
+    const pending = publisher.publish(record);
+    const rejected = expect(pending).rejects.toThrow("stopped");
+    publisher.stop();
+    locked = false;
+    await rejected;
+    await delay(50);
+    expect(written).toEqual([]);
+    await expect(publisher.publish(record)).rejects.toThrow("stopped");
+  });
+
+  it("reports permanent storage failures without retrying", async () => {
+    const publisher = createTerminalWorkerRecordPublisher(() => {
+      throw Object.assign(new Error("disk full"), { code: "ENOSPC" });
+    });
+    await expect(publisher.publish(record)).rejects.toThrow("disk full");
+    publisher.stop();
+  });
+});

packages/server/src/terminal/terminal-worker-record-publisher.ts

@@ -0,0 +1,62 @@
+import type { TerminalWorkerRecord } from "./terminal-worker-record.js";
+
+const TRANSIENT_ERRORS = new Set(["EPERM", "EBUSY", "EACCES"]);
+
+/** Serializes live worker publication, retaining the latest state across reader contention.
+ * Creation awaits publication; exit notifications need not wait for a reader to close.
+ * Stop before retiring the record so a pending retry cannot recreate it.
+ */
+export function createTerminalWorkerRecordPublisher(write: (record: TerminalWorkerRecord) => void) {
+  let stopped = false;
+  let latest: TerminalWorkerRecord;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let pending:
+    | {
+        promise: Promise<void>;
+        resolve: () => void;
+        reject: (error: unknown) => void;
+      }
+    | undefined;
+
+  function attempt(): void {
+    timer = undefined;
+    const publication = pending;
+    if (!publication) return;
+    try {
+      write(latest);
+      publication.resolve();
+      pending = undefined;
+    } catch (error) {
+      if (TRANSIENT_ERRORS.has((error as NodeJS.ErrnoException).code ?? "")) {
+        timer = setTimeout(attempt, 25);
+      } else {
+        publication.reject(error);
+        pending = undefined;
+      }
+    }
+  }
+
+  return {
+    publish(record: TerminalWorkerRecord): Promise<void> {
+      if (stopped) return Promise.reject(new Error("Worker record publisher stopped"));
+      latest = record;
+      if (pending) return pending.promise;
+      let resolve!: () => void;
+      let reject!: (error: unknown) => void;
+      const promise = new Promise<void>((yes, no) => {
+        resolve = yes;
+        reject = no;
+      });
+      pending = { promise, resolve, reject };
+      attempt();
+      return promise;
+    },
+    stop(): void {
+      stopped = true;
+      clearTimeout(timer);
+      timer = undefined;
+      pending?.reject(new Error("Worker record publisher stopped"));
+      pending = undefined;
+    },
+  };
+}

packages/server/src/terminal/worker-terminal-manager.test.ts

@@ -45,7 +45,6 @@ import {
   listTerminalWorkerRecords,
   readTerminalWorkerRecord,
   terminalWorkerDirectory,
-  terminalWorkerLogPath,
   terminalWorkerRecordPath,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
@@ -807,45 +806,35 @@ it.each([false, true])(
     await expect(
       reapTerminalWorkers({
         paseoHome,
-        inspector: { ...inspector, listProcesses: () => Promise.reject(new Error("ps timed out")) },
+        inspector: {
+          ...inspector,
+          listProcesses: () => Promise.reject(new Error("ps timed out")),
+        },
       }),
     ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError);
-    await first.detach();
     expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
-    // Diagnostic: keep the same read handle the poller uses open across the exit write.
+    // Keep an ordinary reader open until the worker delivers the terminal exit.
+    // Windows cannot replace the record while this handle is open. The exit must
+    // still reach the controller, and publication must recover after release.
     const recordFd = holdReader
       ? openSync(terminalWorkerRecordPath(paseoHome, record!.pid), "r")
       : null;
-    const attemptsBeforeExit =
-      readFileSync(terminalWorkerLogPath(paseoHome), "utf8").match(
-        /CI terminal record write attempted/g,
-      )?.length ?? 0;
     try {
       writeFileSync(exitRequestPath, "");
-      await waitForCondition(
-        async () => !(await recordedTerminalPids(paseoHome)).has(ending.id),
-        10000,
-      );
-    } catch (error) {
-      const workerLog = readFileSync(terminalWorkerLogPath(paseoHome), "utf8");
-      console.error("CI terminal exit diagnostics", {
-        holdReader,
-        attemptsAfterExit:
-          (workerLog.match(/CI terminal record write attempted/g)?.length ?? 0) -
-          attemptsBeforeExit,
-        uncaughtExceptionLogged: workerLog.includes("Terminal worker uncaught exception"),
-        exitObserved: existsSync(exitObservedPath),
-        endingAlive: isProcessAlive(endingPid),
-        remainingAlive: isProcessAlive(remainingPid),
-        recordWriteErrors: workerLog.match(/CI terminal record write failed: \w+/g) ?? [],
-      });
-      throw error;
+      await waitForCondition(() => first.getTerminal(ending.id) === undefined, 10000);
+      expect(existsSync(exitObservedPath)).toBe(true);
+      expect(isProcessAlive(endingPid)).toBe(false);
     } finally {
       if (recordFd !== null) closeSync(recordFd);
     }
+    await waitForCondition(
+      async () => !(await recordedTerminalPids(paseoHome)).has(ending.id),
+      10000,
+    );
     expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
+    await first.detach();
     manager = await startRealWorkerManager({ paseoHome });
 
     expect(manager.getTerminal(remaining.id)).toBeUndefined();