Recover terminal worker records after Windows reader contention (#8155)

* test(terminal): let cleanup fixture exit on request

* test(terminal): capture the Windows record rewrite failure

* test(terminal): repeat Windows cleanup exit diagnostics

* fix(server): reopening a Codex agent no longer reads and holds its full history (#8132)

* fix(server): read Codex history only when Paseo replays it

Reopening a Codex agent whose timeline Paseo still held read the full
thread on connect and kept it on the session until the next replay, which
never came, so every reopened agent held a second copy of its transcript.
thread/resume also returned every turn, and only the sandbox policy was
used from it.

Resume now sends excludeTurns (older Codex ignores the unknown field) and
the history read moves into streamHistory, whose caller owns the result.
A resumed session still needs the earlier turns for rewind targets and the
sub-agent routes that keep an earlier child's interactions on its original
card, so the first turn, goal or rewind indexes the thread and drops the
timeline it read.

* test(server): run reopened Codex agent tests without the Codex CLI installed

* test(terminal): isolate record reader contention on exit

* feat(usage): show when each limit resets, and make usage rows easier to read (#8162)

The reset countdown floored to one unit, so a weekly window read "resets 2d"
from 2d 23h down to 2d 0h and looked stuck for a day; the session window had
the same problem with hours. Countdowns now use two units ("2d 23h", "4h 59m"),
and every row also shows the clock time it resets ("7:20 AM", "Sat 8:00 AM",
"Oct 31, 3:00 PM"), so users can plan work around a reset without hovering.

Rows now put the window name and percent on one line, the bar under it, and the
reset on its own full-width line, so the exact time fits at phone width in
longer languages. The pinned-row background is gone; the pin glyph, drawn in
the foreground colour when pinned, is the only pinned signal. Bars are 6px.

* Make Claude session entrypoint relabeling optional (#8164)

* fix(claude): make session entrypoint relabeling opt-in

Keep inherited SDK and parent identities cleared. Honor the explicit session-open entrypoint contribution so users can opt into the label through a plugin instead of every session receiving it unconditionally.

* test(hub): use standard budget for restart and worktree recovery

The Windows CI scenario timed out at its custom 20-second limit while the other Hub cases took 4–10 seconds each. This scenario starts two daemons and creates, archives and restores a worktree. Use the existing 30-second suite budget without changing any assertions.

* Deliver a steer message that arrives while the previous turn is still starting (#6192)

* Deliver a steer message that arrives while the previous turn is still starting

A steer send that reached an agent before its previous turn had a turn id
found no turn to steer and replaced the starting turn, so the earlier
message never reached the provider. Steering now waits for the starting
turn to be accepted (or fail) and steers the turn it became.

Fixes #6186

* Wake a waiting steer only once the starting turn has an id or its run has ended

A failed start used to wake the waiting steer before its run was cleared,
leaving the later send to replace a run that had already failed.

* Clear a failed start's run even when reporting the failure throws

A steer waiting on that start is released only when the run is cleared.

* Test that a failed start keeps a waiting steer until its run is cleared

* fix(server): steer Codex turns while native identification is pending

Wait for the accepted foreground turn to receive its native ID before steering it. Returning unavailable in this window replaced the turn even after the manager had waited for startup.

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

* 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.

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>

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

@@ -18,6 +18,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,
@@ -214,13 +215,17 @@ function runWorker(argv: string[]): void {
     replayController.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)),
     };
-    writeTerminalWorkerRecord(args.paseoHome, record);
+    await recordPublisher.publish(record);
   }
 
   function removeOwnFiles(): void {
@@ -233,6 +238,7 @@ function runWorker(argv: string[]): void {
   function exitWorker(code: number): void {
     if (exiting) return;
     exiting = true;
+    recordPublisher.stop();
     removeOwnFiles();
     server?.close();
     process.exit(code);
@@ -349,7 +355,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;
@@ -424,7 +432,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

@@ -16,7 +16,9 @@ import { afterEach, expect, it } from "vitest";
 import { isPlatform } from "../test-utils/platform.js";
 import { EventEmitter } from "node:events";
 import {
+  closeSync,
   existsSync,
+  openSync,
   mkdirSync,
   mkdtempSync,
   readdirSync,
@@ -58,6 +60,7 @@ import {
   listTerminalWorkerRecords,
   readTerminalWorkerRecord,
   terminalWorkerDirectory,
+  terminalWorkerRecordPath,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
 import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js";
@@ -807,52 +810,81 @@ async function recordedTerminalPids(paseoHome: string): Promise<Map<string, numb
   return pids;
 }
 
-it("never adopts a live worker's terminals after a kill cleanup it could not finish, even once the worker rewrites its record", async () => {
-  const paseoHome = createTemporaryHome();
-  const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-"));
-  temporaryDirs.push(cwd);
-  const first = await startRealWorkerManager({ paseoHome });
-  manager = first;
-  const ending = await first.createTerminal({
-    workspaceId: "ws-unfinished-kill",
-    cwd,
-    ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
-  });
-  const remaining = await first.createTerminal({
-    workspaceId: "ws-unfinished-kill",
-    cwd,
-    ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
-  });
-  const pids = await recordedTerminalPids(paseoHome);
-  const endingPid = pids.get(ending.id)!;
-  const remainingPid = pids.get(remaining.id)!;
+it.each([false, true])(
+  "never adopts a live worker's terminals after a kill cleanup it could not finish, even once the worker rewrites its record (held reader: %s)",
+  async (holdReader) => {
+    const paseoHome = createTemporaryHome();
+    const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-"));
+    temporaryDirs.push(cwd);
+    const first = await startRealWorkerManager({ paseoHome });
+    manager = first;
+    const exitRequestPath = join(cwd, "exit-requested");
+    const exitObservedPath = join(cwd, "exit-observed");
+    const ending = await first.createTerminal({
+      workspaceId: "ws-unfinished-kill",
+      cwd,
+      ...nodeTerminalCommand(`
+      const { existsSync, writeFileSync } = require("node:fs");
+      setInterval(() => {
+        if (existsSync(${JSON.stringify(exitRequestPath)})) {
+          writeFileSync(${JSON.stringify(exitObservedPath)}, "");
+          process.exit(0);
+        }
+      }, 25);
+    `),
+    });
+    const remaining = await first.createTerminal({
+      workspaceId: "ws-unfinished-kill",
+      cwd,
+      ...nodeTerminalCommand("setInterval(() => {}, 1000);"),
+    });
+    const pids = await recordedTerminalPids(paseoHome);
+    const endingPid = pids.get(ending.id)!;
+    const remainingPid = pids.get(remaining.id)!;
+    const [record] = await workerRecords(paseoHome);
 
-  // A kill cleanup whose process table read fails signals nothing and leaves the worker live.
-  const inspector = createTerminalWorkerProcessInspector();
-  await expect(
-    reapTerminalWorkers({
-      paseoHome,
-      inspector: { ...inspector, listProcesses: () => Promise.reject(new Error("ps timed out")) },
-    }),
-  ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError);
-  await first.detach();
-  expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
+    // A kill cleanup whose process table read fails signals nothing and leaves the worker live.
+    const inspector = createTerminalWorkerProcessInspector();
+    await expect(
+      reapTerminalWorkers({
+        paseoHome,
+        inspector: {
+          ...inspector,
+          listProcesses: () => Promise.reject(new Error("ps timed out")),
+        },
+      }),
+    ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError);
+    expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
-  // One terminal ends on its own: the live worker rewrites its record from memory.
-  process.kill(endingPid, "SIGKILL");
-  await waitForCondition(
-    async () => !(await recordedTerminalPids(paseoHome)).has(ending.id),
-    10000,
-  );
-  expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
+    // 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;
+    try {
+      writeFileSync(exitRequestPath, "");
+      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);
 
-  manager = await startRealWorkerManager({ paseoHome });
+    await first.detach();
+    manager = await startRealWorkerManager({ paseoHome });
 
-  expect(manager.getTerminal(remaining.id)).toBeUndefined();
-  expect(await manager.getTerminals(cwd)).toEqual([]);
-  await waitForCondition(() => !isProcessAlive(remainingPid), 10000);
-  expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]);
-});
+    expect(manager.getTerminal(remaining.id)).toBeUndefined();
+    expect(await manager.getTerminals(cwd)).toEqual([]);
+    await waitForCondition(() => !isProcessAlive(remainingPid), 10000);
+    expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]);
+  },
+);
 
 // The worker's own `ps`, delayed: a program that exits while the worker reads its birth time
 // lands its exit in the middle of creation. Only the worker launched while this PATH is set sees