test(codex): reproduce agent prompts restarting auto-compaction (#5728)

A prompt sent with send_agent_prompt's default replaces the running turn.
While Codex auto-compacts, each one aborts the compaction and the next
turn starts another, and the replaced prompt never reaches Codex. The
steer variant compacts once and delivers every message.

The send_agent_prompt case fails on main; the steer case passes.

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

@@ -890,131 +890,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

@@ -2448,7 +2448,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);
@@ -2456,18 +2456,14 @@ 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 {
-        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);
-      }
+      pendingRun.start = { status: "failed", error: errorMsg };
+      await this.handleStreamEvent(agent, {
+        type: "turn_failed",
+        provider: agent.provider,
+        error: errorMsg,
+      });
+      this.finalizeForegroundTurn(agent);
+      this.runs.settleForegroundRun(agentId, pendingRun.token);
       throw error;
     }
   }
@@ -2528,7 +2524,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";
@@ -2734,9 +2730,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-compaction-interrupt.local.e2e.test.ts

@@ -0,0 +1,248 @@
+import { describe, expect, test, vi } from "vitest";
+import { execFileSync } from "node:child_process";
+import os from "node:os";
+import path from "node:path";
+import { createServer, type ServerResponse } from "node:http";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { once } from "node:events";
+
+import { AgentManager } from "../agent-manager.js";
+import { AgentStorage } from "../agent-storage.js";
+import { sendPromptToAgent } from "../agent-prompt.js";
+import { CodexAppServerAgentClient } from "./codex-app-server-agent.js";
+import { createTestLogger } from "../../../test-utils/test-logger.js";
+
+function isCodexInstalled(): boolean {
+  try {
+    return (
+      execFileSync("which", ["codex"], { encoding: "utf8" }).trim().length > 0
+    );
+  } catch {
+    return false;
+  }
+}
+
+function sse(events: Array<Record<string, unknown>>): string {
+  return events
+    .map(
+      (event) =>
+        `event: ${String(event.type)}\ndata: ${JSON.stringify(event)}\n\n`
+    )
+    .join("");
+}
+
+function assistantSse(id: string, text: string, totalTokens: number): string {
+  return sse([
+    { type: "response.created", response: { id } },
+    {
+      type: "response.output_item.done",
+      item: {
+        type: "message",
+        role: "assistant",
+        id: `msg-${id}`,
+        content: [{ type: "output_text", text }],
+      },
+    },
+    {
+      type: "response.completed",
+      response: {
+        id,
+        usage: {
+          input_tokens: totalTokens,
+          input_tokens_details: null,
+          output_tokens: 0,
+          output_tokens_details: null,
+          total_tokens: totalTokens,
+        },
+      },
+    },
+  ]);
+}
+
+function isCompactionRequest(body: string): boolean {
+  return body.includes("CONTEXT CHECKPOINT COMPACTION");
+}
+
+async function startServer() {
+  const requests: Array<{ compaction: boolean; body: string }> = [];
+  const heldCompactions: ServerResponse[] = [];
+  let index = 0;
+  const server = createServer((req, res) => {
+    const chunks: Buffer[] = [];
+    req.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
+    req.on("end", () => {
+      const body = Buffer.concat(chunks).toString("utf8");
+      if (req.method !== "POST" || req.url !== "/v1/responses") {
+        res.statusCode = 404;
+        res.end();
+        return;
+      }
+      const compaction = isCompactionRequest(body);
+      requests.push({ compaction, body });
+      res.statusCode = 200;
+      res.setHeader("content-type", "text/event-stream");
+      if (compaction) {
+        // Hold the summarization open, as a slow real compaction would be.
+        heldCompactions.push(res);
+        return;
+      }
+      index += 1;
+      // The first turn reports a context far above the auto-compact limit.
+      res.end(
+        assistantSse(
+          `resp-${index}`,
+          `reply ${index}`,
+          index === 1 ? 50_000 : 100
+        )
+      );
+    });
+  });
+  server.listen(0, "127.0.0.1");
+  await once(server, "listening");
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("no address");
+  return {
+    url: `http://127.0.0.1:${address.port}`,
+    requests,
+    releaseCompactions() {
+      for (const res of heldCompactions.splice(0)) {
+        res.end(assistantSse("summary", "SUMMARY", 100));
+      }
+    },
+    close: () =>
+      new Promise<void>((resolve) => {
+        for (const res of heldCompactions.splice(0)) res.destroy();
+        server.closeAllConnections();
+        server.close(() => resolve());
+      }),
+  };
+}
+
+function writeConfig(codexHome: string, serverUrl: string): void {
+  writeFileSync(
+    path.join(codexHome, "config.toml"),
+    `
+model = "mock-model"
+approval_policy = "on-request"
+sandbox_mode = "read-only"
+model_provider = "mock_provider"
+model_auto_compact_token_limit = 10000
+
+[model_providers.mock_provider]
+name = "Mock provider for test"
+base_url = "${serverUrl}/v1"
+wire_api = "responses"
+request_max_retries = 0
+stream_max_retries = 0
+`
+  );
+}
+
+async function waitFor(
+  check: () => boolean,
+  label: string,
+  timeoutMs = 15_000
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!check()) {
+    if (Date.now() > deadline)
+      throw new Error(`Timed out waiting for ${label}`);
+    await new Promise((resolve) => setTimeout(resolve, 50));
+  }
+}
+
+describe("Codex compaction and incoming prompts (local e2e)", () => {
+  for (const behavior of ["send_agent_prompt default", "steer"] as const) {
+    test.runIf(isCodexInstalled())(
+      `${behavior}: messages from other agents during auto-compaction do not restart it (#5728)`,
+      async () => {
+        const cwd = mkdtempSync(path.join(os.tmpdir(), "codex-compact-cwd-"));
+        const codexHome = mkdtempSync(
+          path.join(os.tmpdir(), "codex-compact-home-")
+        );
+        const server = await startServer();
+        const logger = createTestLogger();
+        const storage = new AgentStorage(path.join(cwd, "agents"), logger);
+        const manager = new AgentManager({
+          clients: { codex: new CodexAppServerAgentClient(logger) },
+          registry: storage,
+          logger,
+        });
+        let agentId: string | undefined;
+        try {
+          writeConfig(codexHome, server.url);
+          vi.stubEnv("CODEX_HOME", codexHome);
+          const created = await manager.createAgent(
+            { provider: "codex", cwd, modeId: "auto", model: "mock-model" },
+            undefined,
+            { workspaceId: undefined }
+          );
+          agentId = created.id;
+          const id = agentId;
+          await manager.runAgent(id, "first turn");
+
+          const send = (text: string) =>
+            sendPromptToAgent({
+              agentManager: manager,
+              agentStorage: storage,
+              agentId: id,
+              prompt: text,
+              ...(behavior === "steer"
+                ? { activeTurnBehavior: "steer" as const }
+                : {}),
+              logger,
+            });
+          const compactions = () =>
+            server.requests.filter((r) => r.compaction).length;
+
+          await send("second turn");
+          await waitFor(() => compactions() === 1, "first compaction request");
+
+          // Two more messages arrive from other agents while it compacts.
+          await send("update from agent A");
+          await new Promise((resolve) => setTimeout(resolve, 1500));
+          await send("update from agent B");
+          await new Promise((resolve) => setTimeout(resolve, 1500));
+
+          // Each message that replaced the turn aborted the compaction and started another.
+          expect(compactions()).toBe(1);
+          server.releaseCompactions();
+          await waitFor(
+            () => manager.getAgent(id)?.lifecycle === "idle",
+            "agent idle",
+            20_000
+          );
+
+          // Every message reaches Codex and the chat, once compaction finishes.
+          const delivered = server.requests
+            .filter((request) => !request.compaction)
+            .map((request) => request.body)
+            .join("\n");
+          const chat = manager
+            .getTimeline(id)
+            .flatMap((item) =>
+              item.type === "user_message" ? [item.text] : []
+            );
+          for (const text of [
+            "second turn",
+            "update from agent A",
+            "update from agent B",
+          ]) {
+            expect(delivered).toContain(text);
+            expect(chat).toContain(text);
+          }
+          expect(compactions()).toBe(1);
+        } finally {
+          if (agentId && manager.getAgent(agentId))
+            await manager.closeAgent(agentId);
+          await storage.flush();
+          vi.unstubAllEnvs();
+          await server.close();
+          rmSync(cwd, { recursive: true, force: true });
+          rmSync(codexHome, { recursive: true, force: true });
+        }
+      },
+      60_000
+    );
+  }
+});