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/server/agent/agent-manager.test.ts

@@ -956,6 +956,131 @@ test("unavailable steer interrupts once and starts one replacement turn", async
   }
 });
 
+test("steers a send that arrives while the previous turn is still starting", async () => {
+  const startHeld = deferred<void>();
+  class SlowStartSteeringSession extends SteeringTestSession {
+    override async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
+      await startHeld.promise;
+      return await super.startTurn(prompt);
+    }
+  }
+  const session = new SlowStartSteeringSession({
+    provider: "codex",
+    cwd: process.cwd(),
+  });
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-starting-"));
+  const manager = new AgentManager({
+    clients: {
+      codex: new (class extends TestAgentClient {
+        override async createSession(): Promise<AgentSession> {
+          return session;
+        }
+      })(),
+    },
+    logger,
+  });
+  let agentId: string | null = null;
+  try {
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    agentId = agent.id;
+    const steerOptions = (clientMessageId: string) => ({
+      replaceRunning: true,
+      activeTurnBehavior: "steer" as const,
+      runOptions: { clientMessageId },
+    });
+    await startAgentRun(manager, agent.id, "FIRST", logger, steerOptions("first-client"));
+    const second = startAgentRun(
+      manager,
+      agent.id,
+      "SECOND",
+      logger,
+      steerOptions("second-client"),
+    );
+
+    startHeld.resolve();
+    await expect(second).resolves.toEqual({ disposition: "steered" });
+
+    expect(session.startPrompts).toEqual(["FIRST"]);
+    expect(session.interruptCount).toBe(0);
+    expect(
+      manager
+        .fetchTimeline(agent.id, { limit: 0 })
+        .rows.filter((row) => row.item.type === "user_message")
+        .map((row) => ({
+          text: row.item.type === "user_message" ? row.item.text : null,
+          turnId: row.turnId,
+        })),
+    ).toEqual([
+      { text: "FIRST", turnId: "active-turn-1" },
+      { text: "SECOND", turnId: "active-turn-1" },
+    ]);
+  } finally {
+    startHeld.resolve();
+    if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
+test("starts its own turn when the previous turn fails to start under a steer send", async () => {
+  const startHeld = deferred<void>();
+  class FailingFirstStartSession extends SteeringTestSession {
+    override async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
+      if (prompt === "FIRST") {
+        this.startPrompts.push(prompt);
+        await startHeld.promise;
+        throw new Error("first start failed");
+      }
+      return await super.startTurn(prompt);
+    }
+  }
+  const session = new FailingFirstStartSession({
+    provider: "codex",
+    cwd: process.cwd(),
+  });
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-failed-start-"));
+  const manager = new AgentManager({
+    clients: {
+      codex: new (class extends TestAgentClient {
+        override async createSession(): Promise<AgentSession> {
+          return session;
+        }
+      })(),
+    },
+    logger,
+  });
+  let agentId: string | null = null;
+  try {
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    agentId = agent.id;
+    const steerOptions = (clientMessageId: string) => ({
+      replaceRunning: true,
+      activeTurnBehavior: "steer" as const,
+      runOptions: { clientMessageId },
+    });
+    await startAgentRun(manager, agent.id, "FIRST", logger, steerOptions("first-client"));
+    const second = startAgentRun(
+      manager,
+      agent.id,
+      "SECOND",
+      logger,
+      steerOptions("second-client"),
+    );
+
+    startHeld.resolve();
+    await expect(second).resolves.toEqual({ disposition: "turn_started" });
+    await vi.waitFor(() => expect(session.startPrompts).toEqual(["FIRST", "SECOND"]));
+    expect(session.interruptCount).toBe(0);
+  } finally {
+    startHeld.resolve();
+    if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});
+
 test("orders an accepted steer before output emitted while acknowledgement is pending", async () => {
   const entered = deferred<void>();
   const release = deferred<void>();

packages/server/src/server/agent/agent-manager.ts

@@ -2734,7 +2734,7 @@ export class AgentManager {
         throw error;
       }
       if (isStaleProviderSessionError(error)) {
-        pendingRun.start = { status: "failed", error: error.message };
+        this.runs.settleForegroundStart(pendingRun, { status: "failed", error: error.message });
         agent.pendingReplacement = false;
         if (!agent.activeForegroundTurnId) agent.lifecycle = "idle";
         this.runs.settleForegroundRun(agentId, pendingRun.token);
@@ -2742,20 +2742,24 @@ export class AgentManager {
       }
       agent.pendingReplacement = false;
       const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
-      pendingRun.start = { status: "failed", error: errorMsg };
-      // A terminal rejection belongs after the submitted prompt even though no provider turn exists.
-      if (options?.clientMessageId) {
-        this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "rejected", {
-          messageId: options.clientMessageId,
+      this.runs.settleForegroundStart(pendingRun, { status: "failed", error: errorMsg });
+      try {
+        // A terminal rejection belongs after the submitted prompt even though no provider turn exists.
+        if (options?.clientMessageId) {
+          this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, "rejected", {
+            messageId: options.clientMessageId,
+          });
+        }
+        await this.handleStreamEvent(agent, {
+          type: "turn_failed",
+          provider: agent.provider,
+          error: errorMsg,
         });
+        this.finalizeForegroundTurn(agent);
+      } finally {
+        // A steer waiting on this start is released only once the run is cleared.
+        this.runs.settleForegroundRun(agentId, pendingRun.token);
       }
-      await this.handleStreamEvent(agent, {
-        type: "turn_failed",
-        provider: agent.provider,
-        error: errorMsg,
-      });
-      this.finalizeForegroundTurn(agent);
-      this.runs.settleForegroundRun(agentId, pendingRun.token);
       throw error;
     }
   }
@@ -2832,7 +2836,7 @@ export class AgentManager {
         agent.pendingReplacement = false;
       }
       const turnStartedAt = new Date();
-      pendingRun.start = { status: "started", turnId };
+      this.runs.settleForegroundStart(pendingRun, { status: "started", turnId });
       agent.activeForegroundTurnId = turnId;
       this.openActiveTurn(agent, turnId, turnStartedAt);
       agent.lifecycle = "running";
@@ -3041,6 +3045,9 @@ export class AgentManager {
     prompt: AgentPromptInput,
     options?: AgentSteerOptions,
   ): Promise<ActiveTurnSteerDispatchResult> {
+    // A turn that is still starting has no id to steer yet. Replacing it would drop its
+    // prompt, so wait until it is accepted (or fails) and steer the turn it became.
+    await this.runs.waitForForegroundStart(agentId);
     const agent = this.requireSessionAgent(agentId);
     const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId;
     if (!expectedTurnId) {

packages/server/src/server/agent/agent-run-state.test.ts

@@ -0,0 +1,51 @@
+import { describe, expect, test } from "vitest";
+
+import { AgentRunState } from "./agent-run-state.js";
+
+function track(promise: Promise<void>): { readonly resolved: boolean } {
+  const state = { resolved: false };
+  void (async () => {
+    await promise;
+    state.resolved = true;
+  })();
+  return state;
+}
+
+async function flushMicrotasks(): Promise<void> {
+  await new Promise<void>((resolve) => setImmediate(resolve));
+}
+
+describe("waitForForegroundStart", () => {
+  test("resolves once the starting turn has an id", async () => {
+    const runs = new AgentRunState();
+    const run = runs.createPendingRun("agent");
+    const wait = track(runs.waitForForegroundStart("agent"));
+
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(false);
+
+    runs.settleForegroundStart(run, { status: "started", turnId: "turn-1" });
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(true);
+  });
+
+  test("waits for a failed start's run to be cleared", async () => {
+    const runs = new AgentRunState();
+    const run = runs.createPendingRun("agent");
+    const wait = track(runs.waitForForegroundStart("agent"));
+
+    runs.settleForegroundStart(run, { status: "failed", error: "start failed" });
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(false);
+    expect(runs.hasPendingRun("agent")).toBe(true);
+
+    runs.settleForegroundRun("agent", run.token);
+    await flushMicrotasks();
+    expect(wait.resolved).toBe(true);
+  });
+
+  test("resolves at once when no turn is starting", async () => {
+    const runs = new AgentRunState();
+    await expect(runs.waitForForegroundStart("agent")).resolves.toBeUndefined();
+  });
+});

packages/server/src/server/agent/agent-run-state.ts

@@ -10,14 +10,18 @@ export interface ForegroundTurnWaiter {
   resolveSettled: () => void;
 }
 
+type ForegroundRunStart =
+  | { status: "pending" }
+  | { status: "started"; turnId: string }
+  | { status: "failed"; error: string };
+
 export interface PendingForegroundRun {
   token: string;
   kind: "foreground";
   stagedEvents: AgentStreamEvent[];
-  start:
-    | { status: "pending" }
-    | { status: "started"; turnId: string }
-    | { status: "failed"; error: string };
+  start: ForegroundRunStart;
+  turnStartedPromise: Promise<void>;
+  resolveTurnStarted: () => void;
   settled: boolean;
   settledPromise: Promise<void>;
   resolveSettled: () => void;
@@ -54,6 +58,28 @@ export class AgentRunState {
     return run?.kind === "foreground" ? run : null;
   }
 
+  settleForegroundStart(
+    run: PendingForegroundRun,
+    start: Exclude<ForegroundRunStart, { status: "pending" }>,
+  ): void {
+    run.start = start;
+    if (start.status === "started") {
+      run.resolveTurnStarted();
+    }
+  }
+
+  /**
+   * Resolves once the agent's foreground run has a turn id or has ended. A failed start
+   * resolves only once its run is cleared, so callers never see a failed run still tracked.
+   */
+  async waitForForegroundStart(agentId: string): Promise<void> {
+    const run = this.getPendingRun(agentId);
+    if (run?.start.status !== "pending") {
+      return;
+    }
+    await Promise.race([run.turnStartedPromise, run.settledPromise]);
+  }
+
   hasPendingRun(agentId: string): boolean {
     return this.getPendingRun(agentId) !== null;
   }
@@ -277,10 +303,16 @@ export class ForegroundTurnStream {
 }
 
 function createPendingForegroundRun(): PendingForegroundRun {
+  let resolveTurnStarted!: () => void;
+  const turnStartedPromise = new Promise<void>((resolvePromise) => {
+    resolveTurnStarted = resolvePromise;
+  });
   return {
     ...createTrackedRunState(),
     kind: "foreground",
     start: { status: "pending" },
+    turnStartedPromise,
+    resolveTurnStarted,
     stagedEvents: [],
   };
 }

packages/server/src/server/agent/providers/claude/agent.env.test.ts

@@ -30,16 +30,20 @@ function createQueryMock(events: unknown[]): Query {
 }
 
 describe("Claude SDK env", () => {
-  test.each(["node", "bun", process.execPath])(
-    "identifies the spawned %s process as Paseo after clearing parent session env",
-    async (command) => {
+  test.each(
+    ["node", "bun", process.execPath].flatMap((command) =>
+      [undefined, "Paseo", "custom-client"].map((entrypoint) => ({ command, entrypoint })),
+    ),
+  )(
+    "preserves explicit launch identity $entrypoint in $command while clearing inherited session env",
+    async ({ command, entrypoint }) => {
       let output = "";
       let exited = Promise.resolve<unknown[]>([]);
       const query = claudeQuery(
         { prompt: "", options: {} },
         {
           runtimeSettings: { env: { CLAUDE_CODE_ENTRYPOINT: "sdk-ts" } },
-          launchEnv: { CLAUDE_CODE_ENTRYPOINT: "sdk-cli" },
+          launchEnv: entrypoint === undefined ? {} : { CLAUDE_CODE_ENTRYPOINT: entrypoint },
           queryFactory: ({ options }) => {
             if (!options.spawnClaudeCodeProcess) {
               throw new Error("Missing Claude process launcher");
@@ -65,7 +69,10 @@ describe("Claude SDK env", () => {
 
       try {
         expect(await exited).toEqual([0, null]);
-        expect(JSON.parse(output)).toEqual({ entrypoint: "Paseo", parentSession: null });
+        expect(JSON.parse(output)).toEqual({
+          ...(entrypoint === undefined ? {} : { entrypoint }),
+          parentSession: null,
+        });
       } finally {
         query.close();
       }

packages/server/src/server/agent/providers/claude/query.ts

@@ -73,8 +73,9 @@ function applyRuntimeSettingsToClaudeOptions(
         runtimeSettings,
         overlays: [launchEnv],
       });
-      // Set our identity after sanitization clears the parent session's entrypoint.
-      providerEnvSpec.envOverlay.CLAUDE_CODE_ENTRYPOINT = "Paseo";
+      // Session-open hooks may explicitly supply a client identity. Keep inherited
+      // SDK/parent identities cleared while honoring this launch-only contribution.
+      providerEnvSpec.envOverlay.CLAUDE_CODE_ENTRYPOINT = launchEnv?.CLAUDE_CODE_ENTRYPOINT;
       const providerEnv = createExternalProcessEnv(spawnOptions.env, providerEnvSpec.envOverlay);
       const selfNodeCommand = isDefaultRuntime
         ? buildSelfNodeCommand(resolved.args, providerEnv)

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -226,6 +226,64 @@ function deferred<T>() {
 }
 
 describe("Codex active-turn steering admission", () => {
+  test("steers an accepted turn whose native id arrives after startTurn returns", async () => {
+    const appServer = createFakeCodexAppServer({
+      "turn/steer": () => ({ turn: { id: "native-A" } }),
+    });
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.startsTurn({ threadId: "thread-1", turnId: "native-A" });
+
+      await expect(steer).resolves.toEqual({ status: "accepted" });
+      expect(
+        appServer.requests().filter((request) => request.method === "turn/start"),
+      ).toHaveLength(1);
+      expect(appServer.requests().filter((request) => request.method === "turn/interrupt")).toEqual(
+        [],
+      );
+      expect(appServer.requests().find((request) => request.method === "turn/steer")).toMatchObject(
+        {
+          params: {
+            threadId: "thread-1",
+            expectedTurnId: "native-A",
+            input: [{ type: "text", text: "SECOND" }],
+          },
+        },
+      );
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
+  test("releases a waiting steer when the accepted turn ends before identification", async () => {
+    const appServer = createFakeCodexAppServer();
+    const session = new CodexAppServerAgentSession(
+      createConfig({ cwd: "/workspace/project" }),
+      null,
+      createTestLogger(),
+      async () => appServer.child,
+    );
+    try {
+      const { turnId } = await session.startTurn("FIRST");
+      const steer = session.steerActiveTurn("SECOND", { expectedTurnId: turnId });
+      appServer.completeTurn();
+
+      await expect(steer).resolves.toEqual({ status: "unavailable" });
+      expect(appServer.requests().filter((request) => request.method === "turn/steer")).toEqual([]);
+      appServer.assertNoErrors();
+    } finally {
+      await session.close();
+    }
+  });
+
   test("a steer without the clearing contract leaves permissions open", async () => {
     const appServer = createFakeCodexAppServer({
       "turn/steer": () => ({ turn: { id: "native-A" } }),

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -4456,9 +4456,18 @@ export class CodexAppServerAgentSession implements AgentSession {
   ): Promise<SteerResult> {
     const client = this.client;
     const threadId = this.currentThreadId;
-    const nativeTurnId = this.currentTurnId;
+    let nativeTurnId = this.currentTurnId;
     const foregroundTurnId = this.activeForegroundTurnId;
-    if (!client || !threadId || !nativeTurnId || foregroundTurnId !== options.expectedTurnId) {
+    if (!client || !threadId || foregroundTurnId !== options.expectedTurnId) {
+      return { status: "unavailable" };
+    }
+    const pendingIdentification = this.pendingForegroundTurnIdentification;
+    // The start acknowledgement precedes the notification that identifies the native
+    // turn. Keep this steer attached to the accepted foreground turn while it arrives.
+    if (!nativeTurnId && pendingIdentification?.foregroundTurnId === foregroundTurnId) {
+      nativeTurnId = await pendingIdentification.promise;
+    }
+    if (!nativeTurnId) {
       return { status: "unavailable" };
     }
     if (await this.resolveSlashCommandInvocation(prompt)) return { status: "unavailable" };

packages/server/src/server/daemon-e2e/codex-steer-starting.real.e2e.test.ts

@@ -0,0 +1,80 @@
+import { expect, test } from "vitest";
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import pino from "pino";
+
+import { CodexAppServerAgentClient } from "../agent/providers/codex-app-server-agent.js";
+import { DaemonClient } from "../test-utils/daemon-client.js";
+import { createMessageCollector } from "../test-utils/message-collector.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+import { canRunRealProvider } from "./real-provider-test-config.js";
+
+test("keeps two concurrent startup steer messages in one real Codex turn", async (context) => {
+  if (!(await canRunRealProvider("codex"))) {
+    context.skip();
+    return;
+  }
+  const cwd = mkdtempSync(path.join(tmpdir(), "codex-steer-starting-"));
+  const logger = pino({ level: "silent" });
+  const daemon = await createTestPaseoDaemon({
+    agentClients: { codex: new CodexAppServerAgentClient(logger) },
+    logger,
+  });
+  const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` });
+  const collector = createMessageCollector(client);
+  const subscriptions: ReturnType<DaemonClient["subscribeAgentTimeline"]>[] = [];
+  try {
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    for (let attempt = 0; attempt < 3; attempt += 1) {
+      const agent = await client.createAgent({
+        cwd,
+        provider: "codex",
+        modeId: "full-access",
+      });
+      const subscription = client.subscribeAgentTimeline(agent.id, () => {});
+      subscriptions.push(subscription);
+      await subscription.ready;
+      await Promise.all([
+        client.sendAgentMessage(agent.id, "Reply exactly: FIRST", {
+          messageId: `first-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+        client.sendAgentMessage(agent.id, "Reply exactly: SECOND", {
+          messageId: `second-${attempt}`,
+          activeTurnBehavior: "steer",
+        }),
+      ]);
+      await client.waitForFinish(agent.id, 90_000);
+      const timeline = await client.fetchAgentTimeline(agent.id, { limit: 100 });
+      const users = timeline.entries.filter((entry) => entry.item.type === "user_message");
+      const events = collector.messages.flatMap((message) =>
+        message.type === "agent_stream" && message.payload.agentId === agent.id
+          ? [message.payload.event]
+          : [],
+      );
+      expect(users.map((entry) => entry.item.text).sort()).toEqual([
+        "Reply exactly: FIRST",
+        "Reply exactly: SECOND",
+      ]);
+      expect(new Set(users.map((entry) => entry.turnId)).size).toBe(1);
+      expect(events.filter((event) => event.type === "turn_started")).toHaveLength(1);
+      expect(
+        events.filter((event) => event.type === "turn_canceled" || event.type === "turn_failed"),
+      ).toHaveLength(0);
+      expect(
+        timeline.entries.some(
+          (entry) =>
+            entry.item.type === "assistant_message" && entry.item.text.includes("[System Error]"),
+        ),
+      ).toBe(false);
+    }
+  } finally {
+    await Promise.all(subscriptions.map((subscription) => subscription.release()));
+    collector.unsubscribe();
+    await client.close();
+    await daemon.close();
+    rmSync(cwd, { recursive: true, force: true });
+  }
+}, 300_000);

packages/server/src/server/hub/execution-session.websocket.test.ts

@@ -248,7 +248,7 @@ test("ordinary Hub requests survive daemon restart and restore an archived works
     }),
   ).toMatchObject({ payload: { agentId, accepted: true } });
   expect(hub.providerPromptTexts().filter((text) => text === "follow up")).toHaveLength(1);
-}, 20_000);
+});
 
 test("Hub completes the standard hello before rejecting a second hello", async () => {
   const hub = await launchRelationship();

packages/server/src/server/plugins/agent-configuration.e2e.test.ts

@@ -1,10 +1,10 @@
-import { mkdtemp, rm } from "node:fs/promises";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
 import { fileURLToPath } from "node:url";
 import { expect, test } from "vitest";
 import { DaemonClient } from "../test-utils/daemon-client.js";
-import type { AgentSessionConfig } from "../agent/agent-sdk-types.js";
+import type { AgentClient, AgentSessionConfig } from "../agent/agent-sdk-types.js";
 import { createTestAgentClient } from "../test-utils/fake-agent-client.js";
 import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
 
@@ -72,3 +72,71 @@ test("the configuration example saves Codex options and injects launch-only MCP
     await rm(directory, { recursive: true, force: true });
   }
 }, 60_000);
+
+test("session-open entrypoint contributions are optional and rederived for existing agents", async () => {
+  const directory = await mkdtemp(path.join(tmpdir(), "paseo-entrypoint-hook-"));
+  const launches: { provider: string; env?: Record<string, string> }[] = [];
+  function recordingProvider(provider: "claude" | "codex"): AgentClient {
+    const inner = createTestAgentClient(provider);
+    return {
+      provider: inner.provider,
+      capabilities: inner.capabilities,
+      isAvailable: () => inner.isAvailable(),
+      fetchCatalog: (options) => inner.fetchCatalog(options),
+      createSession: async (...args: Parameters<typeof inner.createSession>) => {
+        launches.push({ provider, env: args[1]?.env });
+        return inner.createSession(...args);
+      },
+      resumeSession: async (...args: Parameters<typeof inner.resumeSession>) => {
+        launches.push({ provider, env: args[2]?.env });
+        return inner.resumeSession(...args);
+      },
+    };
+  }
+  const daemon = await createTestPaseoDaemon({
+    daemonVersion: "0.14.0-beta.2",
+    agentClients: { claude: recordingProvider("claude"), codex: recordingProvider("codex") },
+    mcpEnabled: false,
+  });
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    appVersion: "0.14.0-beta.2",
+  });
+  try {
+    await writeFile(
+      path.join(directory, "paseo-plugin.json"),
+      JSON.stringify({ id: "entrypoint-test", requirements: { paseo: ">=0.14.0-beta.2" } }),
+    );
+    await writeFile(
+      path.join(directory, "index.server.ts"),
+      `
+export default function contribute(server) {
+  return server.before("agent.session_open", ({ request }) => {
+    if (request.provider !== "claude") return;
+    return { ...request, env: { ...request.env, CLAUDE_CODE_ENTRYPOINT: "custom-client" } };
+  });
+}
+`,
+    );
+    await client.connect();
+    await client.fetchAgents({ subscribe: {} });
+    const existing = await client.createAgent({ provider: "claude", cwd: directory });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+    await client.patchDaemonConfig({ pluginsEnabled: true });
+    await client.installDirectoryPlugin(directory);
+    await daemon.daemon.agentManager.reloadAgentSession(existing.id);
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBe("custom-client");
+    await client.createAgent({ provider: "claude", cwd: directory });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBe("custom-client");
+    await client.createAgent({ provider: "codex", cwd: directory });
+    expect(launches.at(-1)).toMatchObject({ provider: "codex" });
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+    await client.disablePlugin("entrypoint-test");
+    await daemon.daemon.agentManager.reloadAgentSession(existing.id);
+    expect(launches.at(-1)?.env?.CLAUDE_CODE_ENTRYPOINT).toBeUndefined();
+  } finally {
+    await client.close();
+    await daemon.close();
+    await rm(directory, { recursive: true, force: true });
+  }
+}, 60_000);

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,13 +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)),
     };
-    writeTerminalWorkerRecord(args.paseoHome, record);
+    await recordPublisher.publish(record);
   }
 
   function removeOwnFiles(): void {
@@ -236,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);
@@ -364,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;
@@ -439,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

@@ -2,7 +2,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,
@@ -43,6 +45,7 @@ import {
   listTerminalWorkerRecords,
   readTerminalWorkerRecord,
   terminalWorkerDirectory,
+  terminalWorkerRecordPath,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
 import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js";
@@ -765,52 +768,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