test(terminal): isolate record reader contention on exit

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

@@ -956,131 +956,6 @@ 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)) {
-        this.runs.settleForegroundStart(pendingRun, { status: "failed", error: error.message });
+        pendingRun.start = { status: "failed", error: error.message };
         agent.pendingReplacement = false;
         if (!agent.activeForegroundTurnId) agent.lifecycle = "idle";
         this.runs.settleForegroundRun(agentId, pendingRun.token);
@@ -2742,24 +2742,20 @@ export class AgentManager {
       }
       agent.pendingReplacement = false;
       const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
-      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,
+      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.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;
     }
   }
@@ -2836,7 +2832,7 @@ export class AgentManager {
         agent.pendingReplacement = false;
       }
       const turnStartedAt = new Date();
-      this.runs.settleForegroundStart(pendingRun, { status: "started", turnId });
+      pendingRun.start = { status: "started", turnId };
       agent.activeForegroundTurnId = turnId;
       this.openActiveTurn(agent, turnId, turnStartedAt);
       agent.lifecycle = "running";
@@ -3045,9 +3041,6 @@ 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

@@ -1,51 +0,0 @@
-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,18 +10,14 @@ 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: ForegroundRunStart;
-  turnStartedPromise: Promise<void>;
-  resolveTurnStarted: () => void;
+  start:
+    | { status: "pending" }
+    | { status: "started"; turnId: string }
+    | { status: "failed"; error: string };
   settled: boolean;
   settledPromise: Promise<void>;
   resolveSettled: () => void;
@@ -58,28 +54,6 @@ 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;
   }
@@ -303,16 +277,10 @@ 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/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;
@@ -226,64 +226,6 @@ 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" } }),
@@ -561,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);
 }
@@ -2030,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")),
@@ -4941,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);
@@ -4993,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,
@@ -5062,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);
@@ -5222,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);
@@ -5285,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);
@@ -5460,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);
@@ -5519,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);
@@ -5820,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" } },
     ]);
   });
 
@@ -6658,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();
       }
@@ -4456,18 +4440,9 @@ export class CodexAppServerAgentSession implements AgentSession {
   ): Promise<SteerResult> {
     const client = this.client;
     const threadId = this.currentThreadId;
-    let nativeTurnId = this.currentTurnId;
+    const nativeTurnId = this.currentTurnId;
     const foregroundTurnId = this.activeForegroundTurnId;
-    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) {
+    if (!client || !threadId || !nativeTurnId || foregroundTurnId !== options.expectedTurnId) {
       return { status: "unavailable" };
     }
     if (await this.resolveSlashCommandInvocation(prompt)) return { status: "unavailable" };
@@ -4584,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,
@@ -4972,7 +4956,6 @@ export class CodexAppServerAgentSession implements AgentSession {
     } else {
       await this.ensureThread();
     }
-    await this.ensureHistoryIndexed();
 
     await revertCodexConversation({
       client: this.client,
@@ -4987,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();
       },
     });
   }
@@ -5226,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();
       }
@@ -5421,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);
@@ -6600,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/daemon-e2e/codex-steer-starting.real.e2e.test.ts

@@ -1,80 +0,0 @@
-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/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