fix(server): keep later mode groups queued until turn end

Dispatching every mode group at once prepared later modes against the preceding turn. Keep later groups pending and reject busy sends before changing mode, leaving mode validation outside run startup.

docs/agent-lifecycle.md

@@ -150,12 +150,13 @@ Users can also detach an existing subagent from the subagents track. Detach is d
 
 `notifyOnFinish` defaults to `true` for agent-scoped creation and prompt follow-ups because most delegated work needs to report back to the creating agent. Set it to `false` for explicit reporting or no completion callback.
 
-Agent-originated messages and automatic notifications share a recipient-daemon delivery module. A queued receipt acknowledges admission before provider dispatch, allowing a child to report and finish in the same two-second quiet window. Human prompts dispatch directly. Conflicting explicit interruption or session-mode choices retain separate ordered dispatches.
+Agent-originated messages and automatic notifications share a recipient-daemon delivery module. A queued receipt acknowledges admission before provider dispatch. Idle recipients retain the two-second quiet window. While a recipient has a running or starting turn (including compaction), messages stay queued across quiet windows and are delivered together after that turn settles, including failure or cancellation. A ready batch is released when the daemon observes the recipient settle; the two-second timer also retries it. Agent callers using `steer`, including the default, wait for the turn to end; only explicit `interrupt` replaces a running turn. An explicit interrupt runs before earlier queued messages. Those messages stay queued and are delivered after the replacement turn ends. Human prompts dispatch directly with their existing behavior. Conflicting session-mode choices retain separate ordered dispatches.
 
 The pending message window is in memory. Clean shutdown flushes it before closing agents; an abrupt
 crash can lose queued prompts. A durable send receipt remains pending until actual dispatch
 acknowledgment, so a lost window never becomes a completed receipt. Completion waits cover queued
-admission within their original timeout and cancellation scope.
+admission within their original timeout and cancellation scope, and return permission checkpoints
+while messages are waiting.
 
 Finish notification intent has a separate daemon-owned lifetime. The finish-subscriptions module
 persists caller, child, host, and ownership policy through an injectable store; its JSON adapter uses

packages/cli/src/commands/agent/send.ts

@@ -47,7 +47,7 @@ export function addSendOptions(cmd: Command): Command {
     .addOption(
       new Option(
         "--behavior <behavior>",
-        "Active-turn behavior (agent default: steer; otherwise: interrupt)",
+        "Active-turn behavior (agent default: steer waits for turn end; otherwise: interrupt)",
       ).choices(["interrupt", "steer"]),
     )
     .option("--no-wait", "Return immediately without waiting for completion");
@@ -218,7 +218,7 @@ export async function runSendCommand(
           status: receipt.deliveryStatus ?? "sent",
           message:
             receipt.deliveryStatus === "queued"
-              ? "Message queued for delivery after two quiet seconds"
+              ? "Message queued for delivery after two quiet seconds and the current turn ends, unless interrupt was requested"
               : "Message sent, not waiting for completion",
         },
         schema: agentSendSchema,

packages/client/src/daemon-client.ts

@@ -482,7 +482,7 @@ export interface SendMessageOptions {
   sourceAgentServerId?: string;
   sessionMode?: string;
   messageId?: string;
-  /** What happens when the agent is mid-turn. The daemon interrupts the turn when omitted. */
+  /** Agent sources wait for turn end unless interrupt is explicit. Human sources default to interrupt. */
   activeTurnBehavior?: ActiveTurnBehavior;
   images?: Array<{ data: string; mimeType: string }>;
   attachments?: SendAgentMessageRequest["attachments"];

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

@@ -26,7 +26,8 @@ import { buildStoredAgentPayload, toAgentPayload } from "./agent-projections.js"
 import type { ProviderOptions } from "@getpaseo/protocol/agent-types";
 import { projectTimelineRows } from "./timeline-projection.js";
 import { getOpenAgentTabLabel, PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels";
-import { startAgentRun } from "./agent-prompt.js";
+import { createAgentMessageDelivery } from "./agent-messages/index.js";
+import { AgentTurnBusyError, sendPreparedPromptToAgent, startAgentRun } from "./agent-prompt.js";
 import { StaleProviderSessionError } from "./stale-provider-session-error.js";
 import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js";
 import type { StoredAgentRecord } from "./agent-storage.js";
@@ -14457,3 +14458,243 @@ test("idle cleanup gives a newly restored runtime a fresh idle window while keep
   expect((await h.sweep({ now: h.later(61) })).collected).toHaveLength(1);
   expect(h.closes).toBe(closesBeforeRestore + 1);
 });
+
+test.each([
+  ["supported", "turn_completed"],
+  ["supported", "turn_canceled"],
+  ["supported", "turn_failed"],
+  ["unsupported", "turn_completed"],
+  ["unsupported", "turn_canceled"],
+  ["unsupported", "turn_failed"],
+] as const)(
+  "queued delivery after %s provider %s never steers or interrupts",
+  async (support, ending) => {
+    const workdir = mkdtempSync(join(tmpdir(), "agent-queued-terminal-"));
+    const session =
+      support === "supported"
+        ? new SteeringTestSession({ provider: "codex", cwd: workdir })
+        : new UnsupportedSteeringSession({ provider: "codex", cwd: workdir });
+    const client = new (class extends TestAgentClient {
+      override async createSession(): Promise<AgentSession> {
+        return session;
+      }
+    })();
+    const manager = new AgentManager({ clients: { codex: client }, logger });
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    let tick = () => {};
+    const delivery = createAgentMessageDelivery({
+      isRunning: (id) => manager.hasInFlightRun(id),
+      subscribe: (notify) =>
+        manager.subscribe(
+          (event) => {
+            if (event.type === "agent_state") notify(event.agent.id);
+          },
+          { replayState: false },
+        ),
+      schedule: (_delay, callback) => {
+        tick = callback;
+        return () => {};
+      },
+      dispatch: async (request) =>
+        (
+          await startAgentRun(manager, agent.id, request.prompt, logger, {
+            idleOnly: request.idleOnly,
+            replaceRunning: true,
+            activeTurnBehavior: request.activeTurnBehavior,
+          })
+        ).disposition,
+      onError: (error) => {
+        throw error;
+      },
+    });
+    const providerStarted = new Promise<void>((resolve) => {
+      const unsubscribe = session.subscribe((event) => {
+        if (event.type === "turn_started") {
+          unsubscribe();
+          resolve();
+        }
+      });
+    });
+    try {
+      await startAgentRun(manager, agent.id, "original", logger);
+      await manager.waitForAgentRunStart(agent.id);
+      const receipt = delivery.enqueue({
+        recipientId: agent.id,
+        prompt: "queued",
+        source: { kind: "agent-message", agentId: "child" },
+      });
+      tick();
+      expect(session.startCount).toBe(1);
+      expect(session.interruptCount).toBe(0);
+      const turnId = support === "supported" ? "active-turn-1" : "unsupported-turn-1";
+      await providerStarted;
+      session.pushEvent({
+        type: ending,
+        provider: "codex",
+        turnId,
+        ...(ending === "turn_failed" ? { error: "test failure" } : {}),
+      } as AgentStreamEvent);
+      await receipt.dispatched;
+      await manager.waitForAgentRunStart(agent.id);
+      expect(session.startCount).toBe(2);
+      expect(session.interruptCount).toBe(0);
+    } finally {
+      await delivery.close();
+      await manager.closeAgent(agent.id);
+      rmSync(workdir, { recursive: true, force: true });
+    }
+  },
+);
+
+test.each(["turn_completed", "turn_failed", "turn_canceled"] as const)(
+  "an autonomous %s publishes readiness after settling its run",
+  async (ending) => {
+    const workdir = mkdtempSync(join(tmpdir(), "agent-autonomous-delivery-"));
+    const session = new SteeringTestSession({ provider: "codex", cwd: workdir });
+    const client = new (class extends TestAgentClient {
+      override async createSession(): Promise<AgentSession> {
+        return session;
+      }
+    })();
+    const manager = new AgentManager({ clients: { codex: client }, logger });
+    const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+      workspaceId: undefined,
+    });
+    let tick = () => {};
+    const settledStates: boolean[] = [];
+    const delivery = createAgentMessageDelivery({
+      isRunning: (id) => manager.hasInFlightRun(id),
+      subscribe: (notify) =>
+        manager.subscribe(
+          (event) => {
+            if (event.type !== "agent_state") return;
+            if (event.agent.lifecycle !== "running")
+              settledStates.push(manager.hasInFlightRun(agent.id));
+            notify(event.agent.id);
+          },
+          { agentId: agent.id, replayState: false },
+        ),
+      schedule: (_delay, callback) => {
+        tick = callback;
+        return () => {};
+      },
+      dispatch: async (request) =>
+        (
+          await startAgentRun(manager, agent.id, request.prompt, logger, {
+            idleOnly: request.idleOnly,
+          })
+        ).disposition,
+      onError: (error) => {
+        throw error;
+      },
+    });
+    try {
+      session.pushEvent({ type: "turn_started", provider: "codex", turnId: "autonomous" });
+      await vi.waitFor(() => expect(manager.hasInFlightRun(agent.id)).toBe(true));
+      delivery.enqueue({
+        recipientId: agent.id,
+        prompt: "queued report",
+        source: { kind: "agent-message", agentId: "child" },
+      });
+      tick();
+      expect(session.startCount).toBe(0);
+      session.pushEvent({
+        type: ending,
+        provider: "codex",
+        turnId: "autonomous",
+        ...(ending === "turn_failed" ? { error: "test failure" } : {}),
+      } as AgentStreamEvent);
+      await new Promise<void>((resolve) => setImmediate(resolve));
+      expect(settledStates).toEqual([false]);
+      expect(session.startCount).toBe(1);
+    } finally {
+      await delivery.close();
+      await manager.closeAgent(agent.id);
+      rmSync(workdir, { recursive: true, force: true });
+    }
+  },
+);
+
+test("an idle-only prompt rejects a busy recipient before changing its mode", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-queued-mode-"));
+  class ModeSession extends SteeringTestSession {
+    mode = "plan";
+    applied: string[] = [];
+    startedModes: string[] = [];
+    override async setMode(mode?: string) {
+      if (!mode) throw new Error("Mode required");
+      this.mode = mode;
+      this.applied.push(mode);
+    }
+    override async getCurrentMode() {
+      return this.mode;
+    }
+    override async startTurn(prompt: AgentPromptInput) {
+      this.startedModes.push(this.mode);
+      return super.startTurn(prompt);
+    }
+  }
+  const session = new ModeSession({ provider: "codex", cwd: workdir });
+  const client = new (class extends TestAgentClient {
+    override async createSession(): Promise<AgentSession> {
+      return session;
+    }
+  })();
+  const storage = new AgentStorage(join(workdir, "agents"), logger);
+  const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
+  const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
+    workspaceId: undefined,
+  });
+  await manager.setAgentMode(agent.id, "plan");
+  session.applied.length = 0;
+  const providerStarted = new Promise<void>((resolve) => {
+    const unsubscribe = session.subscribe((event) => {
+      if (event.type === "turn_started") {
+        unsubscribe();
+        resolve();
+      }
+    });
+  });
+  const startOther = async () => {
+    await startAgentRun(manager, agent.id, "other turn", logger);
+    await manager.waitForAgentRunStart(agent.id);
+  };
+  try {
+    await startOther();
+    await expect(
+      sendPreparedPromptToAgent({
+        agentManager: manager,
+        agentStorage: storage,
+        agentId: agent.id,
+        prompt: "queued follow-up",
+        sessionMode: "full-access",
+        idleOnly: true,
+        logger,
+      }),
+    ).rejects.toBeInstanceOf(AgentTurnBusyError);
+    expect(session.mode).toBe("plan");
+    expect(session.applied).toEqual([]);
+    expect(session.startedModes).toEqual(["plan"]);
+    await providerStarted;
+    session.pushEvent({ type: "turn_completed", provider: "codex", turnId: "active-turn-1" });
+    await vi.waitFor(() => expect(manager.hasInFlightRun(agent.id)).toBe(false));
+    await sendPreparedPromptToAgent({
+      agentManager: manager,
+      agentStorage: storage,
+      agentId: agent.id,
+      prompt: "queued follow-up",
+      sessionMode: "full-access",
+      idleOnly: true,
+      logger,
+    });
+    await manager.waitForAgentRunStart(agent.id);
+    expect(session.mode).toBe("full-access");
+    expect(session.applied).toEqual(["full-access"]);
+    expect(session.startedModes).toEqual(["plan", "full-access"]);
+  } finally {
+    await manager.closeAgent(agent.id);
+    rmSync(workdir, { recursive: true, force: true });
+  }
+});

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

