fix: hold completion notifications until the parent run settles

src/index.ts

@@ -24,6 +24,7 @@ import { isModelInScope, readEnabledModels, resolveEnabledModels } from "./enabl
 import { GroupJoinManager } from "./group-join.js";
 import { resolveAgentInvocationConfig, resolveJoinMode } from "./invocation-config.js";
 import { type ModelRegistry, resolveModel } from "./model-resolver.js";
+import { DEFAULT_HOLD_MS, NudgeQueue } from "./nudge-queue.js";
 import { createOutputFilePath, streamToOutputFile, writeInitialEntry } from "./output-file.js";
 import { SubagentScheduler } from "./schedule.js";
 import { resolveStorePath, ScheduleStore } from "./schedule-store.js";
@@ -326,28 +327,23 @@ export default function (pi: ExtensionAPI) {
   const agentActivity = new Map<string, AgentActivity>();
 
   // ---- Cancellable pending notifications ----
-  // Holds notifications briefly so get_subagent_result can cancel them
-  // before they reach pi.sendMessage (fire-and-forget).
-  const pendingNudges = new Map<string, ReturnType<typeof setTimeout>>();
-  const NUDGE_HOLD_MS = 200;
+  // Holds notifications briefly so get_subagent_result can cancel them before
+  // they reach pi.sendMessage (fire-and-forget), and parks any that come due
+  // mid-run until the parent settles — see nudge-queue.ts for why emitting
+  // during a run is worse than holding.
+  const NUDGE_HOLD_MS = DEFAULT_HOLD_MS;
   // A queued result wait must observe completion before its held notification
   // can fire, so successful waits can still suppress that redundant nudge.
   const QUEUE_WAIT_POLL_MS = Math.floor(NUDGE_HOLD_MS / 4);
 
+  const nudges = new NudgeQueue(() => currentCtx?.isIdle() === false, NUDGE_HOLD_MS);
+
   function scheduleNudge(key: string, send: () => void, delay = NUDGE_HOLD_MS) {
-    cancelNudge(key);
-    pendingNudges.set(key, setTimeout(() => {
-      pendingNudges.delete(key);
-      try { send(); } catch { /* ignore stale completion side-effect errors */ }
-    }, delay));
+    nudges.schedule(key, send, delay);
   }
 
   function cancelNudge(key: string) {
-    const timer = pendingNudges.get(key);
-    if (timer != null) {
-      clearTimeout(timer);
-      pendingNudges.delete(key);
-    }
+    nudges.cancel(key);
   }
 
   // ---- Individual nudge helper (async join mode) ----
@@ -569,6 +565,16 @@ export default function (pi: ExtensionAPI) {
     if (isSchedulingEnabled() && !scheduler.isActive()) startScheduler(ctx);
   });
 
+  // Deliver notifications parked while the parent was running. `agent_settled`
+  // rather than `agent_end` or `turn_end`: those fire with a retry, compaction,
+  // or another tool-calling turn still ahead, and a notification emitted then is
+  // parked by pi's follow-up queue exactly as before — unsuppressable and
+  // delivered after the final answer. `agent_settled` is the first point where
+  // pi will not continue on its own.
+  pi.on("agent_settled", () => {
+    nudges.flush();
+  });
+
   pi.on("session_before_switch", () => {
     manager.clearCompleted(true);
     scheduler.stop();
@@ -589,8 +595,7 @@ export default function (pi: ExtensionAPI) {
     }
     scheduler.stop();
     manager.abortAll();
-    for (const timer of pendingNudges.values()) clearTimeout(timer);
-    pendingNudges.clear();
+    nudges.dispose();
     fleet.dispose();
     manager.dispose();
   });
