fix(claude): preserve replies across concurrent session startup

Share initialization and bind prompt input, output consumption, and cleanup to one runtime so overlapping command discovery cannot leave a conversation writing to a process whose replies are never consumed.
2026-10-14 13:22:37 +00:00 · 2026-10-07 14:41:43 +02:00
parent 8ddeb79c8d
commit e3c82d5fe6
5 changed files with 667 additions and 243 deletions
@@ -1,3 +1,10 @@
+import type {
+  NonNullableUsage,
+  Options,
+  Query,
+  SDKMessage,
+  SDKUserMessage,
+} from "@anthropic-ai/claude-agent-sdk";
 import { afterEach, expect, test, vi } from "vitest";

 import { createTestLogger } from "../../../../test-utils/test-logger.js";
@@ -276,6 +283,453 @@ afterEach(() => {
  queryFactory.mockReset();
 });

+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+function unsupportedQueryOperation(): never {
+  throw new Error("Unexpected query operation in runtime lifecycle test");
+}
+
+interface RuntimeQueryHooks {
+  initialize?: (flags: Parameters<Query["applyFlagSettings"]>[0]) => Promise<void>;
+  retire?: () => Promise<void>;
+  beforeReply?: () => Promise<void>;
+}
+
+function createRuntimeQuery(prompt: AsyncIterable<SDKUserMessage>, hooks: RuntimeQueryHooks = {}) {
+  const usage: NonNullableUsage = {
+    input_tokens: 1,
+    output_tokens: 1,
+    cache_creation_input_tokens: 0,
+    cache_read_input_tokens: 0,
+    cache_creation: { ephemeral_1h_input_tokens: 0, ephemeral_5m_input_tokens: 0 },
+    server_tool_use: { web_fetch_requests: 0, web_search_requests: 0 },
+    service_tier: "standard",
+    inference_geo: "",
+    speed: "standard",
+    iterations: [],
+    output_tokens_details: { thinking_tokens: 0 },
+  };
+  const output = createAsyncQueue<SDKMessage>();
+  const received: string[] = [];
+  let closed = false;
+  const stream = (async function* () {
+    for (;;) {
+      const next = await output.next();
+      if (next.done) return;
+      yield next.value;
+    }
+  })();
+  const returnStream = stream.return.bind(stream);
+  const query: Query = Object.assign(stream, {
+    interrupt: async () => undefined,
+    setPermissionMode: async () => undefined,
+    setModel: async () => undefined,
+    applyFlagSettings: async (flags: Parameters<Query["applyFlagSettings"]>[0]) => {
+      await hooks.initialize?.(flags);
+    },
+    return: async () => {
+      output.end();
+      await hooks.retire?.();
+      return returnStream();
+    },
+    supportedCommands: async () => [],
+    supportedModels: async () => [],
+    close: () => {
+      closed = true;
+      output.end();
+    },
+    setMcpPermissionModeOverride: unsupportedQueryOperation,
+    setMaxThinkingTokens: unsupportedQueryOperation,
+    initializationResult: unsupportedQueryOperation,
+    reinitialize: unsupportedQueryOperation,
+    supportedAgents: unsupportedQueryOperation,
+    mcpServerStatus: unsupportedQueryOperation,
+    getContextUsage: unsupportedQueryOperation,
+    usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: unsupportedQueryOperation,
+    readFile: unsupportedQueryOperation,
+    reloadPlugins: unsupportedQueryOperation,
+    reloadSkills: unsupportedQueryOperation,
+    accountInfo: unsupportedQueryOperation,
+    rewindFiles: unsupportedQueryOperation,
+    seedReadState: unsupportedQueryOperation,
+    reconnectMcpServer: unsupportedQueryOperation,
+    toggleMcpServer: unsupportedQueryOperation,
+    setMcpServers: unsupportedQueryOperation,
+    streamInput: unsupportedQueryOperation,
+    stopTask: unsupportedQueryOperation,
+    backgroundTasks: unsupportedQueryOperation,
+  });
+  void (async () => {
+    for await (const message of prompt) {
+      const text = extractPromptText(message);
+      received.push(text);
+      await hooks.beforeReply?.();
+      output.push({
+        type: "assistant",
+        uuid: crypto.randomUUID(),
+        session_id: "runtime-lifecycle",
+        parent_tool_use_id: null,
+        message: {
+          id: crypto.randomUUID(),
+          type: "message",
+          role: "assistant",
+          model: "opus",
+          content: [{ type: "text", text: `reply to ${text}`, citations: null }],
+          stop_reason: "end_turn",
+          stop_sequence: null,
+          usage,
+          container: null,
+          context_management: null,
+          diagnostics: null,
+          stop_details: null,
+        },
+      });
+      output.push({
+        type: "result",
+        subtype: "success",
+        uuid: crypto.randomUUID(),
+        session_id: "runtime-lifecycle",
+        duration_ms: 1,
+        duration_api_ms: 1,
+        is_error: false,
+        num_turns: 1,
+        result: `reply to ${text}`,
+        stop_reason: "end_turn",
+        total_cost_usd: 0,
+        usage,
+        modelUsage: {},
+        permission_denials: [],
+      });
+    }
+  })();
+  return { query, received, isClosed: () => closed };
+}
+
+test("command discovery overlapping startup preserves replies to subsequent prompts", async () => {
+  const lookup = deferred();
+  let lookups = 0;
+  const queries: ReturnType<typeof createRuntimeQuery>[] = [];
+  const session = await new ClaudeAgentClient({
+    logger: createTestLogger(),
+    resolveBinary: async () => {
+      if (++lookups === 1) await lookup.promise;
+      return "/test/claude/bin";
+    },
+    queryFactory: ({ prompt }) => {
+      if (typeof prompt === "string") throw new Error("Expected streaming input");
+      const runtime = createRuntimeQuery(prompt);
+      queries.push(runtime);
+      return runtime.query;
+    },
+  }).createSession({ provider: "claude", cwd: process.cwd() });
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  try {
+    const commands = session.listCommands!();
+    const first = session.startTurn("first");
+    // Finish all work not blocked by executable discovery before releasing it.
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    lookup.resolve();
+    await Promise.all([commands, first]);
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    await session.startTurn("hello");
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    expect(collectAssistantText(events)).toBe("reply to firstreply to hello");
+    expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(2);
+    expect(queries).toHaveLength(1);
+    expect(queries[0].received).toEqual(["first", "hello"]);
+  } finally {
+    lookup.resolve();
+    for (const runtime of queries) runtime.query.close();
+    await session.close();
+  }
+});
+
+test("closing during query initialization disposes the process before initialization settles", async () => {
+  const initializing = deferred();
+  const release = deferred();
+  const queries: ReturnType<typeof createRuntimeQuery>[] = [];
+  const session = await new ClaudeAgentClient({
+    logger: createTestLogger(),
+    resolveBinary: async () => "/test/claude/bin",
+    queryFactory: ({ prompt }) => {
+      if (typeof prompt === "string") throw new Error("Expected streaming input");
+      const runtime = createRuntimeQuery(prompt, {
+        initialize: async () => {
+          initializing.resolve();
+          await release.promise;
+        },
+      });
+      queries.push(runtime);
+      return runtime.query;
+    },
+  }).createSession({ provider: "claude", model: "claude-opus-5-5", cwd: process.cwd() });
+  const discovery = session.listCommands!();
+  const rejected = expect(discovery).rejects.toThrow("Claude session is closed");
+  try {
+    await initializing.promise;
+    await session.close();
+    expect(queries[0].isClosed()).toBe(true);
+  } finally {
+    release.resolve();
+    await rejected;
+    await session.close();
+  }
+});
+
+async function createRuntimeHarness(
+  options: {
+    resolveBinary?: () => Promise<string>;
+    hooks?: RuntimeQueryHooks;
+  } = {},
+) {
+  const launches: Options[] = [];
+  const queries: ReturnType<typeof createRuntimeQuery>[] = [];
+  const session = await new ClaudeAgentClient({
+    logger: createTestLogger(),
+    resolveBinary: options.resolveBinary ?? (async () => "/test/claude/bin"),
+    queryFactory: ({ prompt, options: launch }) => {
+      if (typeof prompt === "string") throw new Error("Expected streaming input");
+      launches.push(launch);
+      const runtime = createRuntimeQuery(prompt, options.hooks);
+      queries.push(runtime);
+      return runtime.query;
+    },
+  }).createSession({
+    provider: "claude",
+    model: "claude-opus-5-5",
+    thinkingOptionId: "low",
+    cwd: process.cwd(),
+  });
+  return { session, queries, launches };
+}
+
+test.each([true, false])(
+  "Fast mode changed to %s during startup reaches the launched query",
+  async (enabled) => {
+    const lookup = deferred();
+    const entered = deferred();
+    const applied: unknown[] = [];
+    const { session, queries } = await createRuntimeHarness({
+      resolveBinary: async () => {
+        entered.resolve();
+        await lookup.promise;
+        return "/test/claude/bin";
+      },
+      hooks: {
+        initialize: async (flags) => {
+          applied.push(flags.fastMode);
+        },
+      },
+    });
+    await session.setFeature!("fast_mode", !enabled);
+    const discovery = session.listCommands!();
+    try {
+      await entered.promise;
+      await session.setFeature!("fast_mode", enabled);
+      lookup.resolve();
+      await discovery;
+      await session.listCommands!();
+      expect(applied).toEqual([enabled]);
+      expect(queries).toHaveLength(1);
+    } finally {
+      lookup.resolve();
+      await discovery;
+      await session.close();
+    }
+  },
+);
+
+test("command discovery leaves a thinking change pending until the active turn finishes", async () => {
+  const reply = deferred();
+  const { session, queries, launches } = await createRuntimeHarness({
+    hooks: { beforeReply: () => reply.promise },
+  });
+  try {
+    const turn = collectUntilTerminal(streamSession(session, "first"));
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    await session.setThinkingOption!("high");
+    await session.listCommands!();
+    expect(queries).toHaveLength(1);
+    reply.resolve();
+    expect(collectAssistantText(await turn)).toBe("reply to first");
+    const next = await collectUntilTerminal(streamSession(session, "hello"));
+    expect(collectAssistantText(next)).toBe("reply to hello");
+    expect(launches.map((launch) => launch.effort)).toEqual(["low", "high"]);
+  } finally {
+    reply.resolve();
+    await session.close();
+  }
+});
+
+test("command discovery and a prompt share replacement until the old runtime retires", async () => {
+  const retiring = deferred();
+  const release = deferred();
+  const { session, queries } = await createRuntimeHarness({
+    hooks: {
+      retire: async () => {
+        retiring.resolve();
+        await release.promise;
+      },
+    },
+  });
+  try {
+    await collectUntilTerminal(streamSession(session, "first"));
+    await session.setThinkingOption!("high");
+    const commands = session.listCommands!();
+    await retiring.promise;
+    const next = collectUntilTerminal(streamSession(session, "second"));
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    expect(queries).toHaveLength(1);
+    release.resolve();
+    await commands;
+    expect(collectAssistantText(await next)).toBe("reply to second");
+    expect(collectAssistantText(await collectUntilTerminal(streamSession(session, "third")))).toBe(
+      "reply to third",
+    );
+    expect(queries).toHaveLength(2);
+    expect(queries[0].isClosed()).toBe(true);
+    expect(queries[1].received).toEqual(["second", "third"]);
+  } finally {
+    release.resolve();
+    await session.close();
+  }
+});
+
+test("a thinking change during startup remains pending for the following turn", async () => {
+  const lookup = deferred();
+  const entered = deferred();
+  const { session, launches } = await createRuntimeHarness({
+    resolveBinary: async () => {
+      entered.resolve();
+      await lookup.promise;
+      return "/test/claude/bin";
+    },
+  });
+  try {
+    await session.setThinkingOption!("medium");
+    const first = collectUntilTerminal(streamSession(session, "first"));
+    await entered.promise;
+    await session.setThinkingOption!("high");
+    lookup.resolve();
+    expect(collectAssistantText(await first)).toBe("reply to first");
+    expect(collectAssistantText(await collectUntilTerminal(streamSession(session, "second")))).toBe(
+      "reply to second",
+    );
+    expect(launches.map((launch) => launch.effort)).toEqual(["medium", "high"]);
+  } finally {
+    lookup.resolve();
+    await session.close();
+  }
+});
+
+test("closing during executable discovery never launches a late query", async () => {
+  const lookup = deferred();
+  const entered = deferred();
+  const { session, queries } = await createRuntimeHarness({
+    resolveBinary: async () => {
+      entered.resolve();
+      await lookup.promise;
+      return "/test/claude/bin";
+    },
+  });
+  const discovery = session.listCommands!();
+  const rejected = expect(discovery).rejects.toThrow("Claude session is closed");
+  await entered.promise;
+  await session.close();
+  lookup.resolve();
+  await rejected;
+  expect(queries).toHaveLength(0);
+  await expect(session.listCommands!()).rejects.toThrow("Claude session is closed");
+});
+
+test("a failed shared initialization releases the runtime for a subsequent prompt", async () => {
+  let attempts = 0;
+  const { session, queries } = await createRuntimeHarness({
+    hooks: {
+      initialize: async () => {
+        if (++attempts === 1) throw new Error("initialization rejected");
+      },
+    },
+  });
+  try {
+    const discovery = session.listCommands!();
+    const rejected = expect(discovery).rejects.toThrow("initialization rejected");
+    const failed = await collectUntilTerminal(streamSession(session, "first"));
+    await rejected;
+    expect(failed.at(-1)).toMatchObject({ type: "turn_failed", error: "initialization rejected" });
+    expect(queries[0].isClosed()).toBe(true);
+    expect(queries[0].received).toEqual([]);
+    const recovered = await collectUntilTerminal(streamSession(session, "retry"));
+    expect(collectAssistantText(recovered)).toBe("reply to retry");
+    expect(queries).toHaveLength(2);
+  } finally {
+    await session.close();
+  }
+});
+
+test("a new turn waits for a naturally ended runtime to finish cleanup", async () => {
+  const retiring = deferred();
+  const release = deferred();
+  const { session, queries } = await createRuntimeHarness({
+    hooks: {
+      retire: async () => {
+        retiring.resolve();
+        await release.promise;
+      },
+    },
+  });
+  try {
+    await collectUntilTerminal(streamSession(session, "first"));
+    queries[0].query.close();
+    await retiring.promise;
+    const next = collectUntilTerminal(streamSession(session, "second"));
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    expect(queries).toHaveLength(1);
+    release.resolve();
+    expect(collectAssistantText(await next)).toBe("reply to second");
+    expect(queries).toHaveLength(2);
+  } finally {
+    release.resolve();
+    await session.close();
+  }
+});
+
+test("an interrupted startup failure does not finish its replacement turn twice", async () => {
+  const entered = deferred();
+  const release = deferred();
+  const { session } = await createRuntimeHarness({
+    hooks: {
+      initialize: async () => {
+        entered.resolve();
+        await release.promise;
+        throw new Error("initialization rejected");
+      },
+    },
+  });
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  try {
+    const first = session.startTurn("first");
+    await entered.promise;
+    await session.interrupt();
+    const replacement = session.startTurn("second");
+    release.resolve();
+    await Promise.all([first, replacement]);
+    expect(events.filter((event) => event.type === "turn_canceled")).toHaveLength(1);
+    expect(events.filter((event) => event.type === "turn_failed")).toHaveLength(1);
+  } finally {
+    release.resolve();
+    await session.close();
+  }
+});
+
 test("interrupt only calls query.interrupt and leaves the query open", async () => {
  const logger = createTestLogger();
  const queries: ScriptedQuery[] = [];
@@ -143,10 +143,6 @@ function getLatestCompletedBashCall(events: AgentStreamEvent[]): ToolCallTimelin
    .find((item) => item.status === "completed" && item.name.toLowerCase() === "bash");
 }

-function getInternalQuery(session: AgentSession): unknown {
-  return (session as AgentSession & { query?: unknown }).query ?? null;
-}
-
 async function createSession(params?: {
  cwdPrefix?: string;
  modeId?: string;
@@ -316,7 +312,7 @@ describe("ClaudeAgentSession integration", () => {
    }
  }, 60_000);

-  test("interrupts a running Bash turn and continues on the same query", async () => {
+  test("interrupts a running Bash turn and continues the same conversation", async () => {
    const handle = await createSession({
      cwdPrefix: "claude-agent-interrupt-continue-",
    });
@@ -340,9 +336,9 @@ describe("ClaudeAgentSession integration", () => {
          event.item.name.toLowerCase() === "bash",
        45_000,
      );
-      const firstQuery = getInternalQuery(handle.session);
+      const firstSessionId = (await handle.session.getRuntimeInfo()).sessionId;

-      expect(firstQuery).toBeTruthy();
+      expect(firstSessionId).toEqual(expect.any(String));

      await handle.session.interrupt();

@@ -360,9 +356,9 @@ describe("ClaudeAgentSession integration", () => {
      const followUpEvents = await collectUntilTerminal(
        streamSession(handle.session, "Respond with exactly: AFTER_INTERRUPT_OK"),
      );
-      const secondQuery = getInternalQuery(handle.session);
+      const secondSessionId = (await handle.session.getRuntimeInfo()).sessionId;

-      expect(secondQuery).toBe(firstQuery);
+      expect(secondSessionId).toBe(firstSessionId);
      expect(compactText(getAssistantText(followUpEvents))).toContain("after_interrupt_ok");
      expect(followUpEvents.at(-1)).toMatchObject({
        type: "turn_completed",
@@ -478,31 +478,23 @@ test("logs redacted query summary and never leaks sentinel secrets", async () =>
 test("interruptActiveTurn only interrupts the active query without info logs", async () => {
  const spy = createSpyLogger();
  const session = await createSessionWithLogger(spy.logger);
-  const internal: {
-    query: {
-      interrupt: () => Promise<void>;
-      return?: () => Promise<void>;
-      close?: () => void;
-    } | null;
-    input: { end: () => void } | null;
-    queryRestartNeeded: boolean;
-    mainTurnInFlight: boolean;
-    interruptActiveTurn: () => Promise<void>;
-  } = asInternals(session);
-  const interrupt = vi.fn(async () => undefined);
-  const queryReturn = vi.fn(async () => undefined);
-  const end = vi.fn(() => undefined);
-  internal.query = {
-    interrupt,
-    return: queryReturn,
-    close: vi.fn(() => undefined),
-  };
-  internal.input = { end };
-  internal.queryRestartNeeded = false;
-  internal.mainTurnInFlight = true;
+  let acknowledgeInit!: () => void;
+  const initialized = new Promise<void>((resolve) => {
+    acknowledgeInit = resolve;
+  });
+  const activeQuery = createBaseQueryMock(
+    vi.fn(() => {
+      acknowledgeInit();
+      return new Promise<never>(() => {});
+    }),
+  );
+  activeQuery.next.mockResolvedValueOnce({ done: false, value: claudeTurnInit() });
+  sdkQueryFactory.mockImplementation(() => activeQuery);
+  await session.startTurn("hello");
+  await initialized;

  try {
-    await internal.interruptActiveTurn();
+    await session.interrupt();

    const interruptInfoMessages = extractStringLogArgs(spy.info.mock.calls).filter((message) =>
      message.includes("interruptActiveTurn"),
@@ -513,12 +505,11 @@ test("interruptActiveTurn only interrupts the active query without info logs", a

    expect(interruptInfoMessages).toEqual([]);
    expect(interruptDebugMessages).toEqual([]);
-    expect(interrupt).toHaveBeenCalledTimes(1);
-    expect(queryReturn).not.toHaveBeenCalled();
-    expect(end).not.toHaveBeenCalled();
-    expect(internal.query).not.toBeNull();
-    expect(internal.input).not.toBeNull();
-    expect(internal.queryRestartNeeded).toBe(false);
+    expect(activeQuery.interrupt).toHaveBeenCalledTimes(1);
+    expect(activeQuery.return).not.toHaveBeenCalled();
+    expect(activeQuery.close).not.toHaveBeenCalled();
+    await session.listCommands!();
+    expect(sdkQueryFactory).toHaveBeenCalledTimes(1);
  } finally {
    await session.close();
  }
@@ -2057,6 +2057,14 @@ class ClaudeContextUsageState {
  }
 }

+interface ClaudeRuntime {
+  query: Query;
+  input: AsyncMessageInput<SDKUserMessage>;
+  process: { child: ChildProcess | null };
+  pump: Promise<void> | null;
+  closing: Promise<void> | null;
+}
+
 class ClaudeAgentSession implements AgentSession {
  readonly provider = "claude" as const;
  readonly capabilities = CLAUDE_CAPABILITIES;
@@ -2070,14 +2078,12 @@ class ClaudeAgentSession implements AgentSession {
  private readonly logger: Logger;
  private readonly queryFactory?: ClaudeQueryFactory;
  private readonly resolveBinary: () => Promise<string>;
-  private query: Query | null = null;
+  private runtime: ClaudeRuntime | null = null;
+  private runtimeStarting: Promise<ClaudeRuntime> | null = null;
  private readonly harnessEnvironment: Record<string, string>;
  private readonly usageSessionKey = randomUUID();
-  private childProcess: ChildProcess | null = null;
-  private input: AsyncMessageInput<SDKUserMessage> | null = null;
-  /** The exact SDK query/input pair that owns the current foreground turn. */
-  private activeForegroundQuery: Query | null = null;
-  private activeForegroundInput: AsyncMessageInput<SDKUserMessage> | null = null;
+  /** The runtime that owns the foreground or autonomous turn. */
+  private activeTurnRuntime: ClaudeRuntime | null = null;
  /**
   * Messages pushed into the live SDK input (steers, and a foreground turn's own prompt) whose
   * command_lifecycle has not reached "started". Interrupting has to withdraw them, or the SDK
@@ -2133,7 +2139,6 @@ class ClaudeAgentSession implements AgentSession {
  private lastRuntimeModel: string | null = null;
  private compacting = false;
  private compactionMarkerOpen = false;
-  private queryPumpPromise: Promise<void> | null = null;
  private queryRestartNeeded = false;
  private pendingInterruptAbort = false;
  /**
@@ -2319,19 +2324,15 @@ class ClaudeAgentSession implements AgentSession {
    this.notifySubscribers({ type: "turn_started", provider: "claude" });

    try {
-      await this.ensureQuery();
-      if (cancelIssued) {
-        // Stopped while Claude was still starting up: withdrawn by never sending it.
+      const runtime = await this.ensureRuntime();
+      if (this.closed || cancelIssued || this.activeForegroundTurnId !== turnId) {
+        // Stopped during startup: withdraw this turn's message by never sending it.
        if (sdkUserMessageId) this.unstartedMessageUuids.delete(sdkUserMessageId);
        return { turnId };
      }
-      if (!this.input) {
-        throw new Error("Claude session input stream not initialized");
-      }
-      this.activeForegroundQuery = this.query;
-      this.activeForegroundInput = this.input;
-      this.startQueryPump();
-      this.input.push(sdkMessage);
+      this.activeTurnRuntime = runtime;
+      runtime.input.push(sdkMessage);
+      this.startQueryPump(runtime);
      setTimeout(() => {
        if (this.activeForegroundTurnId === turnId) {
          this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId);
@@ -2339,9 +2340,13 @@ class ClaudeAgentSession implements AgentSession {
      }, 0);
    } catch (error) {
      if (sdkUserMessageId) this.unstartedMessageUuids.delete(sdkUserMessageId);
-      this.finishForegroundTurn(
-        this.buildTurnFailedEvent(error instanceof Error ? error.message : "Claude stream failed"),
-      );
+      if (this.activeForegroundTurnId === turnId) {
+        this.finishForegroundTurn(
+          this.buildTurnFailedEvent(
+            error instanceof Error ? error.message : "Claude stream failed",
+          ),
+        );
+      }
    }

    return { turnId };
@@ -2361,23 +2366,20 @@ class ClaudeAgentSession implements AgentSession {

    // Capture both ends of the live SDK stream before creating or delivering the message. There
    // is deliberately no await below: a finished A cannot make this input point at a later B.
-    const query = this.activeForegroundQuery;
-    const input = this.activeForegroundInput;
-    if (!query || !input || this.query !== query || this.input !== input) {
+    const runtime = this.activeTurnRuntime;
+    if (!runtime || this.runtime !== runtime) {
      return { status: "unavailable" };
    }
    const message = this.toSdkUserMessage(prompt);
    message.priority = "next";
    if (
      (this.activeForegroundTurnId ?? this.autonomousTurn?.id) !== options.expectedTurnId ||
-      this.activeForegroundQuery !== query ||
-      this.activeForegroundInput !== input ||
-      this.query !== query ||
-      this.input !== input
+      this.activeTurnRuntime !== runtime ||
+      this.runtime !== runtime
    ) {
      return { status: "unavailable" };
    }
-    this.enqueueSteer(input, message, options.clearPendingPermissions === true);
+    this.enqueueSteer(runtime.input, message, options.clearPendingPermissions === true);
    return { status: "accepted" };
  }

@@ -2469,7 +2471,7 @@ class ClaudeAgentSession implements AgentSession {
    const normalized = isPermissionMode(modeId) ? modeId : "default";
    assertClaudeModeCanRun(normalized, this.harnessEnvironment);
    const previousMode = this.currentMode;
-    const launchesQuery = !this.query || this.queryRestartNeeded;
+    const launchesQuery = !this.runtime || this.queryRestartNeeded;
    const activeQuery = await this.ensureQuery(normalized);
    try {
      await activeQuery.setPermissionMode(normalized);
@@ -2568,7 +2570,7 @@ class ClaudeAgentSession implements AgentSession {
      ...this.config.featureValues,
      fast_mode: enabled,
    };
-    const activeQuery = query ?? this.query;
+    const activeQuery = query ?? this.runtime?.query;
    if (activeQuery) {
      await activeQuery.applyFlagSettings({ fastMode: enabled });
    }
@@ -2731,8 +2733,8 @@ class ClaudeAgentSession implements AgentSession {
        sessionId: this.claudeSessionId,
        turnId: this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? undefined,
        turnState: this.turnState,
-        hasQuery: Boolean(this.query),
-        hasInput: Boolean(this.input),
+        hasQuery: Boolean(this.runtime),
+        hasInput: Boolean(this.runtime),
        hasActiveForegroundTurnId: Boolean(this.activeForegroundTurnId),
      },
      "provider.claude.session_close.start",
@@ -2742,35 +2744,17 @@ class ClaudeAgentSession implements AgentSession {
    this.cancelCurrentTurn?.();
    this.subscribers.clear();
    this.activeForegroundTurnId = null;
-    this.activeForegroundQuery = null;
-    this.activeForegroundInput = null;
+    this.activeTurnRuntime = null;
    this.autonomousTurn = null;
    this.cancelCurrentTurn = null;
    this.turnState = "idle";
    this.sidechainTracker.clear();
    this.taskProtocolSource.reset();
-    this.input?.end();
-    this.query?.close?.();
-    await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt");
-    await this.awaitWithTimeout(this.query?.return?.(), "close query return");
-    this.query = null;
-    this.input = null;
-    // Terminate the entire process tree (claude + MCP children) to prevent
-    // orphan accumulation. The SDK's internal cleanup may only kill the
-    // direct child process.
-    if (this.childProcess) {
-      const result = await terminateWithTreeKill(this.childProcess, {
-        gracefulTimeoutMs: 2_000,
-        forceTimeoutMs: 2_000,
-      });
-      if (result === "kill-timeout") {
-        this.logger.warn(
-          { pid: this.childProcess.pid, agentId: this.agentId },
-          "Claude process tree did not report exit after SIGKILL",
-        );
-      }
-      this.childProcess = null;
-    }
+    const runtime = this.runtime;
+    this.runtime = null;
+    if (runtime) await this.disposeRuntime(runtime);
+    // A pending launch checks closed before spawning and after initialization. Do not
+    // wait on executable discovery here: closing must work even when discovery stalls.
    if (this.persistSession === false && this.claudeSessionId) {
      // Claude Code currently ignores --no-session-persistence outside --print mode
      // (see `claude --help`), so the SDK's persistSession=false is silently dropped
@@ -2984,7 +2968,7 @@ class ClaudeAgentSession implements AgentSession {
  }

  private async ensureFreshQuery(): Promise<Query> {
-    if (this.query) {
+    if (this.runtime) {
      this.queryRestartNeeded = true;
    }
    return this.ensureQuery();
@@ -3164,75 +3148,99 @@ class ClaudeAgentSession implements AgentSession {
  }

  private async ensureQuery(launchMode: PermissionMode = this.currentMode): Promise<Query> {
-    if (this.query && !this.queryRestartNeeded) {
-      return this.query;
+    return (await this.ensureRuntime(launchMode)).query;
+  }
+
+  private ensureRuntime(launchMode: PermissionMode = this.currentMode): Promise<ClaudeRuntime> {
+    if (this.closed) return Promise.reject(new Error("Claude session is closed"));
+    if (this.runtimeStarting) return this.runtimeStarting;
+    if (
+      this.runtime &&
+      !this.runtime.closing &&
+      (!this.queryRestartNeeded || this.activeTurnRuntime === this.runtime)
+    ) {
+      return Promise.resolve(this.runtime);
    }

-    if (this.queryRestartNeeded && this.query) {
-      const oldQuery = this.query;
-      const oldInput = this.input;
-      // Null out query/input BEFORE awaiting the old iterator's return so the
-      // old pump sees this.query !== activeQuery and skips failActiveTurns.
-      this.query = null;
-      this.input = null;
-      this.queryPumpPromise = null;
-      this.queryRestartNeeded = false;
-      // Ending the input retires the process on purpose. Detach first so its
-      // exit is not reported as a crash.
-      const retiredChild = this.childProcess;
-      this.childProcess = null;
-      if (retiredChild) this.failRunningRuntimeTasks();
-      oldInput?.end();
-      oldQuery.close?.();
-      try {
-        await oldQuery.return?.();
-      } catch {
-        /* ignore */
-      }
-      // Tree-kill the old process tree now that the SDK has cleaned up.
-      // If we skip this, MCP children of the previous claude process can
-      // survive as orphans when the session spawns a replacement query.
-      if (retiredChild) {
-        await terminateWithTreeKill(retiredChild, {
-          gracefulTimeoutMs: 2_000,
-          forceTimeoutMs: 2_000,
-        }).catch(() => {
-          /* process may already be dead */
-        });
-      }
-    }
+    const starting = this.createRuntime(launchMode).finally(() => {
+      if (this.runtimeStarting === starting) this.runtimeStarting = null;
+    });
+    this.runtimeStarting = starting;
+    return starting;
+  }

-    // Preserve claudeSessionId across query recreation so buildOptions() passes
-    // resume: sessionId and the new query continues the existing conversation.
+  private async createRuntime(launchMode: PermissionMode): Promise<ClaudeRuntime> {
+    const retired = this.runtime;
+    this.runtime = null;
+    if (retired) {
+      this.failRunningRuntimeTasks();
+      await this.disposeRuntime(retired);
+    }
+    if (this.closed) throw new Error("Claude session is closed");
+
+    // Consume only the settings captured by this launch. A change during an await
+    // remains pending for the next turn, and the pump never initiates a restart.
+    this.queryRestartNeeded = false;
    this.persistence = null;
-
-    const input = createAsyncMessageInput<SDKUserMessage>();
    const options = await this.buildOptions(launchMode);
+    if (this.closed) throw new Error("Claude session is closed");
    this.logger.debug({ options: summarizeClaudeOptionsForLog(options) }, "claude query");
-    this.input = input;
    // A fresh Claude process has no turn of its own in flight.
    this.mainTurnInFlight = false;
-    this.query = claudeQuery(
+    const input = createAsyncMessageInput<SDKUserMessage>();
+    const process: ClaudeRuntime["process"] = { child: null };
+    const query = claudeQuery(
      { prompt: input.iterable, options },
      {
        runtimeSettings: this.runtimeSettings,
        launchEnv: this.launchEnv,
        queryFactory: this.queryFactory,
        onChildProcess: (child) => {
-          this.childProcess = child;
+          process.child = child;
          child.once("exit", (code, signal) => this.handleRuntimeExit(child, code, signal));
        },
      },
    );
-    const fastMode = this.resolveFastModeSetting();
-    if (fastMode !== null) {
-      await this.query.applyFlagSettings({ fastMode });
+    const runtime: ClaudeRuntime = { query, input, process, pump: null, closing: null };
+    this.runtime = runtime;
+    try {
+      const fastMode = this.resolveFastModeSetting();
+      if (fastMode !== null) await query.applyFlagSettings({ fastMode });
+      if (this.closed) throw new Error("Claude session is closed");
+      if (this.runtime !== runtime) throw new Error("Claude runtime exited during initialization");
+      return runtime;
+    } catch (error) {
+      if (this.runtime === runtime) this.runtime = null;
+      await this.disposeRuntime(runtime);
+      throw error;
+    }
+  }
+
+  private disposeRuntime(runtime: ClaudeRuntime): Promise<void> {
+    runtime.closing ??= this.closeRuntime(runtime);
+    return runtime.closing;
+  }
+
+  private async closeRuntime(runtime: ClaudeRuntime): Promise<void> {
+    const child = runtime.process.child;
+    runtime.process.child = null;
+    runtime.input.end();
+    runtime.query.close?.();
+    await this.awaitWithTimeout(runtime.query.interrupt?.(), "close query interrupt");
+    await this.awaitWithTimeout(runtime.query.return?.(), "close query return");
+    // Claude's SDK cleanup does not own MCP children; retire the entire tree.
+    if (child) {
+      const result = await terminateWithTreeKill(child, {
+        gracefulTimeoutMs: 2_000,
+        forceTimeoutMs: 2_000,
+      });
+      if (result === "kill-timeout") {
+        this.logger.warn(
+          { pid: child.pid, agentId: this.agentId },
+          "Claude process tree did not report exit after SIGKILL",
+        );
+      }
    }
-    // Do not kick off background control-plane queries here. Methods like
-    // supportedCommands()/setPermissionMode() may execute immediately after
-    // ensureQuery() (for listCommands()/setMode()), and sharing the same query
-    // control plane can cause those calls to wait behind supportedModels().
-    return this.query;
  }

  private async awaitWithTimeout(
@@ -3351,17 +3359,6 @@ class ClaudeAgentSession implements AgentSession {
    const sdkEnv = this.harnessEnvironment;
    assertClaudeModeCanRun(permissionMode, sdkEnv);

-    const claudeBinary = await this.resolveBinary();
-    this.logger.debug(
-      {
-        claudeBinary,
-        pathEnvKey: resolvePathEnvKey(),
-        pathIncludesClaudeLocalBin: (process.env["Path"] ?? process.env["PATH"] ?? "")
-          .toLowerCase()
-          .includes("\\.local\\bin"),
-      },
-      "Resolved Claude executable",
-    );
    const sessionBinding: Pick<ClaudeOptions, "resume" | "sessionId"> = {};
    if (this.pendingFreshSessionId) {
      sessionBinding.sessionId = this.pendingFreshSessionId;
@@ -3379,7 +3376,6 @@ class ClaudeAgentSession implements AgentSession {
      allowDangerouslySkipPermissions: true,
      agents: this.defaults?.agents,
      canUseTool: this.handlePermissionRequest,
-      pathToClaudeCodeExecutable: claudeBinary,
      // Use Claude Code preset system prompt and load CLAUDE.md files
      // Append provider-agnostic system prompts for agents.
      systemPrompt: {
@@ -3429,7 +3425,18 @@ class ClaudeAgentSession implements AgentSession {
        ...this.runtimeSettings.disallowedTools,
      ];
    }
-    return base;
+    const claudeBinary = await this.resolveBinary();
+    this.logger.debug(
+      {
+        claudeBinary,
+        pathEnvKey: resolvePathEnvKey(),
+        pathIncludesClaudeLocalBin: (process.env["Path"] ?? process.env["PATH"] ?? "")
+          .toLowerCase()
+          .includes("\\.local\\bin"),
+      },
+      "Resolved Claude executable",
+    );
+    return { pathToClaudeCodeExecutable: claudeBinary, ...base };
  }

  private buildSettingsOptions(
@@ -3685,8 +3692,7 @@ class ClaudeAgentSession implements AgentSession {
    }
    this.notifySubscribers(event);
    this.activeForegroundTurnId = null;
-    this.activeForegroundQuery = null;
-    this.activeForegroundInput = null;
+    this.activeTurnRuntime = null;
    this.cancelCurrentTurn = null;
    this.activeTurnHasAssistantText = false;
    this.compactionMarkerOpen = false;
@@ -3704,15 +3710,13 @@ class ClaudeAgentSession implements AgentSession {
      this.compactionMarkerOpen = false;
      if (this.activeForegroundTurnId) {
        this.activeForegroundTurnId = null;
-        this.activeForegroundQuery = null;
-        this.activeForegroundInput = null;
+        this.activeTurnRuntime = null;
        this.cancelCurrentTurn = null;
        this.activeTurnHasAssistantText = false;
        this.syncTurnState("foreground turn terminal");
      } else if (this.autonomousTurn) {
        this.autonomousTurn = null;
-        this.activeForegroundQuery = null;
-        this.activeForegroundInput = null;
+        this.activeTurnRuntime = null;
        this.activeTurnHasAssistantText = false;
        this.syncTurnState("autonomous turn terminal");
      }
@@ -3726,8 +3730,7 @@ class ClaudeAgentSession implements AgentSession {
    this.autonomousTurn = {
      id: this.createTurnId("autonomous"),
    };
-    this.activeForegroundQuery = this.query;
-    this.activeForegroundInput = this.input;
+    this.activeTurnRuntime = this.runtime;
    this.activeTurnHasAssistantText = false;
    this.contextUsage.beginTurn();
    this.notifySubscribers({ type: "turn_started", provider: "claude" });
@@ -3740,8 +3743,7 @@ class ClaudeAgentSession implements AgentSession {
    }
    this.notifySubscribers({ type: "turn_completed", provider: "claude" });
    this.autonomousTurn = null;
-    this.activeForegroundQuery = null;
-    this.activeForegroundInput = null;
+    this.activeTurnRuntime = null;
    this.activeTurnHasAssistantText = false;
    this.compactionMarkerOpen = false;
    this.syncTurnState("autonomous turn completed");
@@ -3769,10 +3771,11 @@ class ClaudeAgentSession implements AgentSession {
    code: number | null,
    signal: NodeJS.Signals | null,
  ): void {
-    if (this.closed || this.childProcess !== child) {
+    const runtime = this.runtime;
+    if (this.closed || !runtime || runtime.process.child !== child) {
      return;
    }
-    this.childProcess = null;
+    runtime.process.child = null;
    this.logger.warn(
      { agentId: this.agentId, pid: child.pid, code, signal },
      "Claude runtime exited unexpectedly",
@@ -3781,13 +3784,19 @@ class ClaudeAgentSession implements AgentSession {
    if (this.activeForegroundTurnId || this.autonomousTurn) {
      // The pump is about to throw. It waits for stderr to flush and reports the
      // real cause; reporting here first would replace that with a bare exit code
-      // and detach this.query, which is how the pump recognizes its own stream.
+      // and detach the runtime, which is how the pump recognizes its own stream.
      return;
    }
    // Drop the dead handles so the next write spawns a fresh process instead of
    // failing against a dead transport forever.
-    this.query = null;
-    this.input = null;
+    void this.disposeRuntime(runtime)
+      .then(() => {
+        if (this.runtime === runtime) this.runtime = null;
+        return undefined;
+      })
+      .catch((error) => {
+        this.logger.warn({ err: error }, "Failed to retire exited Claude runtime");
+      });
    this.dispatchEvents([
      this.buildTurnFailedEvent(
        `Claude stopped unexpectedly (${signal ? `signal ${signal}` : `exit code ${code ?? "unknown"}`}). Any background shells, monitors or other work it had running were terminated with it.`,
@@ -3803,12 +3812,12 @@ class ClaudeAgentSession implements AgentSession {
    );
  }

-  private startQueryPump(): void {
-    if (this.closed || this.queryPumpPromise) {
+  private startQueryPump(runtime: ClaudeRuntime): void {
+    if (this.closed || runtime.pump) {
      return;
    }

-    const pump = this.runQueryPump().catch((error) => {
+    const pump = this.runQueryPump(runtime).catch((error) => {
      this.logger.trace(
        {
          agentId: this.agentId,
@@ -3821,33 +3830,16 @@ class ClaudeAgentSession implements AgentSession {
      );
    });

-    this.queryPumpPromise = pump;
+    runtime.pump = pump;
    void pump.finally(() => {
-      if (this.queryPumpPromise === pump) {
-        this.queryPumpPromise = null;
+      if (runtime.pump === pump) {
+        runtime.pump = null;
      }
    });
  }

-  private async runQueryPump(): Promise<void> {
-    let activeQuery: Query;
-    try {
-      activeQuery = await this.ensureQuery();
-    } catch (error) {
-      this.logger.trace(
-        {
-          agentId: this.agentId,
-          provider: "claude",
-          sessionId: this.claudeSessionId,
-          turnId: this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? undefined,
-          err: error,
-        },
-        "provider.claude.query_pump.init_failed",
-      );
-      this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed");
-      return;
-    }
-
+  private async runQueryPump(runtime: ClaudeRuntime): Promise<void> {
+    const activeQuery = runtime.query;
    let consecutiveInterruptAbortRecoveries = 0;
    const logRawMessage = (message: SDKMessage): void => {
      this.logger.trace(
@@ -3865,9 +3857,10 @@ class ClaudeAgentSession implements AgentSession {
      );
    };
    const handlePumpedMessage = async (message: SDKMessage): Promise<boolean> => {
+      if (this.runtime !== runtime) return true;
      logRawMessage(message);
      consecutiveInterruptAbortRecoveries = 0;
-      if (await this.handleMissingResumedConversation(message, activeQuery)) {
+      if (await this.handleMissingResumedConversation(message, runtime)) {
        return true;
      }
      await this.routeSdkMessageFromPump(message);
@@ -3882,19 +3875,19 @@ class ClaudeAgentSession implements AgentSession {
      return false;
    };
    try {
-      while (!this.closed && this.query === activeQuery) {
+      while (!this.closed && this.runtime === runtime) {
        try {
          if (await drainActiveQuery()) {
            return;
          }
-          if (!this.closed && this.query === activeQuery) {
+          if (!this.closed && this.runtime === runtime) {
            this.failActiveTurns("Claude stream ended before terminal result");
          }
          return;
        } catch (error) {
          if (
            !this.closed &&
-            this.query === activeQuery &&
+            this.runtime === runtime &&
            this.shouldRecoverInterruptedQueryAbort(error, consecutiveInterruptAbortRecoveries)
          ) {
            consecutiveInterruptAbortRecoveries += 1;
@@ -3904,7 +3897,7 @@ class ClaudeAgentSession implements AgentSession {
            );
            continue;
          }
-          if (!this.closed && this.query === activeQuery) {
+          if (!this.closed && this.runtime === runtime) {
            await this.awaitRecentStderrAfterProcessExit(error);
            this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed");
          }
@@ -3912,10 +3905,8 @@ class ClaudeAgentSession implements AgentSession {
        }
      }
    } finally {
-      if (this.query === activeQuery) {
-        this.query = null;
-        this.input = null;
-      }
+      await this.disposeRuntime(runtime);
+      if (this.runtime === runtime) this.runtime = null;
    }
  }

@@ -4087,7 +4078,7 @@ class ClaudeAgentSession implements AgentSession {

  private async handleMissingResumedConversation(
    message: SDKMessage,
-    activeQuery: Query,
+    runtime: ClaudeRuntime,
  ): Promise<boolean> {
    const staleResumeError = this.readMissingResumedConversationError(message);
    if (!staleResumeError) {
@@ -4102,29 +4093,8 @@ class ClaudeAgentSession implements AgentSession {
    );

    this.failActiveTurns(staleResumeError);
-    // Ending the input retires the process on purpose. Detach first so its exit
-    // is not reported as a crash.
-    const retiredChild = this.childProcess;
-    this.childProcess = null;
-    this.input?.end();
-    await this.awaitWithTimeout(
-      activeQuery.return?.(),
-      "query pump return on missing resumed conversation",
-    );
-    // Tree-kill for the same reason the restart path does: MCP children of the
-    // retired claude process outlive it otherwise.
-    if (retiredChild) {
-      await terminateWithTreeKill(retiredChild, {
-        gracefulTimeoutMs: 2_000,
-        forceTimeoutMs: 2_000,
-      }).catch(() => {
-        /* process may already be dead */
-      });
-    }
-    if (this.query === activeQuery) {
-      this.query = null;
-      this.input = null;
-    }
+    // Clear the conversation before yielding to teardown, so a later request
+    // cannot resume a session that has already been rejected.
    this.persistence = null;
    this.persistedHistory = [];
    this.persistedProviderSubagentEvents = [];
@@ -4133,9 +4103,10 @@ class ClaudeAgentSession implements AgentSession {
    this.queryRestartNeeded = false;
    this.autonomousTurn = null;
    this.activeForegroundTurnId = null;
-    this.activeForegroundQuery = null;
-    this.activeForegroundInput = null;
+    this.activeTurnRuntime = null;
    this.syncTurnState("missing resumed conversation");
+    await this.disposeRuntime(runtime);
+    if (this.runtime === runtime) this.runtime = null;
    return true;
  }

@@ -4146,7 +4117,7 @@ class ClaudeAgentSession implements AgentSession {
   * perTaskStopAffordance), and no result would follow to consume pendingInterruptAbort.
   */
  private async interruptActiveTurn(claudeStartedTurn = this.mainTurnInFlight): Promise<void> {
-    const queryToInterrupt = this.query;
+    const queryToInterrupt = this.runtime?.query;
    if (!queryToInterrupt || typeof queryToInterrupt.interrupt !== "function") {
      this.logger.trace(
        {
@@ -60,7 +60,7 @@ describe("daemon E2E (real claude) - thinking effort memory", () => {
    }
  });

-  test("changing thinking effort preserves the previous conversation", async () => {
+  test("concurrent command discovery and thinking changes preserve replies and conversation", async () => {
    const logger = pino({ level: "silent" });
    const cwd = tmpCwd();
    const daemon = await createTestPaseoDaemon({
@@ -88,10 +88,13 @@ describe("daemon E2E (real claude) - thinking effort memory", () => {
        model: model.id,
      });

-      await client.sendMessage(
-        agent.id,
-        "Remember the code phrase PASEO_MEMORY_56. Reply exactly: ACK_56",
-      );
+      await Promise.all([
+        client.listCommands(agent.id),
+        client.sendMessage(
+          agent.id,
+          "Remember the code phrase PASEO_MEMORY_56. Reply exactly: ACK_56",
+        ),
+      ]);
      const firstFinish = await client.waitForFinish(agent.id, 180_000);
      expect(firstFinish.status).toBe("idle");
      expect(firstFinish.final?.lastError).toBeUndefined();
@@ -99,16 +102,25 @@ describe("daemon E2E (real claude) - thinking effort memory", () => {

      await client.setAgentThinkingOption(agent.id, "low");

-      await client.sendMessage(
-        agent.id,
-        "What code phrase did I ask you to remember? Reply exactly with that code phrase and nothing else.",
-      );
+      await Promise.all([
+        client.listCommands(agent.id),
+        client.sendMessage(
+          agent.id,
+          "What code phrase did I ask you to remember? Reply exactly with that code phrase and nothing else.",
+        ),
+      ]);
      const secondFinish = await client.waitForFinish(agent.id, 180_000);
      expect(secondFinish.status).toBe("idle");
      expect(secondFinish.final?.lastError).toBeUndefined();

      const assistantText = await getAssistantText(client, agent.id);
      expect(compactText(assistantText)).toContain("paseo_memory_56");
+
+      await client.sendMessage(agent.id, "Reply exactly: FOLLOWUP_56");
+      const followup = await client.waitForFinish(agent.id, 180_000);
+      expect(followup.status).toBe("idle");
+      expect(followup.final?.lastError).toBeUndefined();
+      expect(compactText(await getAssistantText(client, agent.id))).toContain("followup_56");
    } finally {
      await client.close().catch(() => undefined);
      await daemon.close().catch(() => undefined);