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

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

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

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

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

@@ -2448,7 +2448,7 @@ export class AgentManager {
         throw error;
       }
       if (isStaleProviderSessionError(error)) {
-        pendingRun.start = { status: "failed", error: error.message };
+        this.runs.settleForegroundStart(pendingRun, { status: "failed", error: error.message });
         agent.pendingReplacement = false;
         if (!agent.activeForegroundTurnId) agent.lifecycle = "idle";
         this.runs.settleForegroundRun(agentId, pendingRun.token);
@@ -2456,14 +2456,18 @@ export class AgentManager {
       }
       agent.pendingReplacement = false;
       const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
-      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);
+      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);
+      }
       throw error;
     }
   }
@@ -2524,7 +2528,7 @@ export class AgentManager {
         agent.pendingReplacement = false;
       }
       const turnStartedAt = new Date();
-      pendingRun.start = { status: "started", turnId };
+      this.runs.settleForegroundStart(pendingRun, { status: "started", turnId });
       agent.activeForegroundTurnId = turnId;
       this.openActiveTurn(agent, turnId, turnStartedAt);
       agent.lifecycle = "running";
@@ -2730,6 +2734,9 @@ export class AgentManager {
     prompt: AgentPromptInput,
     options?: AgentSteerOptions,
   ): Promise<ActiveTurnSteerDispatchResult> {
+    // A turn that is still starting has no id to steer yet. Replacing it would drop its
+    // prompt, so wait until it is accepted (or fails) and steer the turn it became.
+    await this.runs.waitForForegroundStart(agentId);
     const agent = this.requireSessionAgent(agentId);
     const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId;
     if (!expectedTurnId) {

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

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

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

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

packages/server/src/server/agent/providers/codex-compaction-interrupt.local.e2e.test.ts

@@ -1,248 +0,0 @@
-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
-    );
-  }
-});