@@ -836,6 +841,7 @@ If the target is already known, use a direct tool — \`read\` for a known path,
 - When the agent is done, it returns a single message back to you. The result is not visible to the user — to show the user, send a text message with a concise summary.
 - Trust but verify: an agent's summary describes what it intended to do, not necessarily what it did. When an agent writes or edits code, check the actual changes before reporting work as done.
 - Use run_in_background for work you don't need immediately. You will be notified when it completes — do NOT poll or sleep waiting for it. Continue with other work or respond to the user instead.
+- "Do not poll" forbids sleeping and status-checking loops, not waiting. When the user's request is only satisfied once the agents' output is integrated — reviewing their diffs, opening PRs, writing the summary they asked for — join them with get_subagent_result(wait: true) rather than ending your turn on a "work is running" status report.
 - Foreground vs background: use foreground (default) when you need the agent's results before you can proceed. Use background when you have genuinely independent work to do in parallel.
 - Use resume with an agent ID to continue a previous agent's work. A new (non-resume) Agent call starts a fresh agent with no memory of prior runs, so the prompt must be self-contained.
 - Use steer_subagent to send mid-run messages to a running background agent.

src/nudge-queue.ts

@@ -0,0 +1,104 @@
+/**
+ * nudge-queue.ts — Delivery timing for background agent completion notifications.
+ *
+ * Notifications are sent with `deliverAs: "followUp"`, which pi delivers only
+ * once the agent has no more tool calls. Emitting one mid-run therefore parks it
+ * in pi's follow-up queue until the run ends — and a parked message can no
+ * longer be withdrawn, so an agent the orchestrator joins with
+ * `get_subagent_result` in the meantime still produces a notification after the
+ * final answer. With pi's default `followUpMode: "one-at-a-time"` a batch of
+ * those drains one wasted turn each, and a large enough batch can force a
+ * compaction.
+ *
+ * This queue keeps due notifications in-process instead. `schedule` holds each
+ * one for a short window (so a same-tick join can cancel it), then either sends
+ * it — parent idle, delivery is immediate and useful — or parks it here until
+ * `flush` runs at `agent_settled`. Because every send closure re-checks
+ * `resultConsumed` before emitting, deferring the call defers the check: an
+ * agent joined in the meantime simply never notifies.
+ *
+ * The queue deliberately holds nothing across a session shutdown: results are
+ * undeliverable once the session is gone, matching `abortAll()` on shutdown.
+ */
+
+/** Window that lets a same-tick `get_subagent_result` cancel a due notification. */
+export const DEFAULT_HOLD_MS = 200;
+
+export class NudgeQueue {
+  private pending = new Map<string, ReturnType<typeof setTimeout>>();
+  private held = new Map<string, () => void>();
+
+  /**
+   * @param isBusy Whether the parent agent is mid-run. A due notification is
+   *   parked while this is true. Read at delivery time rather than tracked as
+   *   local state so an unbalanced lifecycle event cannot strand notifications.
+   * @param holdMs Cancellation window applied before a notification comes due.
+   */
+  constructor(
+    private readonly isBusy: () => boolean,
+    private readonly holdMs: number = DEFAULT_HOLD_MS,
+  ) {}
+
+  /** Number of notifications parked waiting for the parent run to settle. */
+  get heldCount(): number {
+    return this.held.size;
+  }
+
+  /** Number of notifications inside their cancellation window. */
+  get pendingCount(): number {
+    return this.pending.size;
+  }
+
+  /** Queue `send` under `key`, replacing any notification already queued for it. */
+  schedule(key: string, send: () => void, delay: number = this.holdMs): void {
+    this.cancel(key);
+    this.pending.set(
+      key,
+      setTimeout(() => {
+        this.pending.delete(key);
+        if (this.isBusy()) {
+          this.held.set(key, send);
+          return;
+        }
+        this.deliver(send);
+      }, delay),
+    );
+  }
+
+  /** Drop `key` from both the cancellation window and the parked set. */
+  cancel(key: string): void {
+    const timer = this.pending.get(key);
+    if (timer != null) {
+      clearTimeout(timer);
+      this.pending.delete(key);
+    }
+    this.held.delete(key);
+  }
+
+  /**
+   * Deliver everything parked while the parent was running. Sends re-check
+   * their own relevance, so this is safe to call whenever the agent settles —
+   * including when nothing is parked.
+   */
+  flush(): void {
+    if (this.held.size === 0) return;
+    const sends = [...this.held.values()];
+    this.held.clear();
+    for (const send of sends) this.deliver(send);
+  }
+
+  /** Drop everything without delivering. */
+  dispose(): void {
+    for (const timer of this.pending.values()) clearTimeout(timer);
+    this.pending.clear();
+    this.held.clear();
+  }
+
+  private deliver(send: () => void): void {
+    try {
+      send();
+    } catch {
+      /* ignore stale completion side-effect errors */
+    }
+  }
+}

test/nudge-queue.test.ts

@@ -0,0 +1,156 @@
+/**
+ * nudge-queue.test.ts — Delivery timing for completion notifications.
+ *
+ * The regression this guards: a notification emitted while the parent agent is
+ * mid-run is parked by pi's follow-up queue until the run ends, where it can no
+ * longer be suppressed — so agents the orchestrator already joined with
+ * `get_subagent_result` still notified after the final answer, one wasted turn
+ * each. Fake timers keep the hold window deterministic.
+ */
+
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { DEFAULT_HOLD_MS, NudgeQueue } from "../src/nudge-queue.js";
+
+describe("NudgeQueue", () => {
+  beforeEach(() => vi.useFakeTimers());
+  afterEach(() => vi.useRealTimers());
+
+  it("delivers after the hold window when the parent is idle", () => {
+    const send = vi.fn();
+    const q = new NudgeQueue(() => false);
+
+    q.schedule("a", send);
+    expect(send).not.toHaveBeenCalled();
+
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+    expect(send).toHaveBeenCalledTimes(1);
+    expect(q.heldCount).toBe(0);
+  });
+
+  it("parks instead of delivering while the parent is mid-run", () => {
+    const send = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", send);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    expect(send).not.toHaveBeenCalled();
+    expect(q.heldCount).toBe(1);
+  });
+
+  it("delivers parked notifications on flush", () => {
+    const first = vi.fn();
+    const second = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", first);
+    q.schedule("b", second);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+    expect(q.heldCount).toBe(2);
+
+    q.flush();
+    expect(first).toHaveBeenCalledTimes(1);
+    expect(second).toHaveBeenCalledTimes(1);
+    expect(q.heldCount).toBe(0);
+  });
+
+  it("never delivers a notification cancelled while parked", () => {
+    const send = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", send);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    // The orchestrator joins the agent with get_subagent_result mid-run.
+    q.cancel("a");
+    q.flush();
+
+    expect(send).not.toHaveBeenCalled();
+    expect(q.heldCount).toBe(0);
+  });
+
+  it("cancels a notification still inside its hold window", () => {
+    const send = vi.fn();
+    const q = new NudgeQueue(() => false);
+
+    q.schedule("a", send);
+    q.cancel("a");
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS * 10);
+
+    expect(send).not.toHaveBeenCalled();
+    expect(q.pendingCount).toBe(0);
+  });
+
+  it("re-reads busy state at delivery time rather than at schedule time", () => {
+    const send = vi.fn();
+    let busy = false;
+    const q = new NudgeQueue(() => busy);
+
+    q.schedule("a", send);
+    busy = true; // parent started a run inside the hold window
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    expect(send).not.toHaveBeenCalled();
+    expect(q.heldCount).toBe(1);
+  });
+
+  it("replaces an earlier notification for the same key", () => {
+    const stale = vi.fn();
+    const fresh = vi.fn();
+    const q = new NudgeQueue(() => false);
+
+    q.schedule("a", stale);
+    q.schedule("a", fresh);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    expect(stale).not.toHaveBeenCalled();
+    expect(fresh).toHaveBeenCalledTimes(1);
+  });
+
+  it("flushes each parked notification exactly once", () => {
+    const send = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", send);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    q.flush();
+    q.flush();
+
+    expect(send).toHaveBeenCalledTimes(1);
+  });
+
+  it("keeps delivering after a send throws", () => {
+    const boom = vi.fn(() => {
+      throw new Error("stale record");
+    });
+    const ok = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", boom);
+    q.schedule("b", ok);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+
+    expect(() => q.flush()).not.toThrow();
+    expect(ok).toHaveBeenCalledTimes(1);
+  });
+
+  it("drops everything undelivered on dispose", () => {
+    const parked = vi.fn();
+    const inWindow = vi.fn();
+    const q = new NudgeQueue(() => true);
+
+    q.schedule("a", parked);
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS);
+    q.schedule("b", inWindow);
+
+    q.dispose();
+    vi.advanceTimersByTime(DEFAULT_HOLD_MS * 10);
+    q.flush();
+
+    expect(parked).not.toHaveBeenCalled();
+    expect(inWindow).not.toHaveBeenCalled();
+    expect(q.heldCount).toBe(0);
+    expect(q.pendingCount).toBe(0);
+  });
+});