@@ -1117,6 +1117,10 @@ export class AgentManager {
     return new Date(nextMs).toISOString();
   }
 
+  hasPendingPermissions(agentId: string): boolean {
+    return (this.agents.get(agentId)?.pendingPermissions.size ?? 0) > 0;
+  }
+
   hasInFlightRun(agentId: string): boolean {
     const agent = this.agents.get(agentId);
     if (!agent) {
@@ -5106,6 +5110,9 @@ export class AgentManager {
         this.runs.settleTerminalRun(agent.id, eventTurnId);
         if (isForegroundEvent) {
           this.finalizeForegroundTurn(agent, eventTurnId);
+        } else if (terminalDisposition !== "stale" && !agent.activeForegroundTurnId) {
+          // Publish autonomous readiness only after the tracked run has settled.
+          this.emitState(agent);
         }
       }
 
@@ -5375,7 +5382,6 @@ export class AgentManager {
       !agent.pendingReplacement
     ) {
       (agent as ActiveManagedAgent).lifecycle = "idle";
-      this.emitState(agent);
     }
     void this.refreshRuntimeInfo(agent);
   }
@@ -5416,9 +5422,6 @@ export class AgentManager {
       options,
     );
     this.resolvePendingPermissionsForAgent(agent, event.provider, options, "Turn failed");
-    if (!isForegroundEvent && !agent.activeForegroundTurnId) {
-      this.emitState(agent);
-    }
   }
 
   private onStreamTurnCanceled(params: {
@@ -5452,9 +5455,6 @@ export class AgentManager {
     }
     agent.lastError = undefined;
     this.resolvePendingPermissionsForAgent(agent, event.provider, options, "Interrupted");
-    if (!isForegroundEvent && !agent.activeForegroundTurnId) {
-      this.emitState(agent);
-    }
   }
 
   private onStreamTurnStarted(params: {

packages/server/src/server/agent/agent-messages/delivery.test.ts

@@ -1,3 +1,4 @@
+import { AgentTurnBusyError } from "../agent-prompt.js";
 import { expect, test } from "vitest";
 import { createAgentMessageDelivery, parseAgentMessages } from "./index.js";
 import type { AgentPromptInput } from "../agent-sdk-types.js";
@@ -117,7 +118,6 @@ test("explicit interrupt and conflicting session modes retain ordered dispatch b
   const delivery = createAgentMessageDelivery({
     dispatch: async (request) => {
       sent.push(request);
-      request.onDispatch("steered");
       return "steered";
     },
     onError: (error) => {
@@ -313,3 +313,346 @@ test("runtime safety sees queued messages until provider dispatch settles", asyn
   await delivery.close();
   expect(delivery.hasPending("parent")).toBe(false);
 });
+
+test("messages wait across quiet windows and batch after the recipient turn ends", async () => {
+  const clock = createClock();
+  let running = true;
+  const sent: Array<readonly string[]> = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => running,
+    dispatch: async (request) => {
+      sent.push(request.messageIds);
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  const source = { kind: "agent-message" as const, agentId: "child" };
+  const report = delivery.enqueue({
+    recipientId: "parent",
+    messageId: "report",
+    source,
+    prompt: "report",
+  });
+  clock.advance(2000);
+  await Promise.resolve();
+  expect(sent).toEqual([]);
+  clock.advance(13000);
+  const finish = delivery.enqueue({
+    recipientId: "parent",
+    messageId: "finish",
+    source: { kind: "agent-notification", agentId: "child", event: "finished" },
+    prompt: "finished",
+    activeTurnBehavior: "steer",
+  });
+  clock.advance(2000);
+  await Promise.resolve();
+  expect(sent).toEqual([]);
+  running = false;
+  clock.advance(2000);
+  await Promise.all([report.dispatched, finish.dispatched]);
+  expect(sent).toEqual([["report", "finish"]]);
+  await delivery.close();
+});
+
+test("a turn starting during dispatch preparation keeps receipts queued and joins later arrivals", async () => {
+  const clock = createClock();
+  let busy = true;
+  const sent: Array<readonly string[]> = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    dispatch: async (request) => {
+      if (busy) throw new AgentTurnBusyError("concurrent turn");
+      sent.push(request.messageIds);
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  const message = {
+    recipientId: "parent",
+    source: { kind: "agent-message" as const, agentId: "child" },
+    prompt: "report",
+  };
+  const first = delivery.enqueue({ ...message, messageId: "first" });
+  clock.advance(2000);
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  expect(delivery.hasPending("parent")).toBe(true);
+  const second = delivery.enqueue({ ...message, messageId: "second" });
+  busy = false;
+  clock.advance(2000);
+  await Promise.all([first.dispatched, second.dispatched]);
+  expect(sent).toEqual([["first", "second"]]);
+  await delivery.close();
+});
+
+test("explicit interrupt bypasses a running turn while ordinary messages keep waiting", async () => {
+  const clock = createClock();
+  const sent: Array<readonly string[]> = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => true,
+    dispatch: async (request) => {
+      sent.push(request.messageIds);
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  const message = {
+    recipientId: "parent",
+    source: { kind: "agent-message" as const, agentId: "child" },
+    prompt: "report",
+  };
+  delivery.enqueue({ ...message, messageId: "wait" });
+  const interrupt = delivery.enqueue({
+    ...message,
+    messageId: "interrupt",
+    activeTurnBehavior: "interrupt",
+  });
+  clock.advance(2000);
+  await interrupt.dispatched;
+  expect(sent).toEqual([["interrupt"]]);
+  expect(delivery.hasPending("parent")).toBe(true);
+  await delivery.close();
+  expect(sent).toEqual([["interrupt"], ["wait"]]);
+});
+
+test("shutdown flushes a dispatch that discovered a concurrent turn during preparation", async () => {
+  const clock = createClock();
+  let release!: () => void;
+  const preparing = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  const modes: boolean[] = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    dispatch: async (request) => {
+      modes.push(request.idleOnly);
+      if (request.idleOnly) {
+        await preparing;
+        throw new AgentTurnBusyError("busy");
+      }
+      return "turn_started";
+    },
+    onError: () => {},
+  });
+  const receipt = delivery.enqueue({
+    recipientId: "parent",
+    prompt: "report",
+    source: { kind: "agent-message", agentId: "child" },
+  });
+  clock.advance(2000);
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  const closing = delivery.close();
+  release();
+  await closing;
+  await expect(receipt.dispatched).resolves.toBe("turn_started");
+  expect(modes).toEqual([true, false]);
+});
+
+test("conflicting modes dispatch on separate turn ends without changing the active turn mode", async () => {
+  const clock = createClock();
+  let busy = false;
+  let mode = "initial";
+  let wake!: (recipientId: string) => void;
+  const sent: Array<{ ids: readonly string[]; mode?: string }> = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => busy,
+    subscribe: (listener) => {
+      wake = listener;
+      return () => {};
+    },
+    dispatch: async (request) => {
+      // Mode preparation precedes run admission in the real dispatcher.
+      mode = request.sessionMode ?? mode;
+      if (busy) throw new AgentTurnBusyError("concurrent turn");
+      sent.push({ ids: request.messageIds, mode });
+      busy = true;
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  const message = {
+    recipientId: "parent",
+    source: { kind: "agent-message" as const, agentId: "child" },
+    prompt: "message",
+  };
+  const a = delivery.enqueue({ ...message, messageId: "A", sessionMode: "plan" });
+  const b = delivery.enqueue({ ...message, messageId: "B", sessionMode: "default" });
+  clock.advance(2000);
+  await a.dispatched;
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  expect(mode).toBe("plan");
+  expect(sent).toEqual([{ ids: ["A"], mode: "plan" }]);
+  busy = false;
+  wake("parent");
+  await b.dispatched;
+  expect(mode).toBe("default");
+  expect(sent).toEqual([
+    { ids: ["A"], mode: "plan" },
+    { ids: ["B"], mode: "default" },
+  ]);
+  await delivery.close();
+});
+
+test("interrupts retain separate mode choices while the recipient is busy", async () => {
+  const clock = createClock();
+  const sent: Array<{ ids: readonly string[]; mode?: string }> = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => true,
+    dispatch: async (request) => {
+      sent.push({ ids: request.messageIds, mode: request.sessionMode });
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  const message = {
+    recipientId: "parent",
+    source: { kind: "agent-message" as const, agentId: "child" },
+    prompt: "message",
+    activeTurnBehavior: "interrupt" as const,
+  };
+  const a = delivery.enqueue({ ...message, messageId: "A", sessionMode: "plan" });
+  const b = delivery.enqueue({ ...message, messageId: "B", sessionMode: "default" });
+  clock.advance(2000);
+  await Promise.all([a.dispatched, b.dispatched]);
+  expect(sent).toEqual([
+    { ids: ["A"], mode: "plan" },
+    { ids: ["B"], mode: "default" },
+  ]);
+  await delivery.close();
+});
+
+test("a settled-turn event releases a ready batch without waiting for the next timer sample", async () => {
+  const clock = createClock();
+  let running = true;
+  let settled: (id: string) => void = () => {};
+  let subscribed = true;
+  const sent: string[] = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => running,
+    subscribe: (listener) => {
+      settled = listener;
+      return () => {
+        subscribed = false;
+      };
+    },
+    dispatch: async (request) => {
+      sent.push(request.recipientId);
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  delivery.enqueue({
+    recipientId: "parent",
+    prompt: "report",
+    source: { kind: "agent-message", agentId: "child" },
+  });
+  clock.advance(2000);
+  expect(sent).toEqual([]);
+  running = false;
+  settled("parent");
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  expect(sent).toEqual(["parent"]);
+  running = true;
+  await delivery.close();
+  expect(subscribed).toBe(false);
+});
+
+test.each([false, true])(
+  "completion waits report permission checkpoints without draining messages (initial: %s)",
+  async (initial) => {
+    const clock = createClock();
+    let permission = initial;
+    let changed: (id: string) => void = () => {};
+    let reported = false;
+    let sent = 0;
+    const delivery = createAgentMessageDelivery({
+      schedule: clock.schedule,
+      isRunning: () => true,
+      hasPendingPermission: () => permission,
+      subscribe: (listener) => {
+        changed = listener;
+        return () => {};
+      },
+      dispatch: async () => {
+        sent++;
+        return "turn_started";
+      },
+      onError: (error) => {
+        throw error;
+      },
+    });
+    delivery.enqueue({
+      recipientId: "child",
+      prompt: "follow up",
+      source: { kind: "agent-message", agentId: "parent" },
+    });
+    const wait = delivery.waitForDispatch("child", new AbortController().signal).then(() => {
+      reported = true;
+      return undefined;
+    });
+    permission = true;
+    changed("child");
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    expect(reported).toBe(true);
+    expect(sent).toBe(0);
+    expect(delivery.hasPending("child")).toBe(true);
+    await wait;
+    await delivery.close();
+  },
+);
+
+test("a busy admission retry stays ready for the next settled-turn event", async () => {
+  const clock = createClock();
+  let running = false;
+  let raced = true;
+  let notify: (id: string) => void = () => {};
+  const sent: string[] = [];
+  const delivery = createAgentMessageDelivery({
+    schedule: clock.schedule,
+    isRunning: () => running,
+    subscribe: (listener) => {
+      notify = listener;
+      return () => {};
+    },
+    dispatch: async (request) => {
+      if (raced) {
+        running = true;
+        throw new AgentTurnBusyError("next turn started");
+      }
+      sent.push(request.recipientId);
+      return "turn_started";
+    },
+    onError: (error) => {
+      throw error;
+    },
+  });
+  delivery.enqueue({
+    recipientId: "parent",
+    prompt: "report",
+    source: { kind: "agent-message", agentId: "child" },
+  });
+  clock.advance(2000);
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  raced = false;
+  running = false;
+  notify("parent");
+  await new Promise<void>((resolve) => setImmediate(resolve));
+  expect(sent).toEqual(["parent"]);
+  await delivery.close();
+});

packages/server/src/server/agent/agent-messages/delivery.ts

@@ -1,4 +1,4 @@
-import type { PromptDispatchDisposition } from "../agent-prompt.js";
+import { AgentTurnBusyError, type PromptDispatchDisposition } from "../agent-prompt.js";
 import { randomUUID } from "node:crypto";
 import type { AgentPromptInput, AgentPromptContentBlock } from "../agent-sdk-types.js";
 import { prepareAgentMessage, type AgentPromptSource } from "./envelope.js";
@@ -21,6 +21,7 @@ interface Receipt {
   dispatched: Promise<PromptDispatchDisposition>;
 }
 interface Admitted extends Message {
+  sequence: number;
   receipt: Receipt;
   resolve(disposition: PromptDispatchDisposition): void;
   reject(error: unknown): void;
@@ -30,12 +31,17 @@ interface DispatchRequest {
   messageIds: readonly string[];
   prompt: AgentPromptInput;
   activeTurnBehavior: "steer" | "interrupt";
+  idleOnly: boolean;
   sessionMode?: string;
   clearPendingPermissions: boolean;
   unarchive: boolean;
-  onDispatch(disposition: PromptDispatchDisposition): void;
 }
 interface Dependencies {
+  /** Includes admitted starts and provider turns, including compaction. */
+  isRunning?: (recipientId: string) => boolean;
+  hasPendingPermission?: (recipientId: string) => boolean;
+  /** Notify when recipient state changes; admission still checks whether it is idle. */
+  subscribe?: (listener: (recipientId: string) => void) => () => void;
   dispatch(request: DispatchRequest): Promise<PromptDispatchDisposition>;
   onError(error: unknown, request: DispatchRequest): void;
   /** Return a cancellation function. Production uses the daemon's event loop. */
@@ -47,7 +53,7 @@ export interface AgentMessageDelivery {
   hasPending(recipientId: string): boolean;
   /** Stop admission, flush pending messages, and await provider dispatch acknowledgments. */
   close(): Promise<void>;
-  /** Existing completion waits include agent messages admitted before the wait. */
+  /** Completion waits include admitted messages, but return at permission checkpoints. */
   waitForDispatch(recipientId: string, signal: AbortSignal): Promise<void>;
 }
 
@@ -59,8 +65,10 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
       const timer = setTimeout(callback, delayMs);
       return () => clearTimeout(timer);
     });
-  const pending = new Map<string, { messages: Admitted[]; cancel(): void }>();
+  const pending = new Map<string, { messages: Admitted[]; ready: boolean; cancel(): void }>();
   const dispatching = new Map<string, Promise<void>>();
+  const receipts = new Map<string, Set<Receipt>>();
+  let sequence = 0;
   let closed = false;
   let closing: Promise<void> | undefined;
 
@@ -76,21 +84,13 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
         );
     const request: DispatchRequest = {
       recipientId,
+      idleOnly: !closed && messages[0]!.activeTurnBehavior !== "interrupt",
       messageIds: messages.map((message) => message.receipt.messageId),
       prompt,
       activeTurnBehavior: messages[0]!.activeTurnBehavior ?? "steer",
       sessionMode: messages.find((message) => message.sessionMode)?.sessionMode,
       clearPendingPermissions: messages.some((message) => message.clearPendingPermissions),
       unarchive: messages.some((message) => message.unarchive !== false),
-      onDispatch: (disposition) => {
-        for (const message of messages) {
-          try {
-            message.onDispatch?.(disposition);
-          } catch (error) {
-            dependencies.onError(error, request);
-          }
-        }
-      },
     };
     const previous = dispatching.get(recipientId) ?? Promise.resolve();
     const dispatched = previous
@@ -98,10 +98,28 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
       .then(async () => {
         try {
           for (const message of messages) message.beforeDispatch?.();
-          const disposition = await dependencies.dispatch(request);
-          for (const message of messages) message.resolve(disposition);
+          let disposition: PromptDispatchDisposition;
+          try {
+            disposition = await dependencies.dispatch(request);
+          } catch (error) {
+            if (!(error instanceof AgentTurnBusyError) || !closed) throw error;
+            // Shutdown drains admitted messages before closing the recipients.
+            disposition = await dependencies.dispatch({ ...request, idleOnly: false });
+          }
+          for (const message of messages) {
+            try {
+              message.onDispatch?.(disposition);
+            } catch (error) {
+              dependencies.onError(error, request);
+            }
+            message.resolve(disposition);
+          }
           return undefined;
         } catch (error) {
+          if (error instanceof AgentTurnBusyError && !closed) {
+            requeue(recipientId, messages);
+            return;
+          }
           for (const message of messages) message.reject(error);
           dependencies.onError(error, request);
           throw error;
@@ -115,30 +133,78 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
       .catch(() => undefined);
   }
 
+  function requeue(recipientId: string, messages: Admitted[]): void {
+    const entry = pending.get(recipientId) ?? {
+      messages: [],
+      ready: true,
+      cancel: () => {},
+    };
+    entry.cancel();
+    entry.messages.push(...messages);
+    entry.messages.sort((a, b) => a.sequence - b.sequence);
+    entry.cancel = schedule(2000, () => flush(recipientId));
+    pending.set(recipientId, entry);
+  }
+
   function flush(recipientId: string): void {
     const entry = pending.get(recipientId);
     if (!entry) return;
-    pending.delete(recipientId);
+    entry.ready = true;
     entry.cancel();
+    if (!closed && dependencies.isRunning?.(recipientId)) {
+      const interrupts = entry.messages.filter(
+        (message) => message.activeTurnBehavior === "interrupt",
+      );
+      entry.messages = entry.messages.filter(
+        (message) => message.activeTurnBehavior !== "interrupt",
+      );
+      if (interrupts.length) dispatchGroups(recipientId, interrupts);
+      if (entry.messages.length) {
+        entry.cancel = schedule(2000, () => flush(recipientId));
+      } else {
+        pending.delete(recipientId);
+      }
+      return;
+    }
+    pending.delete(recipientId);
+    dispatchGroups(recipientId, entry.messages);
+  }
+
+  function dispatchGroups(recipientId: string, messages: Admitted[]): void {
+    const groups: Admitted[][] = [];
     let group: Admitted[] = [];
     let mode: string | undefined;
-    for (const message of entry.messages) {
+    for (const message of messages) {
       // Explicitly conflicting send options require separate, ordered dispatches.
       if (
         group.length &&
         ((message.activeTurnBehavior ?? "steer") !== (group[0]!.activeTurnBehavior ?? "steer") ||
           (mode && message.sessionMode && mode !== message.sessionMode))
       ) {
-        dispatch(recipientId, group);
+        groups.push(group);
         group = [];
         mode = undefined;
       }
       group.push(message);
       mode ??= message.sessionMode;
     }
-    if (group.length) dispatch(recipientId, group);
+    if (group.length) groups.push(group);
+    let started = false;
+    for (const batch of groups) {
+      if (!closed && started && batch[0]!.activeTurnBehavior !== "interrupt") {
+        requeue(recipientId, batch);
+      } else {
+        dispatch(recipientId, batch);
+        started = true;
+      }
+    }
   }
 
+  const unsubscribe = dependencies.subscribe?.((recipientId) => {
+    if (pending.get(recipientId)?.ready && !dependencies.isRunning?.(recipientId))
+      flush(recipientId);
+  });
+
   return {
     hasPending: (recipientId) => pending.has(recipientId) || dispatching.has(recipientId),
     enqueue(message) {
@@ -155,17 +221,40 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
         messageId: message.messageId ?? randomUUID(),
         dispatched,
       };
-      const entry = pending.get(message.recipientId) ?? { messages: [], cancel: () => {} };
+      const owned = receipts.get(message.recipientId) ?? new Set<Receipt>();
+      owned.add(receipt);
+      receipts.set(message.recipientId, owned);
+      void dispatched
+        .finally(() => {
+          owned.delete(receipt);
+          if (!owned.size) receipts.delete(message.recipientId);
+        })
+        .catch(() => undefined);
+      const entry = pending.get(message.recipientId) ?? {
+        messages: [],
+        ready: false,
+        cancel: () => {},
+      };
       entry.cancel();
-      entry.messages.push({ ...message, receipt, resolve, reject });
+      entry.ready = false;
+      entry.messages.push({ ...message, receipt, resolve, reject, sequence: sequence++ });
       entry.cancel = schedule(2000, () => flush(message.recipientId));
       pending.set(message.recipientId, entry);
       return receipt;
     },
     async waitForDispatch(recipientId, signal) {
       signal.throwIfAborted();
-      const admitted =
-        pending.get(recipientId)?.messages.map((message) => message.receipt.dispatched) ?? [];
+      const admitted = Array.from(receipts.get(recipientId) ?? [], (receipt) => receipt.dispatched);
+      let unsubscribePermission: (() => void) | undefined;
+      const permission = new Promise<void>((resolve) => {
+        const check = () => {
+          if (dependencies.hasPendingPermission?.(recipientId)) resolve();
+        };
+        unsubscribePermission = dependencies.subscribe?.((agentId) => {
+          if (agentId === recipientId) check();
+        });
+        check();
+      });
       let aborted!: () => void;
       const cancellation = new Promise<never>((_resolve, reject) => {
         aborted = () => reject(new DOMException("Wait aborted", "AbortError"));
@@ -175,14 +264,17 @@ export function createAgentMessageDelivery(dependencies: Dependencies): AgentMes
         await Promise.race([
           Promise.all([dispatching.get(recipientId), ...admitted]),
           cancellation,
+          permission,
         ]);
       } finally {
         signal.removeEventListener("abort", aborted);
+        unsubscribePermission?.();
       }
     },
     close() {
       if (closing) return closing;
       closed = true;
+      unsubscribe?.();
       for (const recipientId of pending.keys()) flush(recipientId);
       // Each failure is reported and rejects its receipts; shutdown still drains other recipients.
       closing = Promise.allSettled(dispatching.values()).then(() => undefined);

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

@@ -23,7 +23,10 @@ export type AgentRunController = Pick<
   reloadAgentSession(agentId: string): Promise<unknown>;
 };
 
+export class AgentTurnBusyError extends Error {}
+
 export interface StartAgentRunOptions {
+  idleOnly?: boolean;
   /** Called after selecting the run, before consuming its events. */
   onDispatch?: (disposition: PromptDispatchDisposition) => void;
   replaceRunning?: boolean;
@@ -136,7 +139,12 @@ async function startAgentRunInner(
   options?: StartAgentRunOptions,
 ): Promise<{ disposition: PromptDispatchDisposition }> {
   const snapshot = agentManager.getAgent(agentId);
-  const steered = await steerOrReplaceActiveRun(agentManager, agentId, prompt, options);
+  const steered = options?.idleOnly
+    ? null
+    : await steerOrReplaceActiveRun(agentManager, agentId, prompt, options);
+  if (options?.idleOnly && agentManager.hasInFlightRun(agentId)) {
+    throw new AgentTurnBusyError(`Agent ${agentId} is running`);
+  }
   if (steered?.disposition === "steered") {
     options?.onDispatch?.("steered");
     return steered;
@@ -210,6 +218,7 @@ export async function unarchiveAgentState(
 }
 
 export interface SendPromptToAgentParams {
+  idleOnly?: boolean;
   /** Awaited after the agent is loaded and its mode is set, before the prompt is handed to a run. */
   beforeDispatch?: () => Promise<void>;
   onDispatch?: StartAgentRunOptions["onDispatch"];
@@ -355,13 +364,17 @@ export async function sendPreparedPromptToAgent(
     throw error;
   }
 
+  if (params.idleOnly && params.agentManager.hasInFlightRun(params.agentId)) {
+    throw new AgentTurnBusyError(`Agent ${params.agentId} is running`);
+  }
   if (params.sessionMode) {
     await params.agentManager.setAgentMode(params.agentId, params.sessionMode);
   }
 
   await params.beforeDispatch?.();
   return await startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, {
-    replaceRunning: true,
+    replaceRunning: !params.idleOnly,
+    idleOnly: params.idleOnly,
     onDispatch: params.onDispatch,
     activeTurnBehavior: params.activeTurnBehavior,
     clearPendingPermissions: params.clearPendingPermissions,

packages/server/src/server/agent/finish-subscriptions/finish-subscriptions.test.ts

@@ -498,3 +498,26 @@ test("finish recovery protects callers before durable intent has loaded", async
   expect(recovered.hasPendingForCaller("parent")).toBe(true);
   expect(recovered.hasPendingForCaller("unrelated")).toBe(false);
 });
+
+test("an undispatched follow-up does not block the running child's permission notification", async () => {
+  const state = await scenario();
+  await state.service.register(intent);
+  await until(() => state.observation.waits.length === 1);
+  await state.service.register(intent, true);
+  const result = outcome("running");
+  result.status = "permission";
+  result.final!.pendingPermissions = [
+    {
+      id: "approval",
+      provider: "codex",
+      name: "Bash",
+      kind: "tool",
+      input: { command: "run tests" },
+    } as NonNullable<typeof result.final>["pendingPermissions"][number],
+  ];
+  state.observation.waits[0]!.result.resolve(result);
+  await until(() => state.delivery.prompts.length === 1);
+  expect(state.delivery.prompts[0]).toContain("needs permission");
+  expect(state.delivery.prompts[0]).toContain("approval");
+  expect(await state.store.list()).toHaveLength(2);
+});

packages/server/src/server/agent/finish-subscriptions/index.ts

@@ -267,7 +267,9 @@ export function createFinishSubscriptions(deps: Dependencies): FinishSubscriptio
             continue;
           }
           if (signal.aborted) return;
-          await decisions(group);
+          // A queued follow-up cannot start until the current permission is answered.
+          // Its pending registration must not hold the running turn's checkpoint.
+          if (pending.status !== "permission" || watch.pending) await decisions(group);
           if (signal.aborted) return;
           if (await suppressed(record, pending)) {
             await remove(watch);

packages/server/src/server/agent/host-operations/index.ts

@@ -298,7 +298,7 @@ export function createHostResolver(input: {
               status: agent.status,
               guidance:
                 receipt.deliveryStatus === "queued"
-                  ? "Queued for delivery after two quiet seconds. You will get notified when the prompted agent finishes, errors, or needs permission unless notifyOnFinish is false."
+                  ? "Queued for delivery after two quiet seconds and the current turn ends, unless interrupt was requested. You will get notified when the prompted agent finishes, errors, or needs permission unless notifyOnFinish is false."
                   : "Dispatch acknowledged. Observe completion using get_agent_status/get_agent_activity or SDK/RPC subscriptions and wait APIs.",
             });
           },

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

@@ -172,11 +172,12 @@ function registeredTool(
 async function invokeToolWithParsedInput(
   tool: RegisteredMcpToolWithHandler,
   input: Record<string, unknown>,
+  awaitDispatch = true,
 ) {
   const parsed = await tool.inputSchema.safeParseAsync(input);
   expect(parsed.success).toBe(true);
   const response = await tool.handler(parsed.data);
-  if (response.structuredContent.deliveryStatus === "queued") {
+  if (awaitDispatch && response.structuredContent.deliveryStatus === "queued") {
     await toolDeliveries
       .get(tool)!
       .waitForDispatch(input.agentId as string, new AbortController().signal);
@@ -274,6 +275,7 @@ function buildAgentManagerSpies() {
     effectiveOptionsFor: (_provider: string, own?: Record<string, unknown>) => own ?? {},
     reloadStalePluginSession: async () => {},
     hasInFlightRun: vi.fn().mockReturnValue(false),
+    hasPendingPermissions: () => false,
     steerOrReplaceActiveTurn: async () => ({ status: "inactive" as const }),
     tryRunOutOfBand: vi.fn().mockReturnValue(false),
     subscribe: vi.fn().mockReturnValue(() => {}),
@@ -623,6 +625,12 @@ class FakeBrowserToolsBroker {
 }
 
 class BoundaryAgentManagerFake {
+  subscribe() {
+    return () => {};
+  }
+  hasPendingPermissions() {
+    return false;
+  }
   public getPluginTools(): [] {
     return [];
   }
@@ -3677,15 +3685,15 @@ class HeldTurnAgentClient implements AgentClient {
 
 describe("send_agent_prompt MCP tool", () => {
   it.each([
-    { agentScoped: true, behavior: undefined, available: true, steered: true },
-    { agentScoped: true, behavior: "interrupt", available: true, steered: false },
-    { agentScoped: true, behavior: "steer", available: true, steered: true },
-    { agentScoped: false, behavior: undefined, available: true, steered: false },
-    { agentScoped: false, behavior: "steer", available: true, steered: true },
-    { agentScoped: true, behavior: undefined, available: false, steered: false },
+    { agentScoped: true, behavior: undefined, available: true },
+    { agentScoped: true, behavior: "interrupt", available: true },
+    { agentScoped: true, behavior: "steer", available: true },
+    { agentScoped: false, behavior: undefined, available: true },
+    { agentScoped: false, behavior: "steer", available: true },
+    { agentScoped: true, behavior: undefined, available: false },
   ] as const)(
     "dispatches behavior $behavior for agent-scoped=$agentScoped, steering available=$available",
-    async ({ agentScoped, behavior, available, steered }) => {
+    async ({ agentScoped, behavior, available }) => {
       const workdir = await mkdtemp(join(tmpdir(), "mcp-send-behavior-"));
       const storage = new AgentStorage(join(workdir, "agents"), logger);
       const client = new HeldTurnAgentClient("codex", true, {
@@ -3717,18 +3725,33 @@ describe("send_agent_prompt MCP tool", () => {
           notifyOnFinish: false,
         });
         const originalTurn = manager.getAgent(child.id)!.activeTurnId;
-        const response = await invokeToolWithParsedInput(tool, {
-          agentId: child.id,
-          prompt: "New message",
-          notifyOnFinish: false,
-          ...(behavior ? { activeTurnBehavior: behavior } : {}),
-        });
+        const queued = agentScoped && behavior !== "interrupt";
+        const steered = !agentScoped && behavior === "steer";
+        const response = await invokeToolWithParsedInput(
+          tool,
+          {
+            agentId: child.id,
+            prompt: "New message",
+            notifyOnFinish: false,
+            ...(behavior ? { activeTurnBehavior: behavior } : {}),
+          },
+          !queued,
+        );
         expect(response.structuredContent).toMatchObject({
           success: true,
           ...(agentScoped
             ? { deliveryStatus: "queued" }
             : { disposition: steered ? "steered" : "turn_started" }),
         });
+        if (queued) {
+          expect(client.sessions[1]!.prompts).toHaveLength(1);
+          expect(client.sessions[1]!.steeredPrompts).toEqual([]);
+          expect(manager.getAgent(child.id)!.activeTurnId).toBe(originalTurn);
+          client.sessions[1]!.finishTurn();
+          await serverDeliveries
+            .get(server)!
+            .waitForDispatch(child.id, new AbortController().signal);
+        }
         expect(client.sessions[1]!.prompts).toHaveLength(steered ? 1 : 2);
         expect(client.sessions[1]!.steeredPrompts).toHaveLength(steered ? 1 : 0);
         expect(manager.getAgent(child.id)!.activeTurnId === originalTurn).toBe(steered);
@@ -3740,18 +3763,16 @@ describe("send_agent_prompt MCP tool", () => {
 
   it.each([
     { rejected: false, priorNotification: false },
-    { rejected: true, priorNotification: false },
     { rejected: false, priorNotification: true },
+    { rejected: true, priorNotification: false },
+    { rejected: true, priorNotification: true },
   ])(
-    "buffers completion until steer acknowledgement (rejected: $rejected, prior notification: $priorNotification)",
+    "keeps completion tied to queued dispatch acknowledgement (rejected: $rejected, prior notification: $priorNotification)",
     async ({ rejected, priorNotification }) => {
-      const workdir = await mkdtemp(join(tmpdir(), "mcp-send-behavior-"));
+      const workdir = await mkdtemp(join(tmpdir(), "mcp-queued-ack-"));
       const storage = new AgentStorage(join(workdir, "agents"), logger);
-      const client = new HeldTurnAgentClient("codex", true, {
-        steerResult: "accepted",
-        finishDuringSteer: true,
-        rejectSteer: rejected,
-      });
+      const options: HeldTurnOptions = {};
+      const client = new HeldTurnAgentClient("codex", true, options);
       const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
       let mcpClient: Client | undefined;
       try {
@@ -3773,104 +3794,47 @@ describe("send_agent_prompt MCP tool", () => {
           logger,
         });
         mcpClient = await connectInMemoryMcpClient(server);
-        const send = (input: Record<string, unknown>) =>
-          sendMcpAndAwaitDispatch(mcpClient!, server, input);
-        await send({
+        await sendMcpAndAwaitDispatch(mcpClient, server, {
           agentId: child.id,
           prompt: "Original work",
           notifyOnFinish: priorNotification,
         });
-        const sending = send({
-          agentId: child.id,
-          prompt: "New message",
-          notifyOnFinish: true,
+        const session = client.sessions[1]!;
+        const release = session.holdTurnStart();
+        options.finishTurnsDuringStart = true;
+        if (rejected) session.rejectNextTurnStart(new Error("Start rejected"));
+        const response = await mcpClient.callTool({
+          name: "send_agent_prompt",
+          arguments: { agentId: child.id, prompt: "New message", notifyOnFinish: true },
         });
-        if (rejected) {
-          await expect(sending).rejects.toThrow("Steer rejected");
-          // Flush asynchronous notification work before checking that a rejected
-          // dispatch did not notify. Then prove a later successful send still can.
-          await new Promise((resolve) => setTimeout(resolve, 50));
-          expect(client.sessions[0]!.prompts).toHaveLength(0);
-          await send({
-            agentId: child.id,
-            prompt: "Retry after rejection",
+        expect(response.structuredContent).toMatchObject({
+          success: true,
+          deliveryStatus: "queued",
+        });
+        const dispatch = serverDeliveries
+          .get(server)!
+          .waitForDispatch(child.id, new AbortController().signal);
+        const rejectedDispatch = rejected
+          ? expect(dispatch).rejects.toThrow("Start rejected")
+          : null;
+        expect(client.sessions[0]!.prompts).toHaveLength(0);
+        session.finishTurn();
+        await vi.waitFor(() => expect(session.prompts).toHaveLength(2));
+        expect(client.sessions[0]!.prompts).toHaveLength(0);
+        release();
+        if (rejectedDispatch) await rejectedDispatch;
+        else await dispatch;
+        if (!rejected || priorNotification) {
+          await vi.waitFor(() => expect(client.sessions[0]!.prompts).toHaveLength(1), {
+            timeout: 5000,
+          });
+          expect(parseAgentMessage(client.sessions[0]!.prompts[0])).toMatchObject({
+            source: { kind: "agent-notification", agentId: child.id, event: "finished" },
           });
-          client.sessions[1]!.finishTurn();
         } else {
-          const response = await sending;
-          expect(response.structuredContent).toMatchObject({
-            success: true,
-            deliveryStatus: "queued",
-          });
-        }
-        await vi.waitFor(() => expect(client.sessions[0]!.prompts).toHaveLength(1));
-        expect(parseAgentMessage(client.sessions[0]!.prompts[0])).toMatchObject({
-          source: { kind: "agent-notification", agentId: child.id, event: "finished" },
-        });
-      } finally {
-        await mcpClient?.close();
-        await removeAgentStateDir(manager, storage, workdir);
-      }
-    },
-  );
-
-  it.each([false, true])(
-    "preserves the original notification after rejection (completion during send: %s)",
-    async (finishDuringSteer) => {
-      const workdir = await mkdtemp(join(tmpdir(), "mcp-send-behavior-"));
-      const storage = new AgentStorage(join(workdir, "agents"), logger);
-      const client = new HeldTurnAgentClient("codex", true, {
-        steerResult: "accepted",
-        finishDuringSteer,
-        rejectSteer: true,
-      });
-      const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
-      let mcpClient: Client | undefined;
-      try {
-        const parent = await manager.createAgent(
-          { provider: "codex", cwd: existingCwd },
-          undefined,
-          {
-            workspaceId: "wks_parent",
-          },
-        );
-        const child = await manager.createAgent(
-          { provider: "codex", cwd: existingCwd },
-          undefined,
-          {
-            workspaceId: "wks_parent",
-          },
-        );
-        const server = await createAgentMcpServer({
-          agentManager: manager,
-          agentStorage: storage,
-          callerAgentId: parent.id,
-          providerSnapshotManager: createOpenCodeManager().manager,
-          logger,
-        });
-        mcpClient = await connectInMemoryMcpClient(server);
-        const send = (input: Record<string, unknown>) =>
-          sendMcpAndAwaitDispatch(mcpClient!, server, input);
-        await send({
-          agentId: child.id,
-          prompt: "Original work",
-          notifyOnFinish: true,
-        });
-        const sending = send({
-          agentId: child.id,
-          prompt: "New message",
-          notifyOnFinish: true,
-        });
-        await expect(sending).rejects.toThrow("Steer rejected");
-        if (!finishDuringSteer) {
-          await new Promise((resolve) => setTimeout(resolve, 50));
+          await serverSubscriptions.get(server)!.close();
           expect(client.sessions[0]!.prompts).toHaveLength(0);
-          client.sessions[1]!.finishTurn();
         }
-        await vi.waitFor(() => expect(client.sessions[0]!.prompts).toHaveLength(1));
-        expect(parseAgentMessage(client.sessions[0]!.prompts[0])).toMatchObject({
-          source: { kind: "agent-notification", agentId: child.id, event: "finished" },
-        });
       } finally {
         await mcpClient?.close();
         await removeAgentStateDir(manager, storage, workdir);
@@ -4012,7 +3976,7 @@ describe("send_agent_prompt MCP tool", () => {
     const response = await tool.handler(parsed.data as Record<string, unknown>);
 
     expect(response.structuredContent).toMatchObject({ deliveryStatus: "queued" });
-    expect(spies.agentManager.subscribe).not.toHaveBeenCalled();
+    expect(serverSubscriptionRecords.get(server)!.size).toBe(1);
     await serverDeliveries
       .get(server)!
       .waitForDispatch("child-agent", new AbortController().signal);
@@ -4053,7 +4017,7 @@ describe("send_agent_prompt MCP tool", () => {
 
     await tool.handler(parsed.data as Record<string, unknown>);
 
-    expect(spies.agentManager.subscribe).not.toHaveBeenCalled();
+    expect(serverSubscriptionRecords.get(server)!.size).toBe(0);
     expect(spies.agentManager.waitForAgentEvent).not.toHaveBeenCalled();
   });
 
@@ -4096,7 +4060,7 @@ describe("send_agent_prompt MCP tool", () => {
       notifyOnFinish: false,
     });
 
-    expect(spies.agentManager.subscribe).not.toHaveBeenCalled();
+    expect(serverSubscriptionRecords.get(server)!.size).toBe(0);
     expect(spies.agentManager.waitForAgentEvent).not.toHaveBeenCalled();
   });
 
@@ -4220,11 +4184,15 @@ describe("send_agent_prompt MCP tool", () => {
           notifyOnFinish: false,
         });
         if (scenario !== "startup_failure") {
-          const response = await invokeToolWithParsedInput(tool, {
-            agentId: child.id,
-            prompt: "/side-command",
-            activeTurnBehavior: scenario === "out_of_band_interrupt" ? "interrupt" : "steer",
-          });
+          const response = await invokeToolWithParsedInput(
+            tool,
+            {
+              agentId: child.id,
+              prompt: "/side-command",
+              activeTurnBehavior: scenario === "out_of_band_interrupt" ? "interrupt" : "steer",
+            },
+            scenario !== "out_of_band",
+          );
           expect(response.structuredContent).toMatchObject({
             success: true,
             deliveryStatus: "queued",
@@ -4236,6 +4204,9 @@ describe("send_agent_prompt MCP tool", () => {
           });
           expect(parentClient.sessions[0]!.prompts).toHaveLength(0);
           childClient.sessions[0]!.finishTurn("Ongoing task done");
+          await serverDeliveries
+            .get(server)!
+            .waitForDispatch(child.id, new AbortController().signal);
           await vi.waitFor(() => {
             expect(parentClient.sessions[0]!.prompts).toHaveLength(1);
             expect(parentClient.sessions[0]!.prompts[0]).toContain("finished.");
@@ -4243,7 +4214,11 @@ describe("send_agent_prompt MCP tool", () => {
         } else {
           childClient.sessions[0]!.rejectNextTurnStart(new Error("Requested turn could not start"));
           await expect(
-            invokeToolWithParsedInput(tool, { agentId: child.id, prompt: "Replacement task" }),
+            invokeToolWithParsedInput(tool, {
+              agentId: child.id,
+              prompt: "Replacement task",
+              activeTurnBehavior: "interrupt",
+            }),
           ).rejects.toThrow("Requested turn could not start");
           // A later real turn must not inherit a subscription from the rejected dispatch.
           await invokeToolWithParsedInput(tool, {
@@ -4354,6 +4329,7 @@ describe("send_agent_prompt MCP tool", () => {
       const sent = await invokeToolWithParsedInput(registeredTool(server, "send_agent_prompt"), {
         agentId: childId,
         prompt: "Stop and reply instead",
+        activeTurnBehavior: "interrupt",
       });
       expect(sent.structuredContent).toMatchObject({
         success: true,
@@ -4581,7 +4557,7 @@ describe("send_agent_prompt MCP tool to an OMP child", () => {
     { command: "follow-up", streamingBehavior: "followUp" },
     { command: "steer", streamingBehavior: "steer" },
   ] as const)(
-    "delivers /$command into a turn OMP started on its own, without aborting it",
+    "queues /$command until the turn OMP started on its own ends",
     async ({ command, streamingBehavior }) => {
       const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } =
         await setupOmpChild();
@@ -4592,10 +4568,22 @@ describe("send_agent_prompt MCP tool to an OMP child", () => {
           expect(agentManager.getAgent(childId)?.activeTurnId).toMatch(/^autonomous-/);
         });
 
-        const response = await invokeToolWithParsedInput(tool, {
-          agentId: childId,
-          prompt: `/${command} then summarize the logs`,
-        });
+        const commandStarted = ompSession.nextPrompt();
+        const response = await invokeToolWithParsedInput(
+          tool,
+          {
+            agentId: childId,
+            prompt: `/${command} then summarize the logs`,
+          },
+          false,
+        );
+        expect(ompSession.prompts).toEqual([]);
+        expect(ompSession.abortRequested).toBe(false);
+        ompSession.finishTurn();
+        await commandStarted;
+        ompSession.beginTurn();
+        ompSession.streamAssistantText("Summarizing the logs");
+        await toolDeliveries.get(tool)!.waitForDispatch(childId, new AbortController().signal);
 
         expect(response.structuredContent).toMatchObject({
           success: true,

packages/server/src/server/agent/tools/paseo-tools.ts

@@ -1281,7 +1281,7 @@ function createToolCatalog(
     .strict();
   const commonSendAgentPromptInputSchema = {
     activeTurnBehavior: ActiveTurnBehaviorSchema.optional().describe(
-      'How to send into a running turn: "steer" joins it, falling back to interrupt when unavailable; "interrupt" replaces it. Defaults to steer for agent callers and interrupt otherwise.',
+      'Agent callers: "steer" (default) waits until the current turn ends; "interrupt" replaces it. Human callers: "steer" joins the turn, falling back to interrupt; default is interrupt.',
     ),
     agentId: z.string(),
     prompt: z.string(),
@@ -1968,7 +1968,7 @@ function createToolCatalog(
     {
       title: "Send agent prompt",
       description:
-        "Send a task. Agent callers receive an immediate queued receipt; delivery follows after two quiet seconds per recipient. Completion reports include the target’s last response by default. Paseo supplies caller identity.",
+        "Send a task. Agent callers receive an immediate queued receipt; delivery follows after two quiet seconds per recipient and waits for the current turn to end unless interrupt is requested. Completion reports include the target’s last response by default. Paseo supplies caller identity.",
       inputSchema: sendAgentPromptInputSchema,
       outputSchema: {
         success: z.boolean(),
@@ -2045,8 +2045,8 @@ function createToolCatalog(
             messageId: receipt.messageId,
             status: agentManager.getAgent(agentId)?.lifecycle ?? "idle",
             guidance: notifying
-              ? `Queued for delivery after two quiet seconds. ${PROMPTED_AGENT_NOTIFICATION_GUIDANCE}`
-              : "Queued for delivery after two quiet seconds; provider dispatch has not occurred.",
+              ? `Queued for delivery after two quiet seconds and the current turn ends. ${PROMPTED_AGENT_NOTIFICATION_GUIDANCE}`
+              : "Queued for delivery after two quiet seconds and the current turn ends; provider dispatch has not occurred.",
           },
         };
       }

packages/server/src/server/daemon-e2e/peer-message-permissions.test.ts

@@ -95,7 +95,7 @@ test("an agent-sourced steer leaves the question pending and answerable", async
     await client.sendAgentMessage(recipient.id, "Peer status check", {
       sourceAgentId: sender.id,
     });
-    // The public wait includes admitted message delivery before reporting permission status.
+    // Permission checkpoints are reported while the peer message waits for turn end.
     const waiting = await client.waitForFinish(recipient.id);
     expect(waiting.status).toBe("permission");
     expect(waiting.final?.pendingPermissions).toMatchObject([

packages/server/src/server/test-utils/agent-messages.ts

@@ -14,6 +14,15 @@ export function createTestAgentMessages(params: {
   logger: Logger;
 }) {
   return createAgentMessageDelivery({
+    isRunning: (agentId) => params.agentManager.hasInFlightRun(agentId),
+    hasPendingPermission: (agentId) => params.agentManager.hasPendingPermissions(agentId),
+    subscribe: (notify) =>
+      params.agentManager.subscribe(
+        (event) => {
+          if (event.type === "agent_state") notify(event.agent.id);
+        },
+        { replayState: false },
+      ),
     schedule: (_delay, callback) => {
       const timer = setTimeout(callback, 0);
       return () => clearTimeout(timer);
@@ -26,8 +35,8 @@ export function createTestAgentMessages(params: {
         sessionMode: request.sessionMode,
         unarchive: request.unarchive,
         activeTurnBehavior: request.activeTurnBehavior,
+        idleOnly: request.idleOnly,
         clearPendingPermissions: request.clearPendingPermissions,
-        onDispatch: request.onDispatch,
         runOptions: { clientMessageId: request.messageIds[0] },
       });
       if (result.disposition === "turn_started")

packages/server/src/server/websocket-server.ts

@@ -728,6 +728,15 @@ export class VoiceAssistantWebSocketServer {
     this.agentManager = agentManager;
     this.agentStorage = agentStorage;
     this.agentMessages = createAgentMessageDelivery({
+      isRunning: (agentId) => agentManager.hasInFlightRun(agentId),
+      hasPendingPermission: (agentId) => agentManager.hasPendingPermissions(agentId),
+      subscribe: (notify) =>
+        agentManager.subscribe(
+          (event) => {
+            if (event.type === "agent_state") notify(event.agent.id);
+          },
+          { replayState: false },
+        ),
       dispatch: async (batch) => {
         const { disposition } = await sendPreparedPromptToAgent({
           agentManager,
@@ -735,11 +744,11 @@ export class VoiceAssistantWebSocketServer {
           agentId: batch.recipientId,
           prompt: batch.prompt,
           activeTurnBehavior: batch.activeTurnBehavior,
+          idleOnly: batch.idleOnly,
           sessionMode: batch.sessionMode,
           clearPendingPermissions: batch.clearPendingPermissions,
           unarchive: batch.unarchive,
           runOptions: { clientMessageId: batch.messageIds[0] },
-          onDispatch: batch.onDispatch,
           logger: this.logger,
         });
         if (disposition === "turn_started")

skills/paseo/SKILL.md

@@ -84,7 +84,7 @@ Detach is an explicit user action in the subagents track, not an agent tool. A c
 
 Agent-scoped `create_agent` defaults `notifyOnFinish` to true. Set it to `false` when you want explicit reporting or no completion callback.
 
-**`send_agent_prompt`** — `{ agentId, prompt, activeTurnBehavior? }`. Use for follow-ups to an existing agent. Agent-scoped calls default to `steer`; top-level calls default to `interrupt`. Set `activeTurnBehavior` to `steer` or `interrupt` to override the default. Steering falls back to interruption when unavailable. CLI agent sends use the same defaults and accept `--behavior steer|interrupt`. Agent-scoped calls return a queued receipt immediately; messages to each recipient deliver after two quiet seconds. Top-level calls return after dispatch acknowledgement. Agent-scoped calls default `notifyOnFinish` to true; top-level calls have no parent callback.
+**`send_agent_prompt`** — `{ agentId, prompt, activeTurnBehavior? }`. Use for follow-ups to an existing agent. Agent-scoped calls default to `steer`; top-level calls default to `interrupt`. For agent-scoped calls, `steer` waits until the recipient's current turn ends, and only `interrupt` stops a running turn. For top-level calls, `steer` joins the running turn and falls back to interruption when unavailable. CLI agent sends use the same defaults and accept `--behavior steer|interrupt`. Agent-scoped calls return a queued receipt immediately; messages to each recipient deliver together after two quiet seconds, or after the recipient's turn ends if it is running. Top-level calls return after dispatch acknowledgement. Agent-scoped calls default `notifyOnFinish` to true; top-level calls have no parent callback.
 
 **`update_agent`** — `{ agentId, name?, labels?, settings? }`. Use `settings` for runtime changes on an existing agent: `modeId`, `model`, `thinkingOptionId`, and provider-specific `features`. For Codex fast mode, pass `settings: { features: { "fast_mode": true } }`.