test(terminal): isolate record reader contention on exit

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

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

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

@@ -73,9 +73,8 @@ function applyRuntimeSettingsToClaudeOptions(
         runtimeSettings,
         overlays: [launchEnv],
       });
-      // 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;
+      // Set our identity after sanitization clears the parent session's entrypoint.
+      providerEnvSpec.envOverlay.CLAUDE_CODE_ENTRYPOINT = "Paseo";
       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

@@ -16,7 +16,6 @@ import type {
   AgentSessionConfig,
   AgentSlashCommand,
   AgentStreamEvent,
-  AgentTimelineItem,
 } from "../agent-sdk-types.js";
 import {
   buildCodexAppServerEnv,
@@ -115,6 +114,7 @@ interface CodexSessionTestAccess {
   ensureThreadLoaded(): Promise<void>;
   handleToolApprovalRequest(params: unknown): Promise<unknown>;
   handleNotification(method: string, params: unknown): Promise<void>;
+  loadPersistedHistory(client: CodexClientLike | null): Promise<void>;
   refreshResolvedCollaborationMode(): void;
   serviceTier: "fast" | null;
   planModeEnabled: boolean;
@@ -503,81 +503,6 @@ function archivedThreadErrorMessage(threadId: string): string {
   );
 }
 
-function codexUserItem(id: string, text: string) {
-  return { type: "userMessage", id, content: [{ type: "text", text }] };
-}
-
-function codexAgentItem(text: string) {
-  return { type: "agentMessage", id: `agent-${text}`, text };
-}
-
-// An AgentManager over a real Codex provider whose every spawn gets a fresh fake
-// app-server serving `threads` from thread/read.
-function createReopenableCodexAgent(threads: Record<string, unknown[]>) {
-  const appServers: FakeCodexAppServer[] = [];
-  const provider = new CodexAppServerAgentClient(createTestLogger());
-  const internals = castInternals<{
-    goalsEnabledPromise: Promise<boolean> | null;
-    autoReviewEnabledPromise: Promise<boolean> | null;
-    spawnAppServer: () => Promise<ChildProcessWithoutNullStreams>;
-  }>(provider);
-  internals.goalsEnabledPromise = Promise.resolve(false);
-  internals.autoReviewEnabledPromise = Promise.resolve(false);
-  internals.spawnAppServer = async () => {
-    const appServer = createFakeCodexAppServer({
-      "thread/read": (params) => {
-        const { threadId } = params as { threadId: string };
-        return { thread: { id: threadId, turns: threads[threadId] ?? [] } };
-      },
-    });
-    appServers.push(appServer);
-    return appServer.child;
-  };
-  // The fake app-server stands in for the Codex CLI, which test machines may not have.
-  const client: AgentClient = {
-    provider: "codex",
-    capabilities: provider.capabilities,
-    isAvailable: async () => true,
-    fetchCatalog: async () => ({ models: [], modes: [] }),
-    createSession: async () => {
-      throw new Error("Unexpected create");
-    },
-    resumeSession: (...args) => provider.resumeSession(...args),
-  };
-  const manager = new AgentManager({ clients: { codex: client }, logger: createTestLogger() });
-  const agentId = "00000000-0000-4000-8000-000000000001";
-  const handle = { provider: "codex", sessionId: "saved-thread" };
-  const config = createConfig({ cwd: process.cwd() });
-
-  async function open(): Promise<void> {
-    await manager.resumeAgentFromPersistence(handle, config, agentId);
-    await manager.hydrateTimelineFromProvider(agentId);
-  }
-
-  return {
-    manager,
-    agentId,
-    latestAppServer: () => appServers[appServers.length - 1],
-    threadMethods: (index: number) =>
-      appServers[index]
-        .requests()
-        .map((request) => request.method)
-        .filter((method) => typeof method === "string" && method.startsWith("thread/")),
-    // Cold-loads the agent, closes its runtime, and reopens it while Paseo keeps its timeline.
-    async loadCloseAndReopen(): Promise<AgentTimelineItem[]> {
-      await open();
-      const projected = manager.getTimeline(agentId);
-      await manager.closeAgent(agentId);
-      await open();
-      return projected;
-    },
-    async close(): Promise<void> {
-      await manager.closeAgent(agentId);
-      for (const appServer of appServers) appServer.assertNoErrors();
-    },
-  };
-}
-
 function asInternals(session: CodexTestSession): CodexSessionTestAccess {
   return castInternals<CodexSessionTestAccess>(session);
 }
@@ -1972,147 +1897,12 @@ describe("Codex app-server provider", () => {
       "thread/resume",
       "thread/unarchive",
       "thread/resume",
+      "thread/read",
     ]);
     await session.close();
     appServer.assertNoErrors();
   });
 
-  test("resumes a Codex thread without asking for its turns", async () => {
-    const appServer = createFakeCodexAppServer();
-    const provider = createProviderWithFakeAppServer(appServer);
-
-    const session = await provider.resumeSession({ sessionId: "saved-thread" });
-
-    expect(
-      appServer
-        .requests()
-        .filter((request) => request.method === "thread/resume")
-        .map((request) => request.params),
-    ).toEqual([expect.objectContaining({ threadId: "saved-thread", excludeTurns: true })]);
-    await session.close();
-    appServer.assertNoErrors();
-  });
-
-  test("reopening an agent whose timeline Paseo still holds does not read Codex history", async () => {
-    const codex = createReopenableCodexAgent({
-      "saved-thread": [
-        { id: "turn-1", items: [codexUserItem("user-1", "Hello"), codexAgentItem("Hi there.")] },
-      ],
-    });
-    try {
-      const projected = await codex.loadCloseAndReopen();
-
-      expect(projected).toEqual([
-        expect.objectContaining({ type: "user_message", messageId: "user-1", text: "Hello" }),
-        expect.objectContaining({ type: "assistant_message", text: "Hi there." }),
-      ]);
-      expect(codex.threadMethods(0)).toEqual([
-        "thread/loaded/list",
-        "thread/resume",
-        "thread/read",
-      ]);
-      expect(codex.threadMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]);
-      expect(codex.manager.getTimeline(codex.agentId)).toEqual(projected);
-    } finally {
-      await codex.close();
-    }
-  });
-
-  test("rewinds an agent reopened without reading its Codex history", async () => {
-    const firstTurn = {
-      id: "turn-1",
-      items: [codexUserItem("user-1", "First"), codexAgentItem("First answer.")],
-    };
-    const codex = createReopenableCodexAgent({
-      "saved-thread": [
-        firstTurn,
-        { id: "turn-2", items: [codexUserItem("user-2", "Second"), codexAgentItem("Second.")] },
-      ],
-      "forked-thread": [firstTurn],
-    });
-    try {
-      await codex.loadCloseAndReopen();
-
-      await codex.manager.rewind(codex.agentId, "user-2", "conversation");
-
-      expect(codex.latestAppServer().recordedRollbacks).toEqual([
-        { threadId: "forked-thread", numTurns: 1 },
-      ]);
-      expect(codex.manager.getTimeline(codex.agentId)).toEqual([
-        expect.objectContaining({ type: "user_message", messageId: "user-1", text: "First" }),
-        expect.objectContaining({ type: "assistant_message", text: "First answer." }),
-      ]);
-    } finally {
-      await codex.close();
-    }
-  });
-
-  test("keeps a reopened agent's interaction with an earlier sub-agent on its original card", async () => {
-    const codex = createReopenableCodexAgent({
-      "saved-thread": [
-        {
-          id: "turn-1",
-          items: [
-            codexUserItem("user-1", "Delegate"),
-            {
-              type: "subAgentActivity",
-              id: "spawn-child",
-              kind: "started",
-              agentThreadId: "child-thread",
-              agentPath: "/root/child",
-            },
-          ],
-        },
-      ],
-      "child-thread": [{ id: "child-turn-1", items: [codexAgentItem("Child started.")] }],
-    });
-    try {
-      await codex.loadCloseAndReopen();
-      const appServer = codex.latestAppServer();
-
-      const run = codex.manager.runAgent(codex.agentId, "Ask the child again");
-      await appServer.waitForTurnStart();
-      appServer.startsTurn({ threadId: "saved-thread", turnId: "turn-2" });
-      appServer.beginsSubAgentActivity({
-        callId: "interact-child",
-        threadId: "child-thread",
-        parentThreadId: "saved-thread",
-        kind: "interacted",
-        agentPath: "/root/child",
-      });
-      appServer.says({ threadId: "child-thread", itemId: "child-reply", text: "Child replied." });
-      appServer.completesSubAgentActivity({
-        callId: "interact-child",
-        threadId: "child-thread",
-        parentThreadId: "saved-thread",
-        kind: "interacted",
-        agentPath: "/root/child",
-      });
-      appServer.completeTurn({ threadId: "saved-thread" });
-      await run;
-
-      const toolCalls = codex.manager
-        .getTimeline(codex.agentId)
-        .filter((item) => item.type === "tool_call");
-      expect(new Set(toolCalls.map((item) => item.callId))).toEqual(new Set(["spawn-child"]));
-      expect(toolCalls.at(-1)).toMatchObject({ callId: "spawn-child", status: "completed" });
-      expect(codex.manager.getProviderSubagent(codex.agentId, "child-thread")).toMatchObject({
-        toolCallId: "spawn-child",
-        status: "completed",
-      });
-      expect(
-        codex.manager
-          .fetchProviderSubagentTimeline(codex.agentId, "child-thread")
-          .rows.map((row) => row.item),
-      ).toEqual([
-        expect.objectContaining({ type: "assistant_message", text: "Child started." }),
-        expect.objectContaining({ type: "assistant_message", text: "Child replied." }),
-      ]);
-    } finally {
-      await codex.close();
-    }
-  });
-
   test("closes Codex app-server when an interactive resume fails", async () => {
     const appServer = createFakeCodexAppServer({
       "thread/resume": () => Promise.reject(new Error("thread resume failed")),
@@ -4883,6 +4673,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -4935,7 +4727,7 @@ describe("Codex app-server provider", () => {
       })),
     };
 
-    await Array.fromAsync(session.streamHistory());
+    await asInternals(session).loadPersistedHistory(session.client);
 
     expect(asInternals(session).codexUserMessageTurns().resolve("message-history")).toEqual({
       index: 0,
@@ -5004,6 +4796,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -5164,6 +4958,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -5227,6 +5023,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -5402,6 +5200,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -5461,6 +5261,8 @@ describe("Codex app-server provider", () => {
       }),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);
@@ -5762,10 +5564,7 @@ describe("Codex app-server provider", () => {
     expect(session.currentThreadId).toBe("archived-thread-id");
     expect(requests).toEqual([
       { method: "thread/loaded/list", params: {} },
-      {
-        method: "thread/resume",
-        params: { threadId: "archived-thread-id", excludeTurns: true },
-      },
+      { method: "thread/resume", params: { threadId: "archived-thread-id" } },
     ]);
   });
 
@@ -6600,6 +6399,8 @@ describe("Codex app-server provider", () => {
       })),
     };
 
+    await asInternals(session).loadPersistedHistory(session.client);
+
     const history: AgentStreamEvent[] = [];
     for await (const event of session.streamHistory()) {
       history.push(event);

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

@@ -496,11 +496,6 @@ interface PersistedTimelineEntry {
   providerTurnId?: string;
 }
 
-interface CodexThreadHistory {
-  timeline: PersistedTimelineEntry[];
-  providerSubagentEvents: AgentStreamEvent[];
-}
-
 interface PersistedSubAgentRoute {
   childThreadId: string;
   toolCall: ToolCallTimelineItem;
@@ -3469,12 +3464,10 @@ export class CodexAppServerAgentSession implements AgentSession {
   private serviceTier: string | null = null;
   private speedModels: CodexModel[] = [];
   private planModeEnabled = false;
-  // History read by rewind or an archived resume, held until streamHistory consumes it.
-  private stagedHistory: CodexThreadHistory | null = null;
-  // Collects provider subagent events emitted while a thread history read rebuilds routes.
-  private historySubagentEvents: AgentStreamEvent[] | null = null;
-  // A resumed thread's earlier turns and sub-agents are known only after reading its history.
-  private historyIndexed = true;
+  private historyPending = false;
+  private persistedHistory: PersistedTimelineEntry[] = [];
+  private loadingPersistedHistory = false;
+  private persistedProviderSubagentEvents: AgentStreamEvent[] = [];
   private pendingPermissions = new Map<string, AgentPermissionRequest>();
   private mcpElicitationPermissionIds = new Map<number, string>();
   private pendingPermissionHandlers = new Map<string, CodexPendingPermissionHandler>();
@@ -3581,7 +3574,7 @@ export class CodexAppServerAgentSession implements AgentSession {
 
     if (this.resumeHandle?.sessionId) {
       this.currentThreadId = this.resumeHandle.sessionId;
-      this.historyIndexed = false;
+      this.historyPending = true;
       // Resuming can start a goal before the caller receives the session.
       if (this.initialResumePurpose === "interactive") this.startupEvents = [];
     }
@@ -3629,7 +3622,7 @@ export class CodexAppServerAgentSession implements AgentSession {
 
   private async establishConnection(): Promise<void> {
     if (this.initialResumePurpose === "history") {
-      this.stagedHistory = await this.readArchivedHistory();
+      await this.readArchivedHistory();
       this.connectionState = "history-ready";
       return;
     }
@@ -3662,6 +3655,7 @@ export class CodexAppServerAgentSession implements AgentSession {
 
       if (this.currentThreadId) {
         await this.ensureThreadLoaded();
+        await this.loadPersistedHistory(this.client);
         await this.applyDefaultModelAndThinking();
       }
 
@@ -3976,40 +3970,39 @@ export class CodexAppServerAgentSession implements AgentSession {
     );
   }
 
-  private async readArchivedHistory(): Promise<CodexThreadHistory> {
+  private async readArchivedHistory(): Promise<void> {
     const child = await this.spawnAppServer();
     const client = new CodexAppServerClient(child, this.logger, () => this.traceContext());
     try {
       await client.request("initialize", buildCodexAppServerInitializeParams());
       client.notify("initialized", {});
-      return await this.readThreadHistory(client);
+      await this.loadPersistedHistory(client);
     } finally {
       await client.dispose();
     }
   }
 
-  // Reads the thread and its sub-agent threads, rebuilding the turn index and
-  // sub-agent routes. The caller owns the returned history; nothing retains it.
-  private async readThreadHistory(
-    client: CodexAppServerClientLike | null,
-  ): Promise<CodexThreadHistory> {
-    if (!client || !this.currentThreadId) {
-      return { timeline: [], providerSubagentEvents: [] };
-    }
-    const { timeline, subAgentRoutes } = await loadCodexThreadHistoryTimeline({
-      threadId: this.currentThreadId,
+  private async loadPersistedHistory(client: CodexAppServerClientLike | null): Promise<void> {
+    if (!client || !this.currentThreadId) return;
+    const threadId = this.currentThreadId;
+
+    const history = await loadCodexThreadHistoryTimeline({
+      threadId,
       cwd: this.config.cwd ?? null,
-      requestThread: (threadIdToRead) => readCodexThread(client, threadIdToRead),
+      requestThread: (threadIdToRead) => {
+        return readCodexThread(client, threadIdToRead);
+      },
     });
+    const { timeline, subAgentRoutes } = history;
     this.subAgentCallsByCallId.clear();
     this.subAgentCallIdByChildThreadId.clear();
     this.pendingSubAgentNotificationsByThreadId.clear();
-    const providerSubagentEvents: AgentStreamEvent[] = [];
-    this.historySubagentEvents = providerSubagentEvents;
+    this.persistedProviderSubagentEvents = [];
+    this.loadingPersistedHistory = true;
     try {
       await this.loadPersistedSubAgentHistories(client, subAgentRoutes);
     } finally {
-      this.historySubagentEvents = null;
+      this.loadingPersistedHistory = false;
     }
     this.resetCodexUserMessageTurns();
     for (const entry of timeline) {
@@ -4020,15 +4013,8 @@ export class CodexAppServerAgentSession implements AgentSession {
         this.rememberCodexUserMessageTurn(entry.item.messageId, entry.providerTurnId);
       }
     }
-    this.historyIndexed = true;
-    return { timeline, providerSubagentEvents };
-  }
-
-  // Turns and rewind route by the thread's earlier turns and sub-agents. A resumed
-  // session whose projection Paseo already holds reads them here and drops the rest.
-  private async ensureHistoryIndexed(): Promise<void> {
-    if (this.historyIndexed) return;
-    await this.readThreadHistory(this.client);
+    this.persistedHistory = timeline;
+    this.historyPending = timeline.length > 0;
   }
 
   private async loadPersistedSubAgentHistories(
@@ -4080,8 +4066,7 @@ export class CodexAppServerAgentSession implements AgentSession {
 
   private async ensureThreadLoaded(): Promise<void> {
     if (!this.client || !this.currentThreadId) return;
-    // Paseo reads history only when it replays it; the resume response needs no turns.
-    const params: Record<string, unknown> = { threadId: this.currentThreadId, excludeTurns: true };
+    const params: Record<string, unknown> = { threadId: this.currentThreadId };
     const developerInstructions = composeSystemPromptParts(
       this.config.systemPrompt,
       this.config.daemonAppendSystemPrompt,
@@ -4401,7 +4386,6 @@ export class CodexAppServerAgentSession implements AgentSession {
 
       if (this.currentThreadId) {
         await this.ensureThreadLoaded();
-        await this.ensureHistoryIndexed();
       } else {
         await this.ensureThread();
       }
@@ -4575,12 +4559,21 @@ export class CodexAppServerAgentSession implements AgentSession {
   }
 
   async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
-    const history = this.stagedHistory ?? (await this.readThreadHistory(this.client));
-    this.stagedHistory = null;
-    for (const event of history.providerSubagentEvents) {
+    if (
+      (!this.historyPending || this.persistedHistory.length === 0) &&
+      this.persistedProviderSubagentEvents.length === 0
+    ) {
+      return;
+    }
+    const history = this.persistedHistory;
+    const providerSubagents = this.persistedProviderSubagentEvents;
+    this.persistedHistory = [];
+    this.persistedProviderSubagentEvents = [];
+    this.historyPending = false;
+    for (const event of providerSubagents) {
       yield event;
     }
-    for (const entry of history.timeline) {
+    for (const entry of history) {
       yield {
         type: "timeline",
         provider: CODEX_PROVIDER,
@@ -4963,7 +4956,6 @@ export class CodexAppServerAgentSession implements AgentSession {
     } else {
       await this.ensureThread();
     }
-    await this.ensureHistoryIndexed();
 
     await revertCodexConversation({
       client: this.client,
@@ -4978,8 +4970,10 @@ export class CodexAppServerAgentSession implements AgentSession {
       setThreadId: async (threadId) => {
         this.currentThreadId = threadId;
         this.cachedRuntimeInfo = null;
-        this.stagedHistory = await this.readThreadHistory(this.client);
-        this.reconcileAsyncQuestionsAfterRewind(this.stagedHistory.timeline);
+        this.persistedHistory = [];
+        this.historyPending = false;
+        await this.loadPersistedHistory(this.client);
+        this.reconcileAsyncQuestionsAfterRewind();
       },
     });
   }
@@ -5217,8 +5211,6 @@ export class CodexAppServerAgentSession implements AgentSession {
       await this.connect();
       if (this.currentThreadId) {
         await this.ensureThreadLoaded();
-        // A goal runs turns of its own.
-        await this.ensureHistoryIndexed();
       } else {
         await this.ensureThread();
       }
@@ -5412,8 +5404,8 @@ export class CodexAppServerAgentSession implements AgentSession {
   }
 
   private emitEvent(event: AgentStreamEvent): void {
-    if (this.historySubagentEvents && event.type === "provider_subagent") {
-      this.historySubagentEvents.push(event);
+    if (this.loadingPersistedHistory && event.type === "provider_subagent") {
+      this.persistedProviderSubagentEvents.push(event);
       return;
     }
     this.notifySubscribers(event);
@@ -6591,9 +6583,9 @@ export class CodexAppServerAgentSession implements AgentSession {
     }
   }
 
-  private reconcileAsyncQuestionsAfterRewind(timeline: readonly PersistedTimelineEntry[]): void {
+  private reconcileAsyncQuestionsAfterRewind(): void {
     const retainedIds = new Set(
-      timeline.flatMap(({ item }) =>
+      this.persistedHistory.flatMap(({ item }) =>
         item.type === "tool_call" && item.name === "request_user_input_async" ? [item.callId] : [],
       ),
     );

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

@@ -1,10 +1,10 @@
-import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { mkdtemp, rm } 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 { AgentClient, AgentSessionConfig } from "../agent/agent-sdk-types.js";
+import type { 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,71 +72,3 @@ 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

@@ -223,7 +223,13 @@ function runWorker(argv: string[]): void {
       ...record,
       terminals: record.terminals.filter((entry) => sessionsById.has(entry.id)),
     };
-    writeTerminalWorkerRecord(args.paseoHome, record);
+    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;
+    }
   }
 
   function removeOwnFiles(): void {

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,8 @@ import {
   listTerminalWorkerRecords,
   readTerminalWorkerRecord,
   terminalWorkerDirectory,
+  terminalWorkerLogPath,
+  terminalWorkerRecordPath,
   type TerminalWorkerRecord,
 } from "./terminal-worker-record.js";
 import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js";
@@ -765,52 +769,91 @@ 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);
+    await first.detach();
+    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);
+    // Diagnostic: keep the same read handle the poller uses open across the exit write.
+    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;
+    } finally {
+      if (recordFd !== null) closeSync(recordFd);
+    }
+    expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0);
 
-  manager = await startRealWorkerManager({ paseoHome });
+    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