fix(pi): restore async workflow transcripts on replay

Project saved async status when replaying a launch as well as during live polling. Retain each terminal verdict when a late transcript is attached so the extension host can finish its reader immediately.

docs/agent-lifecycle.md

@@ -210,7 +210,7 @@ The rows combine two kinds of children:
 parentAgentId === thisAgent.id  AND  !archivedAt
 ```
 
-- **Provider subagents** are child executions owned by Claude, Codex, or OpenCode. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
+- **Provider subagents** are child executions owned by Claude, Codex, OpenCode, or Pi. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent. Pi rows that a poll keeps alive carry `liveness: "poll"`, and only a marked row can be settled by a sweep once its poll stops reporting — that gate is what stops an idle parent from leaving a card spinning and keeps other providers' rows untouched. Unmarked rows and every other provider settle their own rows.
 
 Clicking either kind opens a workspace tab. On compact layouts it opens in a [sheet](explorer-sidebar.md#sheets) over the subagents list instead; the sheet hides the composer and shows the agent's model and thinking as one line in its place (`PaneSubtitle`, the row provider subagents already use), and a Paseo subagent's sheet offers **Open as tab** (the full-screen glyph in the header) to get the composer back. A Paseo subagent tab is a normal interactive agent pane. A provider subagent tab is a read-only timeline pane with no composer, archive, detach, rewind, or fork actions. It shows its own direct children in a subagents track. Both panes use `AgentStreamView`, so message, reasoning, tool-call, and layout rendering stay identical.
 

docs/markdown-renderer.md

@@ -8,7 +8,7 @@ Native selection belongs to the library's text view. Keep an assistant message i
 
 Workspace file links use the shared file-link resolver. The workspace supplies its client, root and open-file action once; chat can supply the agent's more specific context. URLs use the same link press path. Recognized inline-code paths stay tappable. The library callback carries the tapped occurrence’s syntax context. Preserve it when opening files: explicit Markdown links decode URL escapes, while recognized inline-code paths preserve literal percent characters and use suffix lookup. Identical URL strings in different occurrences must stay distinct. Explicit absolute workspace links receive a relative hover title; existing authored titles remain intact.
 
-Resolve workspace image paths over the existing session RPC, then give the library a cached native file URL or web blob URL. HTTP(S) images pass through unchanged. This preserves relay, SSH and local socket connections, which do not expose a daemon HTTP endpoint. Each mounted renderer owns its image acquisition identity, so a later reply reads a rewritten path again while the earlier mounted reply retains its snapshot. Image taps open the attachment lightbox. File previews resolve relative images against the Markdown file's directory. An image alone on its line becomes its own paragraph before the library sees it, so it renders as a block on every platform; the library only treats sole-child paragraph images as blocks, and with hard soft breaks every newline is a break node, which left own-line images inline on iOS and web. Images that share a line with text stay inline.
+Resolve workspace image paths over the existing session RPC, then give the library a cached native file URL or web blob URL. HTTP(S) images pass through unchanged. This preserves relay, SSH and local socket connections, which do not expose a daemon HTTP endpoint. Each mounted renderer owns its image acquisition identity, so a later reply reads a rewritten path again while the earlier mounted reply retains its snapshot. Image taps open the attachment lightbox. File previews resolve relative images against the Markdown file's directory.
 
 ## Engine constraints
 

packages/app/src/components/markdown/renderer.test.ts

@@ -1,8 +1,5 @@
 import { fromMarkdown } from "mdast-util-from-markdown";
 import { toNativeLinkRecognition } from "react-native-enriched-markdown/src/utils/regexParser";
-import { parseMarkdown } from "react-native-enriched-markdown/src/web/parseMarkdown";
-import { markInlineImages } from "react-native-enriched-markdown/src/web/utils";
-import type { ASTNode } from "react-native-enriched-markdown/src/web/types";
 import { QueryClient } from "@tanstack/react-query";
 import { decodeMarkdownImage, getMarkdownImageStatus } from "./image-decoding";
 import { describe, expect, it } from "vitest";
@@ -34,56 +31,6 @@ describe("shared Markdown source adaptation", () => {
       "> - ![Image](<file:///cache/a.png>)\n\n[picture]: ../a.png",
     );
   });
-  it("gives an image alone on its line its own paragraph", () => {
-    expect(
-      prepareMarkdownSource("Before the shot:\n![Shot](https://x/a.png)\nAfter it.").render(),
-    ).toBe("Before the shot:\n\n![Shot](https://x/a.png)\n\nAfter it.");
-    expect(prepareMarkdownSource("![A](https://x/a.png)\n![B](https://x/b.png)").render()).toBe(
-      "![A](https://x/a.png)\n\n![B](https://x/b.png)",
-    );
-    expect(
-      prepareMarkdownSource(
-        "Line one\\\n[![Linked](https://x/a.png)](https://x)\nLine three",
-      ).render(),
-    ).toBe("Line one\n\n[![Linked](https://x/a.png)](https://x)\n\nLine three");
-  });
-  it("keeps images that share a line with text inline", () => {
-    const source = "Icon ![i](https://x/i.png) in text\n![a](https://x/a.png) caption\nNext line";
-    expect(prepareMarkdownSource(source).render()).toBe(source);
-  });
-  it("splits own-line images inside quotes and list items without ending the container", () => {
-    expect(prepareMarkdownSource("> Quote text\n> ![Shot](https://x/a.png)\n> More").render()).toBe(
-      "> Quote text\n>\n> ![Shot](https://x/a.png)\n>\n> More",
-    );
-    expect(
-      prepareMarkdownSource("- Item text\n  ![Shot](https://x/a.png)\n  More\n- Next").render(),
-    ).toBe("- Item text\n\n  ![Shot](https://x/a.png)\n\n  More\n- Next");
-    expect(prepareMarkdownSource("- ![Shot](https://x/a.png)\n  Caption line").render()).toBe(
-      "- ![Shot](https://x/a.png)\n\n  Caption line",
-    );
-  });
-  it("renders own-line images as blocks through the library parser", async () => {
-    const source =
-      "Before:\n![Shot](https://x/a.png)\nIcon ![i](https://x/i.png) in text\n> Quoted\n> ![Q](https://x/q.png)";
-    const tree = await parseMarkdown(prepareMarkdownSource(source).render(), {
-      hardSoftBreaks: true,
-      latexMath: false,
-    });
-    markInlineImages(tree);
-    const images: ASTNode[] = [];
-    const collect = (node: ASTNode) => {
-      if (node.type === "Image") images.push(node);
-      node.children?.forEach(collect);
-    };
-    collect(tree);
-    expect(
-      images.map((image) => [image.attributes?.url, image.attributes?.isInline === true]),
-    ).toEqual([
-      ["https://x/a.png", false],
-      ["https://x/i.png", true],
-      ["https://x/q.png", false],
-    ]);
-  });
   it("escapes spoiler delimiters only in prose", () => {
     expect(prepareMarkdownSource("Show ||words|| and `||code||`.").render()).toBe(
       "Show \\|\\|words\\|\\| and `||code||`.",

packages/app/src/components/markdown/source.ts

@@ -5,7 +5,7 @@ import { gfmTableFromMarkdown } from "mdast-util-gfm-table";
 import type { AssistantFileLinkSource } from "@/assistant-file-links";
 import { decodeString } from "micromark-util-decode-string";
 import { fromMarkdown } from "mdast-util-from-markdown";
-import type { Link, Nodes, Paragraph, PhrasingContent, Root } from "mdast";
+import type { Link, Nodes, Root } from "mdast";
 
 // Keep this regex in the syntax shared by JavaScript, ICU and java.util.regex.
 // The Unicode flag makes Metro expand it to escapes rejected by native engines.
@@ -130,50 +130,10 @@ export function prepareMarkdownSource(
         end,
       });
   }
-  // Line boundaries already turned into paragraph breaks, by offset of their newline.
-  const splitBoundaries = new Set<number>();
-  function isImageLike(node: PhrasingContent): boolean {
-    if (node.type === "image" || node.type === "imageReference") return true;
-    return (
-      (node.type === "link" || node.type === "linkReference") &&
-      node.children.length === 1 &&
-      isImageLike(node.children[0])
-    );
-  }
-  // An image alone on its line becomes its own paragraph, so every platform renders
-  // it as a block. The library only treats sole-child paragraph images as blocks;
-  // with hardSoftBreaks every newline is a break node, which made these inline.
-  function splitStandaloneImages(paragraph: Paragraph, start: number, end: number) {
-    for (const child of paragraph.children) {
-      const childStart = child.position?.start.offset;
-      const childEnd = child.position?.end.offset;
-      if (childStart === undefined || childEnd === undefined || !isImageLike(child)) continue;
-      const lineStart = source.lastIndexOf("\n", childStart - 1) + 1;
-      const newline = source.indexOf("\n", childEnd);
-      const lineEnd = newline === -1 ? source.length : newline;
-      const prefix = source.slice(lineStart, childStart);
-      const suffix = source.slice(childEnd, lineEnd);
-      const aloneBefore = childStart === start || /^[\s>]*$/.test(prefix);
-      const aloneAfter = /^\s*\\?$/.test(suffix);
-      if (!aloneBefore || !aloneAfter) continue;
-      // Continuation lines inside quotes and lists keep their quote markers on the blank line.
-      const blank = prefix.replace(/[^>]/g, " ").trimEnd();
-      const split = (boundary: number) => {
-        if (splitBoundaries.has(boundary)) return;
-        splitBoundaries.add(boundary);
-        // A trailing backslash was a hard break; at a paragraph end it would render literally.
-        const cut = source[boundary - 1] === "\\" ? boundary - 1 : boundary;
-        edits.push({ start: cut, end: boundary + 1, text: `\n${blank}\n` });
-      };
-      if (lineStart > start) split(lineStart - 1);
-      if (lineEnd < end) split(lineEnd);
-    }
-  }
   function visit(node: Nodes) {
     const start = node.position?.start.offset;
     const end = node.position?.end.offset;
     if (start !== undefined && end !== undefined) {
-      if (node.type === "paragraph") splitStandaloneImages(node, start, end);
       if (node.type === "link") {
         const title = workspaceHoverTitle(node, source.slice(start, end), workspaceRoot);
         if (title !== null)

packages/cli/src/cli-surface.test.ts

@@ -2,22 +2,6 @@ import { describe, expect, it } from "vitest";
 import { createCli } from "./cli.js";
 
 describe("canonical CLI surface", () => {
-  it.each(["send", "agent send"])("%s accepts an explicit behavior override", async (path) => {
-    const cli = createCli();
-    const owner =
-      path === "send" ? cli : cli.commands.find((command) => command.name() === "agent")!;
-    const send = owner.commands.find((command) => command.name() === "send")!;
-    send.action(() => {});
-    await send.parseAsync(["target", "message", "--behavior", "interrupt"], { from: "user" });
-    expect(send.opts().behavior).toBe("interrupt");
-    await send.parseAsync(["target", "message", "--behavior", "steer"], { from: "user" });
-    expect(send.opts().behavior).toBe("steer");
-    send.exitOverride().configureOutput({ writeErr: () => {} });
-    await expect(
-      send.parseAsync(["target", "message", "--behavior", "queue"], { from: "user" }),
-    ).rejects.toThrow();
-  });
-
   it("offers daemon host selection as a global option", () => {
     expect(createCli().helpInformation()).toContain("--host <host>");
   });

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

@@ -1,5 +1,4 @@
-import { type Command, Option } from "commander";
-import type { ActiveTurnBehavior } from "@getpaseo/protocol/messages";
+import { type Command } from "commander";
 import { collectMultiple } from "../../utils/command-options.js";
 import { connectToDaemon } from "../../utils/client.js";
 import type {
@@ -29,7 +28,6 @@ export const agentSendSchema: OutputSchema<AgentSendResult> = {
 };
 
 export interface AgentSendOptions extends CommandOptions {
-  behavior?: ActiveTurnBehavior;
   wait?: boolean;
   image?: string[];
   prompt?: string;
@@ -44,12 +42,6 @@ export function addSendOptions(cmd: Command): Command {
     .option("--prompt <text>", "Provide the message inline as a flag")
     .option("--prompt-file <path>", "Read the message from a UTF-8 text file")
     .option("--image <path>", "Attach image(s) to the message", collectMultiple, [])
-    .addOption(
-      new Option(
-        "--behavior <behavior>",
-        "Active-turn behavior (agent default: steer; otherwise: interrupt)",
-      ).choices(["interrupt", "steer"]),
-    )
     .option("--no-wait", "Return immediately without waiting for completion");
 }
 
@@ -199,12 +191,10 @@ export async function runSendCommand(
     const images =
       options.image && options.image.length > 0 ? await readImageFiles(options.image) : undefined;
 
-    const sourceAgentId = process.env.PASEO_AGENT_ID?.trim() || undefined;
     // Send the message
     await client.sendAgentMessage(agentIdArg, promptInput, {
       images,
-      sourceAgentId,
-      activeTurnBehavior: options.behavior ?? (sourceAgentId ? "steer" : "interrupt"),
+      sourceAgentId: process.env.PASEO_AGENT_ID?.trim() || undefined,
     });
 
     // If --no-wait, return immediately

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

@@ -0,0 +1,661 @@
+import { mkdtempSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import pino from "pino";
+import { afterEach, describe, expect, test, vi } from "vitest";
+
+import { createTestLogger } from "../../test-utils/test-logger.js";
+import { AgentManager } from "./agent-manager.js";
+import { PiRpcAgentClient } from "./providers/pi/agent.js";
+import { FakePi } from "./providers/pi/test-utils/fake-pi.js";
+import type {
+  AgentCapabilityFlags,
+  AgentClient,
+  AgentPersistenceHandle,
+  AgentPromptInput,
+  AgentProvider,
+  AgentRunResult,
+  AgentSession,
+  AgentSessionConfig,
+  AgentStreamEvent,
+  ProviderCatalog,
+} from "./agent-sdk-types.js";
+import type { ProviderSubagentInputEvent } from "./provider-subagents/store.js";
+
+/**
+ * Stuck-running provider subagent cards are a lifecycle problem, not a presentation one: a row is
+ * born `running` and only a provider update can settle it. These cover the manager's two finalize
+ * points — the turn boundary and history replay — and, above all, that neither touches a row whose
+ * run can still report.
+ */
+
+const TEST_CAPABILITIES: AgentCapabilityFlags = {
+  supportsStreaming: true,
+  supportsSessionPersistence: true,
+  supportsDynamicModes: false,
+  supportsMcpServers: false,
+  supportsReasoningStream: false,
+  supportsToolInvocations: false,
+};
+
+const AGENT_ID = "00000000-0000-4000-8000-000000000001";
+
+/** Long enough that no poll could have re-reported the row inside it. */
+const QUIET_FOR_MS = 60_000;
+
+class TestAgentSession implements AgentSession {
+  readonly capabilities = TEST_CAPABILITIES;
+  readonly id = "test-session";
+  private historyEvents: AgentStreamEvent[] = [];
+  private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
+
+  constructor(readonly provider: AgentProvider) {}
+
+  async run(): Promise<AgentRunResult> {
+    return { sessionId: this.id, finalText: "", timeline: [] };
+  }
+
+  async startTurn(_prompt: AgentPromptInput): Promise<{ turnId: string }> {
+    return { turnId: "turn-1" };
+  }
+
+  subscribe(callback: (event: AgentStreamEvent) => void): () => void {
+    this.subscribers.add(callback);
+    return () => {
+      this.subscribers.delete(callback);
+    };
+  }
+
+  setHistory(events: AgentStreamEvent[]): void {
+    this.historyEvents = [...events];
+  }
+
+  pushEvent(event: AgentStreamEvent): void {
+    for (const callback of this.subscribers) callback(event);
+  }
+
+  async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
+    for (const event of this.historyEvents) yield event;
+  }
+
+  async getRuntimeInfo() {
+    return { provider: this.provider, sessionId: this.id, model: null, modeId: null };
+  }
+
+  async getAvailableModes() {
+    return [];
+  }
+
+  async getCurrentMode() {
+    return null;
+  }
+
+  async setMode(): Promise<void> {}
+
+  getPendingPermissions() {
+    return [];
+  }
+
+  async respondToPermission(): Promise<void> {}
+
+  describePersistence(): AgentPersistenceHandle {
+    return { provider: this.provider, sessionId: this.id };
+  }
+
+  async interrupt(): Promise<void> {}
+
+  async close(): Promise<void> {}
+}
+
+class TestAgentClient implements AgentClient {
+  readonly capabilities = TEST_CAPABILITIES;
+  session?: TestAgentSession;
+
+  constructor(readonly provider: AgentProvider) {}
+
+  async createSession(config: AgentSessionConfig): Promise<AgentSession> {
+    this.session = new TestAgentSession(config.provider);
+    return this.session;
+  }
+
+  async resumeSession(): Promise<AgentSession> {
+    this.session ??= new TestAgentSession(this.provider);
+    return this.session;
+  }
+
+  async fetchCatalog(): Promise<ProviderCatalog> {
+    return {
+      models: [{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true }],
+      modes: [],
+    };
+  }
+
+  async isAvailable(): Promise<boolean> {
+    return true;
+  }
+}
+
+interface Harness {
+  manager: AgentManager;
+  session: TestAgentSession;
+  agentId: string;
+  cleanup: () => void;
+}
+
+async function createHarness(provider: AgentProvider): Promise<Harness> {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagents-"));
+  const client = new TestAgentClient(provider);
+  const manager = new AgentManager({
+    clients: { [provider]: client },
+    idFactory: () => AGENT_ID,
+    logger: createTestLogger(),
+  });
+  await manager.createAgent({ provider, cwd: workdir }, AGENT_ID, { workspaceId: undefined });
+  if (!client.session) throw new Error("Test client did not create a session");
+  return {
+    manager,
+    session: client.session,
+    agentId: AGENT_ID,
+    cleanup: () => rmSync(workdir, { recursive: true, force: true }),
+  };
+}
+
+function subagentEvent(
+  provider: AgentProvider,
+  event: ProviderSubagentInputEvent,
+): AgentStreamEvent {
+  return { type: "provider_subagent", provider, event };
+}
+
+function turnCompleted(provider: AgentProvider): AgentStreamEvent {
+  return { type: "turn_completed", provider, turnId: "turn-1" };
+}
+
+async function drainSessionEvents(): Promise<void> {
+  for (let index = 0; index < 10_000; index++) await Promise.resolve();
+}
+
+async function pushEvent(harness: Harness, event: AgentStreamEvent): Promise<void> {
+  harness.session.pushEvent(event);
+  await drainSessionEvents();
+}
+
+function status(harness: Harness, subagentId: string) {
+  return harness.manager.getProviderSubagent(harness.agentId, subagentId)?.status;
+}
+
+/** Moves the clock past the liveness window without touching timers the manager runs on. */
+function advanceClock(ms: number): void {
+  vi.useFakeTimers({ toFake: ["Date"] });
+  vi.setSystemTime(new Date(Date.now() + ms));
+}
+
+/** Waits on real time: the Pi adapter polls on a real interval. */
+async function waitFor(condition: () => boolean): Promise<void> {
+  for (let attempt = 0; attempt < 250; attempt++) {
+    if (condition()) return;
+    await new Promise((resolve) => setTimeout(resolve, 20));
+  }
+  throw new Error("Timed out waiting for the Pi adapter poll");
+}
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+describe("provider subagent finalize", () => {
+  test("a turn end cancels a running Pi row that nothing has re-reported", async () => {
+    const harness = await createHarness("pi");
+    try {
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-1",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      expect(status(harness, "call-1")).toBe("running");
+
+      advanceClock(QUIET_FOR_MS);
+      await pushEvent(harness, turnCompleted("pi"));
+
+      expect(status(harness, "call-1")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a re-check armed by one turn leaves a later turn's synchronous child alone", async () => {
+    const harness = await createHarness("pi");
+    try {
+      vi.useFakeTimers();
+      // A spawn call interrupted before it reported: the turn end arms a re-check for the moment its
+      // silence becomes proof.
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-orphan",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      await pushEvent(harness, turnCompleted("pi"));
+      expect(status(harness, "call-orphan")).toBe("running");
+
+      // The next turn launches a synchronous child. That call is poll-marked at spawn and gets no
+      // poll until its result lands, so it stays quiet for longer than the window while it works.
+      await vi.advanceTimersByTimeAsync(2_000);
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-sync",
+          title: "reviewer",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      expect(status(harness, "call-sync")).toBe("running");
+
+      // The armed re-check settles the row it was armed for. Judging the next turn's fresh row here
+      // would cancel a child whose foreground call is still running.
+      await vi.advanceTimersByTimeAsync(3_001);
+      expect(status(harness, "call-orphan")).toBe("canceled");
+      expect(status(harness, "call-sync")).toBe("running");
+
+      // Nothing captured is left to prove, so nothing is left armed: the fresh row survives a whole
+      // further window and no timer of ours is pending.
+      await vi.advanceTimersByTimeAsync(QUIET_FOR_MS);
+      expect(status(harness, "call-sync")).toBe("running");
+      expect(vi.getTimerCount()).toBe(0);
+
+      // The call's result lands and settles its own row normally.
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-sync",
+          title: "reviewer",
+          status: "completed",
+        }),
+      );
+      expect(status(harness, "call-sync")).toBe("completed");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a turn end inside the window re-checks an orphaned placeholder once its silence proves it", async () => {
+    const harness = await createHarness("pi");
+    try {
+      vi.useFakeTimers();
+      // A spawn call interrupted before it reported leaves this placeholder behind: nothing polls it,
+      // so no later report refreshes it either.
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-1",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      // The stop's turn end lands inside the liveness window, where the sweep cannot tell this row
+      // from a live poll's.
+      await pushEvent(harness, turnCompleted("pi"));
+      expect(status(harness, "call-1")).toBe("running");
+
+      // The parent stays idle, so this re-check is the only thing that will ever look again.
+      await vi.advanceTimersByTimeAsync(5_001);
+      expect(status(harness, "call-1")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a poll that re-dates the row inside the sweep window re-arms the re-check", async () => {
+    const harness = await createHarness("pi");
+    try {
+      vi.useFakeTimers();
+      const poll = () =>
+        pushEvent(
+          harness,
+          subagentEvent("pi", {
+            type: "upsert",
+            id: "call-1",
+            title: "scout",
+            status: "running",
+            liveness: "poll",
+          }),
+        );
+      await poll();
+      // The turn end lands inside the window, so it arms a re-check instead of canceling.
+      await pushEvent(harness, turnCompleted("pi"));
+      expect(status(harness, "call-1")).toBe("running");
+
+      // The live poll re-dates the row before that re-check fires: the scheduled look no longer
+      // proves anything, and cancelling here would kill a run that is still reporting.
+      await vi.advanceTimersByTimeAsync(2_000);
+      await poll();
+      await vi.advanceTimersByTimeAsync(3_001);
+      expect(status(harness, "call-1")).toBe("running");
+
+      // The re-check re-armed itself, so the row settles once the poll really does go quiet.
+      await vi.advanceTimersByTimeAsync(2_000);
+      expect(status(harness, "call-1")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a turn end leaves a Pi row that is still reporting alone", async () => {
+    const harness = await createHarness("pi");
+    try {
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-1",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      await pushEvent(harness, turnCompleted("pi"));
+
+      expect(status(harness, "call-1")).toBe("running");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a turn end leaves a quiet non-polling Pi-extension row alone", async () => {
+    const harness = await createHarness("pi");
+    try {
+      await pushEvent(
+        harness,
+        subagentEvent("pi", { type: "upsert", id: "call-1", title: "scout", status: "running" }),
+      );
+
+      advanceClock(QUIET_FOR_MS);
+      await pushEvent(harness, turnCompleted("pi"));
+
+      expect(status(harness, "call-1")).toBe("running");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("a poll-backed row survives consecutive turn ends while its poll keeps reporting", async () => {
+    const harness = await createHarness("pi");
+    try {
+      const poll = () =>
+        pushEvent(
+          harness,
+          subagentEvent("pi", {
+            type: "upsert",
+            id: "call-1",
+            title: "scout",
+            status: "running",
+            liveness: "poll",
+          }),
+        );
+      await poll();
+      advanceClock(QUIET_FOR_MS);
+      await poll();
+      await pushEvent(harness, turnCompleted("pi"));
+      expect(status(harness, "call-1")).toBe("running");
+
+      advanceClock(QUIET_FOR_MS);
+      await poll();
+      await pushEvent(harness, turnCompleted("pi"));
+      expect(status(harness, "call-1")).toBe("running");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test(
+    "a turn end during a status-file miss streak leaves the live poll's row running",
+    { timeout: 15_000 },
+    async () => {
+      // The real Pi adapter, so the miss ticks come from the poll that protects the row instead of
+      // from a hand-written upsert.
+      const pi = new FakePi();
+      const client = new PiRpcAgentClient({
+        logger: pino({ level: "silent" }),
+        runtime: pi,
+        // The injected runtime owns execution; use an existing executable for the real availability
+        // probe so this test does not depend on a Pi installation or replace a client method.
+        runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
+      });
+      const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagent-poll-"));
+      const manager = new AgentManager({
+        clients: { pi: client },
+        idFactory: () => AGENT_ID,
+        logger: createTestLogger(),
+      });
+      try {
+        await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
+          workspaceId: undefined,
+        });
+        const piSession = pi.latestSession();
+        // An async spawn whose status directory never appears: every poll tick is a miss.
+        piSession.emit({
+          type: "tool_execution_start",
+          toolCallId: "call-1",
+          toolName: "subagent",
+          args: { agent: "scout", task: "Inspect", async: true },
+        });
+        piSession.emit({
+          type: "tool_execution_end",
+          toolCallId: "call-1",
+          toolName: "subagent",
+          result: {
+            details: {
+              mode: "single",
+              runId: "run-1",
+              asyncId: "run-1",
+              asyncDir: join(workdir, "gone"),
+              results: [],
+            },
+          },
+        });
+        const row = () => manager.getProviderSubagent(AGENT_ID, "call-1");
+        await waitFor(() => row()?.status === "running" && row()?.liveness === "poll");
+
+        for (let turn = 0; turn < 2; turn++) {
+          // A miss tick re-reports the row, then the turn ends inside the miss budget: the silence is
+          // a file the run has not written yet, not a dead poll.
+          const reportedAt = row()?.updatedAt;
+          await waitFor(() => row()?.updatedAt !== reportedAt);
+          piSession.emit({ type: "agent_start" });
+          piSession.emit({ type: "agent_end" });
+          await drainSessionEvents();
+          expect(row()?.status).toBe("running");
+        }
+      } finally {
+        try {
+          await manager.closeAgent(AGENT_ID);
+        } catch {
+          // A createAgent failure leaves nothing to close; the workdir cleanup must still run.
+        } finally {
+          rmSync(workdir, { recursive: true, force: true });
+        }
+      }
+    },
+  );
+
+  test("a turn end leaves another provider's quiet row alone: it has no poll to prove life", async () => {
+    const harness = await createHarness("claude");
+    try {
+      await pushEvent(
+        harness,
+        subagentEvent("claude", {
+          type: "upsert",
+          id: "task-1",
+          title: "general-purpose",
+          status: "running",
+        }),
+      );
+
+      advanceClock(QUIET_FOR_MS);
+      await pushEvent(harness, turnCompleted("claude"));
+
+      expect(status(harness, "task-1")).toBe("running");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("replay settles a poll-backed running row: replay has no poll to adopt it", async () => {
+    const harness = await createHarness("pi");
+    try {
+      harness.session.setHistory([
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-polled",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      ]);
+      await harness.manager.hydrateTimelineFromProvider(harness.agentId, { force: true });
+
+      expect(status(harness, "call-polled")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("the prime on agent load settles a poll-backed running row: a restart has no poll either", async () => {
+    const harness = await createHarness("pi");
+    try {
+      // A daemon restart or an agent reload re-reads provider history once, without force. Nothing
+      // re-creates the live session's poll there either, so the row must not come back every reload.
+      harness.session.setHistory([
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-prime",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      ]);
+      await harness.manager.reloadAgentSession(harness.agentId, undefined, {
+        rehydrateFromDisk: true,
+      });
+      expect(harness.manager.getAgent(harness.agentId)?.historyPrimed).toBe(false);
+      await harness.manager.hydrateTimelineFromProvider(harness.agentId, { broadcast: true });
+
+      expect(status(harness, "call-prime")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("replay settles a row that was already canceled: the replay proves no life either", async () => {
+    const harness = await createHarness("pi");
+    try {
+      // A turn-end sweep canceled this row while its poll was quiet, and a rewind replays the same
+      // async spawn: replay maps it back to running with the poll marker, while no live poll is
+      // behind it. Settling on id presence alone would leave a spinner nothing can update, so the
+      // row settles again.
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-canceled",
+          title: "scout",
+          status: "canceled",
+          liveness: "poll",
+        }),
+      );
+      harness.session.setHistory([
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-canceled",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      ]);
+      await harness.manager.hydrateTimelineFromProvider(harness.agentId, { force: true });
+
+      expect(status(harness, "call-canceled")).toBe("canceled");
+      // Nothing is left that could update it, so the row stays settled once the replay completes.
+      await drainSessionEvents();
+      expect(status(harness, "call-canceled")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("replay leaves a row the live poll already owns alone", async () => {
+    const harness = await createHarness("pi");
+    try {
+      // The live session's poll is re-reporting this row when a rewind replays the same spawn. The
+      // replay must not cancel it: its poll is still running and would flip it back to running.
+      await pushEvent(
+        harness,
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-live",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+      );
+      harness.session.setHistory([
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-live",
+          title: "scout",
+          status: "running",
+          liveness: "poll",
+        }),
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-replayed",
+          title: "reviewer",
+          status: "running",
+          liveness: "poll",
+        }),
+      ]);
+      await harness.manager.hydrateTimelineFromProvider(harness.agentId, { force: true });
+
+      expect(status(harness, "call-live")).toBe("running");
+      // The replay's own row has no poll behind it, so it still settles.
+      expect(status(harness, "call-replayed")).toBe("canceled");
+    } finally {
+      harness.cleanup();
+    }
+  });
+
+  test("replay leaves an unmarked running row: its reporter settles itself", async () => {
+    const harness = await createHarness("pi");
+    try {
+      harness.session.setHistory([
+        subagentEvent("pi", { type: "upsert", id: "call-live", title: "scout", status: "running" }),
+        subagentEvent("pi", {
+          type: "upsert",
+          id: "call-done",
+          title: "reviewer",
+          status: "completed",
+        }),
+      ]);
+      await harness.manager.hydrateTimelineFromProvider(harness.agentId, { force: true });
+
+      expect(status(harness, "call-live")).toBe("running");
+      expect(status(harness, "call-done")).toBe("completed");
+    } finally {
+      harness.cleanup();
+    }
+  });
+});

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

@@ -726,6 +726,17 @@ function detachedAgentLabelPatch(labels: Record<string, string>): AgentLabelPatc
 const RETIRED_INTERNAL_AGENT_TTL_MS = 10 * 60 * 1000;
 const RETIRED_INTERNAL_AGENT_LIMIT = 500;
 
+/**
+ * How recently a poll-backed Pi subagent row must have been updated to count as alive. Only the
+ * pi-subagents extension polls: it re-upserts every live row on each 250 ms follow tick and marks
+ * those upserts `liveness: "poll"`, so a marked row untouched for a whole window can never change
+ * again. The other Pi extensions (gotgenes/tintinweb pi-subagents and friends) are event-driven
+ * and may stay silent while a background child still runs, and Claude
+ * (providers/claude/agent.ts:4779,4805) and OpenCode (providers/opencode/v2/children.ts:64-68)
+ * settle their own rows at turn or child end. No row without the marker is ever swept.
+ */
+const PI_SUBAGENT_LIVENESS_MS = 5_000;
+
 export interface RetiredInternalAgent {
   record: ArchivedStoredAgentRecord;
   lastMessage: string | null;
@@ -743,6 +754,7 @@ export class AgentManager {
   private readonly agents = new Map<string, LiveManagedAgent>();
   private readonly timelineStore = new InMemoryAgentTimelineStore();
   private readonly providerSubagents = new ProviderSubagentStore();
+  private readonly providerSubagentSweepTimers = new Map<string, ReturnType<typeof setTimeout>>();
   private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
   private readonly sessionEventTails = new Map<string, Promise<void>>();
   private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
@@ -1812,15 +1824,157 @@ export class AgentManager {
       if (subagent.status !== "running") {
         continue;
       }
-      const event = this.providerSubagents.apply(parentAgentId, subagent.provider, {
-        type: "upsert",
-        id: subagent.id,
-        status: "canceled",
-      });
-      this.dispatch({ type: "provider_subagent", event });
+      this.applyProviderSubagentStatus(parentAgentId, subagent.provider, subagent.id, "canceled");
     }
   }
 
+  private applyProviderSubagentStatus(
+    parentAgentId: string,
+    provider: AgentProvider,
+    subagentId: string,
+    status: ProviderSubagentDescriptor["status"],
+  ): void {
+    const event = this.providerSubagents.apply(parentAgentId, provider, {
+      type: "upsert",
+      id: subagentId,
+      status,
+    });
+    this.dispatch({ type: "provider_subagent", event });
+  }
+
+  /**
+   * Poll-backed Pi subagent rows are born `running` and leave that state only through a provider
+   * update. A row whose poll is gone would keep its spinner forever, so a turn boundary settles
+   * only the rows that prove they had a poll and stopped reporting, as canceled — never completed,
+   * because nothing here proves they succeeded. A row a live poll keeps re-reporting is fresh and
+   * left alone, and a row without the marker belongs to an event-driven reporter that may still be
+   * running its child, so it is never swept.
+   *
+   * Gate: the poll marker plus no report inside the liveness window. The window is wall clock, so an
+   * event-loop stall, a GC pause, or a resume longer than it makes a live run's row look silent; this
+   * path is the only one that reads time — the replay settle (finalizeReplayedProviderSubagents) gates
+   * on the store's pre-replay state instead.
+   *
+   * A turn end can arrive inside that window, and for a spawn call a stop interrupted before it
+   * reported there is no later turn end to look again: the placeholder would stay `running` while the
+   * parent sits idle. So a row still inside the window schedules one re-check for the moment its
+   * silence becomes proof, instead of waiting for an event that may never come.
+   *
+   * That re-check is scoped to the rows it was armed for. A row born after it was armed belongs to a
+   * turn that is still running — a synchronous `subagent` call is poll-marked at spawn and only gets
+   * its poll when the result lands — so the timer must not judge it: that turn's own end sweeps it.
+   */
+  // ponytail: staleness is wall-clock — upgrade to a poll heartbeat (sequence/epoch from the adapter) if stalls prove real
+  private finalizeUnprovenProviderSubagents(
+    agentId: string,
+    scheduledRowIds?: ReadonlySet<string>,
+  ): void {
+    const now = Date.now();
+    let recheckAt: number | undefined;
+    const recheckRowIds: string[] = [];
+    for (const subagent of this.providerSubagents.list(agentId)) {
+      if (scheduledRowIds && !scheduledRowIds.has(subagent.id)) continue;
+      if (subagent.status !== "running" || subagent.liveness !== "poll") continue;
+      const updatedAt = Date.parse(subagent.updatedAt);
+      if (updatedAt >= now - PI_SUBAGENT_LIVENESS_MS) {
+        // A live poll may still own this row, so it stays. Look again once its silence is old enough.
+        recheckAt = Math.min(
+          recheckAt ?? Number.POSITIVE_INFINITY,
+          updatedAt + PI_SUBAGENT_LIVENESS_MS,
+        );
+        recheckRowIds.push(subagent.id);
+        continue;
+      }
+      this.applyProviderSubagentStatus(agentId, subagent.provider, subagent.id, "canceled");
+    }
+    this.scheduleProviderSubagentSweep(agentId, recheckAt, recheckRowIds);
+  }
+
+  /**
+   * The one pending re-check per agent: a turn end while a poll-backed row was still fresh reschedules
+   * it for the rows that were fresh then, and a sweep that finds nothing left to prove clears it. The
+   * timer stops itself once the agent is gone.
+   */
+  private scheduleProviderSubagentSweep(
+    agentId: string,
+    recheckAt: number | undefined,
+    recheckRowIds: readonly string[],
+  ): void {
+    const pending = this.providerSubagentSweepTimers.get(agentId);
+    if (pending) clearTimeout(pending);
+    this.providerSubagentSweepTimers.delete(agentId);
+    if (recheckAt === undefined) return;
+    const scheduledRowIds = new Set(recheckRowIds);
+    const timer = setTimeout(
+      () => {
+        this.providerSubagentSweepTimers.delete(agentId);
+        if (!this.agents.has(agentId)) return;
+        this.finalizeUnprovenProviderSubagents(agentId, scheduledRowIds);
+      },
+      Math.max(0, recheckAt - Date.now()) + 1,
+    );
+    timer.unref();
+    this.providerSubagentSweepTimers.set(agentId, timer);
+  }
+
+  /**
+   * Replay re-creates a row for every spawn in the transcript, including runs that never finished.
+   * Replay re-runs the pi-subagents adapter mapping, so a replayed running pi-subagents row carries
+   * the poll marker. Replay builds its host without `follow()`
+   * (providers/pi/extensions/host.ts:61-69), and the live host's `callsByRun` is per-adapter-session,
+   * so no live poll is behind that row and nothing would ever update it again: left running, it is a
+   * spinner that returns on every reload.
+   *
+   * Gate: the poll marker plus the row was not already `running` in the store before the replay. A
+   * row that was running pre-replay belongs to the live session's poll (rewind force-hydrates without
+   * recreating the session) and is left with it, or the settle would cancel a live run and the next
+   * tick would flip it back to running. Every other replayed running row settles as canceled, never
+   * completed: a row that was already canceled or completed before the replay settles again as
+   * canceled, because replay re-reports every unfinished run as running and the transcript holds no
+   * proof that it survived. That is the verdict for unproven rows, and it is still better than the
+   * spinner that stays until the next reload.
+   *
+   * Known limitation: a row replayed for a run that no live poll owns may still describe work that
+   * is writing status.json, and this cancel is final — such a run keeps working but its card can
+   * never recover. Making replay resumable needs the live host to adopt replayed runs, which this
+   * change does not attempt.
+   *
+   * Rows without the marker belong to a reporter that settles its own children — Claude
+   * (providers/claude/agent.ts:4779,4805), OpenCode (providers/opencode/v2/children.ts:64-68), the
+   * event-driven Pi extensions — so replay leaves them alone. A live turn is the other case where a
+   * replayed tool call may still be in flight, so it is left alone too. Normal callers clear the
+   * turn before hydrating, but the interrupt-timeout rescue (cancelAgentRunNow) settles the
+   * foreground run without a terminal event, so `activeTurnId` can linger; those rows are left to
+   * the next turn-end sweep rather than to a poll.
+   */
+  // ponytail: replay has no live poll; proper fix is letting the live host adopt replayed runs — revisit when host replay shares adapter state
+  private finalizeReplayedProviderSubagents(
+    agent: ActiveManagedAgent,
+    replayedRunningIds: ReadonlySet<string>,
+    preexistingRunningIds: ReadonlySet<string>,
+  ): void {
+    if (agent.activeForegroundTurnId != null || agent.activeTurnId != null) return;
+    for (const id of replayedRunningIds) {
+      if (preexistingRunningIds.has(id)) continue;
+      const subagent = this.providerSubagents.get(agent.id, id);
+      if (!subagent || subagent.status !== "running" || subagent.liveness !== "poll") continue;
+      this.applyProviderSubagentStatus(agent.id, subagent.provider, id, "canceled");
+    }
+  }
+
+  /**
+   * Ids the store reports running now. Before a replay applies, a running row belongs to the live
+   * session's poll; a terminal row belongs to nobody and is replayed evidence like any other.
+   */
+  private providerSubagentRunningIds(agentId: string): Set<string> {
+    return new Set(
+      this.providerSubagents
+        .list(agentId)
+        .filter((subagent) => subagent.status === "running")
+        .map((subagent) => subagent.id),
+    );
+  }
+
   async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
     return this.runLifecycleMutation(agentId, () => this.archiveAgentUnlocked(agentId));
   }
@@ -4146,13 +4300,20 @@ export class AgentManager {
     this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
     agent.historyPrimed = true;
 
+    // Rows the store already reports running are the live session's: a rewind replays history
+    // without recreating the session, so their poll keeps running.
+    const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
     for (const event of this.providerSubagents.deleteParent(agent.id)) {
       if (broadcast) {
         this.dispatch({ type: "provider_subagent", event });
       }
     }
+    const replayedRunningIds = new Set<string>();
     for (const event of providerSubagentEvents) {
       const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
+      if (event.event.type === "upsert" && event.event.status === "running") {
+        replayedRunningIds.add(event.event.id);
+      }
       if (broadcast) {
         this.dispatch({ type: "provider_subagent", event: update });
       }
@@ -4171,6 +4332,7 @@ export class AgentManager {
         });
       }
     }
+    this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
     this.touchUpdatedAt(agent);
     this.emitState(agent);
   }
@@ -4217,8 +4379,15 @@ export class AgentManager {
       row: AgentTimelineRow;
     }> = [];
     const providerSubagentEvents: AgentManagerEvent[] = [];
+    const replayedRunningIds = new Set<string>();
+    // A row the store still reports running is not a fresh resume: it belongs to the live
+    // session's poll.
+    const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
     for (const event of historySubagentEvents) {
       const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
+      if (event.event.type === "upsert" && event.event.status === "running") {
+        replayedRunningIds.add(event.event.id);
+      }
       const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
       if (deferredBroadcast) {
         providerSubagentEvents.push(managerEvent);
@@ -4245,6 +4414,7 @@ export class AgentManager {
     agent.historyPrimed = true;
 
     if (typeof broadcast !== "function" || !broadcast()) {
+      this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
       return;
     }
     for (const event of providerSubagentEvents) {
@@ -4257,6 +4427,7 @@ export class AgentManager {
         timestamp: row.timestamp,
       });
     }
+    this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
   }
 
   private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
@@ -4342,6 +4513,7 @@ export class AgentManager {
         if (isForegroundEvent) {
           this.finalizeForegroundTurn(agent, eventTurnId);
         }
+        this.finalizeUnprovenProviderSubagents(agent.id);
       }
 
       if (flags.shouldDispatchEvent) {

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

@@ -402,8 +402,6 @@ export interface SetupFinishNotificationParams {
   childAgentId: string;
   callerAgentId: string;
   requireParentOwnership?: boolean;
-  /** Retain pre-dispatch outcomes only when dispatch keeps the existing turn. */
-  dispatchResult?: Promise<PromptDispatchDisposition | false>;
   /** A replacement must observe its new start, not the displaced running snapshot. */
   waitForTurnStart?: boolean;
   logger: Logger;
@@ -468,17 +466,10 @@ interface NotifySafelyOptions {
   permissionRequest?: AgentPermissionRequest;
 }
 
-// Pending follow-ups coexist until acknowledgement. A successful dispatch replaces
-// every earlier observation for the caller/child pair; a failure removes only itself.
-interface ArmedFinishNotification {
-  stop(): void;
-  supersedeWhen(accepted: Promise<boolean>): void;
-}
-
-const armedFinishNotifications = new WeakMap<
-  AgentManager,
-  Map<string, Set<ArmedFinishNotification>>
->();
+// A caller waits on a child through one armed notification. Arming again, such as a
+// follow-up prompt while the child still runs, replaces the earlier one so the child's
+// next finish reaches the caller once.
+const armedFinishNotifications = new WeakMap<AgentManager, Map<string, () => void>>();
 
 export function setupFinishNotification(params: SetupFinishNotificationParams): () => void {
   const {
@@ -500,49 +491,15 @@ export function setupFinishNotification(params: SetupFinishNotificationParams):
   const armedByManager = armedFinishNotifications.get(agentManager) ?? new Map();
   armedFinishNotifications.set(agentManager, armedByManager);
   const armedKey = JSON.stringify([childAgentId, callerAgentId]);
-  const registrations = armedByManager.get(armedKey) ?? new Set<ArmedFinishNotification>();
-  armedByManager.set(armedKey, registrations);
-  const previous = [...registrations];
-  let superseded = Promise.resolve(false);
-  let dispatchPending = Boolean(params.dispatchResult);
-  const registration: ArmedFinishNotification = {
-    stop,
-    supersedeWhen(accepted) {
-      superseded = superseded.then(
-        async (alreadySuperseded) => alreadySuperseded || (await accepted),
-      );
-    },
-  };
-  registrations.add(registration);
-  if (params.dispatchResult) {
-    const dispatchAccepted = params.dispatchResult.then((result) => result !== false);
-    for (const observer of previous) observer.supersedeWhen(dispatchAccepted);
-    void params.dispatchResult.then((accepted) => {
-      dispatchPending = false;
-      if (accepted) {
-        for (const observer of previous) observer.stop();
-        if (stopped || accepted === "turn_started") stop();
-      } else {
-        stop();
-      }
-      return undefined;
-    });
-  } else {
-    for (const observer of previous) {
-      observer.supersedeWhen(Promise.resolve(true));
-      observer.stop();
-    }
-  }
+  armedByManager.get(armedKey)?.();
+  armedByManager.set(armedKey, stop);
 
   function stop(): void {
+    if (stopped) return;
     stopped = true;
     unsubscribe?.();
-    // Pending observations remain visible to overlapping sends until acknowledgement.
-    if (!dispatchPending) {
-      registrations.delete(registration);
-      if (registrations.size === 0 && armedByManager.get(armedKey) === registrations) {
-        armedByManager.delete(armedKey);
-      }
+    if (armedByManager.get(armedKey) === stop) {
+      armedByManager.delete(armedKey);
     }
   }
 
@@ -550,11 +507,6 @@ export function setupFinishNotification(params: SetupFinishNotificationParams):
     reason: FinishNotificationReason,
     permissionRequest?: AgentPermissionRequest,
   ): Promise<void> {
-    if (params.dispatchResult) {
-      const result = await params.dispatchResult;
-      if (result === false || result === "turn_started") return;
-    }
-    if (await superseded) return;
     const callerRecord = await agentStorage.get(callerAgentId);
     if (callerRecord?.archivedAt) {
       return;

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

@@ -1,4 +1,3 @@
-import { setupFinishNotification, type PromptDispatchDisposition } from "./agent-prompt.js";
 import { parseAgentMessage } from "./agent-messages/index.js";
 import { execFileSync } from "node:child_process";
 import { randomUUID } from "node:crypto";
@@ -243,7 +242,6 @@ function buildAgentManagerSpies() {
     appendTimelineItem: vi.fn().mockResolvedValue(undefined),
     emitLiveTimelineItem: vi.fn().mockResolvedValue(undefined),
     hasInFlightRun: vi.fn().mockReturnValue(false),
-    steerOrReplaceActiveTurn: async () => ({ status: "inactive" as const }),
     tryRunOutOfBand: vi.fn().mockReturnValue(false),
     subscribe: vi.fn().mockReturnValue(() => {}),
     streamAgent: vi.fn(() => (async function* noop() {})()),
@@ -3862,9 +3860,6 @@ const HELD_TURN_CAPABILITIES = {
 } as const;
 
 interface HeldTurnOptions {
-  finishDuringSteer?: boolean;
-  rejectSteer?: boolean;
-  steerResult?: "accepted" | "unavailable";
   /** Run the whole turn before the turn-start request returns. */
   finishTurnsDuringStart?: boolean;
 }
@@ -3877,7 +3872,6 @@ class HeldTurnAgentSession implements AgentSession {
   readonly capabilities = HELD_TURN_CAPABILITIES;
   readonly id = randomUUID();
   readonly prompts: string[] = [];
-  readonly steeredPrompts: AgentPromptInput[] = [];
   private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
   private activeTurnId: string | null = null;
   private turnStartGate: Promise<void> | null = null;
@@ -3910,14 +3904,6 @@ class HeldTurnAgentSession implements AgentSession {
     return prompt === "/side-command" ? { run: async () => {} } : null;
   }
 
-  async steerActiveTurn(prompt: AgentPromptInput) {
-    const status = this.options.steerResult ?? "unavailable";
-    if (status === "accepted") this.steeredPrompts.push(prompt);
-    if (status === "accepted" && this.options.finishDuringSteer) this.finishTurn();
-    if (this.options.rejectSteer) throw new Error("Steer rejected");
-    return { status };
-  }
-
   async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
     this.prompts.push(typeof prompt === "string" ? prompt : JSON.stringify(prompt));
     await this.turnStartGate;
@@ -4064,295 +4050,6 @@ 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 },
-  ] as const)(
-    "dispatches behavior $behavior for agent-scoped=$agentScoped, steering available=$available",
-    async ({ agentScoped, behavior, available, steered }) => {
-      const workdir = await mkdtemp(join(tmpdir(), "mcp-send-behavior-"));
-      const storage = new AgentStorage(join(workdir, "agents"), logger);
-      const client = new HeldTurnAgentClient("codex", true, {
-        steerResult: available ? "accepted" : "unavailable",
-      });
-      const manager = new AgentManager({ clients: { codex: client }, registry: storage, logger });
-      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: agentScoped ? parent.id : undefined,
-          providerSnapshotManager: createOpenCodeManager().manager,
-          logger,
-        });
-        const tool = registeredTool(server, "send_agent_prompt");
-        await invokeToolWithParsedInput(tool, {
-          agentId: child.id,
-          prompt: "Original work",
-          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 } : {}),
-        });
-        expect(response.structuredContent).toMatchObject({
-          success: true,
-          disposition: steered ? "steered" : "turn_started",
-        });
-        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);
-      } finally {
-        await removeAgentStateDir(manager, storage, workdir);
-      }
-    },
-  );
-
-  it.each([
-    { rejected: false, priorNotification: false },
-    { rejected: true, priorNotification: false },
-    { rejected: false, priorNotification: true },
-  ])(
-    "buffers completion until steer acknowledgement (rejected: $rejected, prior notification: $priorNotification)",
-    async ({ rejected, priorNotification }) => {
-      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: true,
-        rejectSteer: rejected,
-      });
-      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>) =>
-          mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
-        await send({
-          agentId: child.id,
-          prompt: "Original work",
-          notifyOnFinish: priorNotification,
-        });
-        const sending = send({
-          agentId: child.id,
-          prompt: "New message",
-          notifyOnFinish: true,
-        });
-        if (rejected) {
-          expect(await sending).toMatchObject({
-            isError: true,
-            content: [{ type: "text", text: expect.stringContaining("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",
-          });
-          client.sessions[1]!.finishTurn();
-        } else {
-          const response = await sending;
-          expect(response.structuredContent).toMatchObject({
-            success: true,
-            disposition: "steered",
-          });
-        }
-        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>) =>
-          mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
-        await send({
-          agentId: child.id,
-          prompt: "Original work",
-          notifyOnFinish: true,
-        });
-        const sending = send({
-          agentId: child.id,
-          prompt: "New message",
-          notifyOnFinish: true,
-        });
-        expect(await sending).toMatchObject({
-          isError: true,
-          content: [{ type: "text", text: expect.stringContaining("Steer rejected") }],
-        });
-        if (!finishDuringSteer) {
-          await new Promise((resolve) => setTimeout(resolve, 50));
-          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);
-      }
-    },
-  );
-
-  it.each([
-    [false, true],
-    [true, false],
-    [false, false],
-    [true, true],
-  ])(
-    "overlapping follow-ups notify once (first accepted: %s, second accepted: %s)",
-    async (firstAccepted, secondAccepted) => {
-      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: false,
-        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>) =>
-          mcpClient!.callTool({ name: "send_agent_prompt", arguments: input });
-        await send({
-          agentId: child.id,
-          prompt: "Original work",
-          notifyOnFinish: true,
-        });
-        let firstResolve!: (result: PromptDispatchDisposition | false) => void;
-        let secondResolve!: (result: PromptDispatchDisposition | false) => void;
-        const first = new Promise<PromptDispatchDisposition | false>((resolve) => {
-          firstResolve = resolve;
-        });
-        const second = new Promise<PromptDispatchDisposition | false>((resolve) => {
-          secondResolve = resolve;
-        });
-        for (const dispatchResult of [first, second]) {
-          setupFinishNotification({
-            agentManager: manager,
-            agentStorage: storage,
-            childAgentId: child.id,
-            callerAgentId: parent.id,
-            dispatchResult,
-            logger,
-          });
-        }
-        firstResolve(firstAccepted ? "steered" : false);
-        await first;
-        secondResolve(secondAccepted ? "steered" : false);
-        await second;
-        client.sessions[1]!.finishTurn();
-        await vi.waitFor(() => expect(client.sessions[0]!.prompts).toHaveLength(1));
-        await new Promise((resolve) => setTimeout(resolve, 50));
-        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);
-      }
-    },
-  );
-
   const logger = createTestLogger();
   const existingCwd = process.cwd();
 
@@ -4403,7 +4100,7 @@ describe("send_agent_prompt MCP tool", () => {
 
     const response = await tool.handler(parsed.data as Record<string, unknown>);
 
-    expect(spies.agentManager.subscribe).toHaveBeenCalled();
+    expect(spies.agentManager.subscribe).toHaveBeenCalledTimes(1);
     expect(spies.agentManager.waitForAgentEvent).not.toHaveBeenCalled();
     expect(response.structuredContent.guidance).toBe(
       "You will get notified when the prompted agent finishes, errors, or needs permission. Do not poll for status; continue with other work until the notification arrives.",
@@ -4573,7 +4270,7 @@ describe("send_agent_prompt MCP tool", () => {
     },
   );
 
-  it.each(["out_of_band", "out_of_band_interrupt", "startup_failure"] as const)(
+  it.each(["out_of_band", "startup_failure"] as const)(
     "preserves %s dispatch notification semantics",
     async (scenario) => {
       const workdir = await mkdtemp(join(tmpdir(), "mcp-dispatch-notification-"));
@@ -4609,11 +4306,10 @@ describe("send_agent_prompt MCP tool", () => {
           prompt: "Ongoing task",
           notifyOnFinish: false,
         });
-        if (scenario !== "startup_failure") {
+        if (scenario === "out_of_band") {
           const response = await invokeToolWithParsedInput(tool, {
             agentId: child.id,
             prompt: "/side-command",
-            activeTurnBehavior: scenario === "out_of_band_interrupt" ? "interrupt" : "steer",
           });
           expect(response.structuredContent).toMatchObject({
             success: true,

packages/server/src/server/agent/provider-subagents/store.test.ts

@@ -60,6 +60,23 @@ describe("ProviderSubagentStore", () => {
     expect(subagents.list("parent-b")).toHaveLength(1);
   });
 
+  test("keeps a poll liveness marker across presentation-only upserts", () => {
+    const subagents = new ProviderSubagentStore();
+    subagents.apply("parent-a", "pi", {
+      type: "upsert",
+      id: "child-1",
+      status: "running",
+      liveness: "poll",
+    });
+    subagents.apply("parent-a", "pi", { type: "upsert", id: "child-1", title: "scout" });
+
+    expect(subagents.get("parent-a", "child-1")).toMatchObject({
+      title: "scout",
+      status: "running",
+      liveness: "poll",
+    });
+  });
+
   test("limits oversized provider child tool output before storage", () => {
     const subagents = new ProviderSubagentStore();
     const output = "x".repeat(70 * 1024);

packages/server/src/server/agent/provider-subagents/store.ts

@@ -23,6 +23,11 @@ export interface ProviderSubagentDescriptor {
   toolCallId: string | null;
   cwd: string | null;
   subtitle: string | null;
+  /**
+   * Set when a live poll re-reports this row on every tick, so silence proves the poll is gone.
+   * Rows without it are event-driven and may stay quiet while their child still runs.
+   */
+  liveness?: "poll";
 }
 
 export type ProviderSubagentInputEvent =
@@ -40,6 +45,8 @@ export type ProviderSubagentInputEvent =
       cwd?: string | null;
       subtitle?: string | null;
       parentSubagentId?: string | null;
+      /** Only a poll-backed reporter may set this; it is what makes a stale row sweepable. */
+      liveness?: "poll";
       timestamp?: string;
     }
   | {
@@ -113,6 +120,7 @@ export class ProviderSubagentStore {
       this.timelines.initialize(key);
     }
     const timestamp = event.timestamp ?? new Date().toISOString();
+    const liveness = event.liveness ?? previous?.liveness;
     const subagent: ProviderSubagentDescriptor = {
       id: event.id,
       parentAgentId,
@@ -126,6 +134,7 @@ export class ProviderSubagentStore {
       cwd: stickyField(event.cwd, previous?.cwd),
       subtitle: stickyField(event.subtitle, previous?.subtitle),
       parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId),
+      ...(liveness === undefined ? {} : { liveness }),
     };
     this.descriptors.set(key, subagent);
     return { type: "upsert", subagent };

packages/server/src/server/agent/providers/pi/extensions/pi-subagents/fixtures/workflow.json

@@ -0,0 +1,153 @@
+{
+  "provenance": {
+    "package": "pi-subagents",
+    "version": "0.78.0",
+    "sourceCommit": "reconstructed-workflow-launch",
+    "piVersion": "1.2.0",
+    "captureDate": "2026-10-17"
+  },
+  "events": [
+    {
+      "type": "tool_execution_start",
+      "toolCallId": "call_wf1",
+      "toolName": "subagent",
+      "args": {
+        "workflow": "tmp/rigor-ui-test.js",
+        "async": true
+      }
+    },
+    {
+      "type": "tool_execution_end",
+      "toolCallId": "call_wf1",
+      "toolName": "subagent",
+      "result": {
+        "content": [
+          {
+            "type": "text",
+            "text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)"
+          }
+        ],
+        "details": {
+          "mode": "workflow",
+          "runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+          "results": [],
+          "asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+          "asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+          "workflowGraph": {
+            "lanes": [{ "key": "alpha", "agent": "scout" }]
+          },
+          "context": "fresh",
+          "timeoutMs": 1800000,
+          "deadlineAt": 1790199499998
+        }
+      },
+      "isError": false
+    },
+    {
+      "type": "message_end",
+      "message": {
+        "role": "custom",
+        "customType": "subagent-incremental-child-notify",
+        "content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running",
+        "display": false,
+        "details": {
+          "childKey": "alpha",
+          "outcome": "completed"
+        },
+        "timestamp": 1790197711046
+      }
+    },
+    {
+      "type": "message_end",
+      "message": {
+        "role": "custom",
+        "customType": "subagent-notify",
+        "content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl",
+        "display": false,
+        "timestamp": 1790197722046
+      }
+    }
+  ],
+  "messages": [
+    {
+      "role": "assistant",
+      "content": [
+        {
+          "type": "toolCall",
+          "id": "call_wf1",
+          "name": "subagent",
+          "arguments": {
+            "workflow": "tmp/rigor-ui-test.js",
+            "async": true
+          }
+        }
+      ],
+      "api": "openai-completions",
+      "provider": "openrouter",
+      "model": "google/gemini-3.6-flash",
+      "usage": {
+        "input": 3303,
+        "output": 114,
+        "cacheRead": 7053,
+        "cacheWrite": 0,
+        "reasoning": 81,
+        "totalTokens": 10470,
+        "cost": {
+          "input": 0.00247725,
+          "output": 0.00042750000000000004,
+          "cacheRead": 0.000528975,
+          "cacheWrite": 0,
+          "total": 0.003433725
+        }
+      },
+      "stopReason": "toolUse",
+      "timestamp": 1790197697608,
+      "responseId": "gen-1790197697-lxluaTrDkdaxjSXsO957",
+      "rawStopReason": "tool_calls"
+    },
+    {
+      "role": "toolResult",
+      "toolCallId": "call_wf1",
+      "toolName": "subagent",
+      "content": [
+        {
+          "type": "text",
+          "text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)"
+        }
+      ],
+      "details": {
+        "mode": "workflow",
+        "runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+        "results": [],
+        "asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+        "asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1",
+        "workflowGraph": {
+          "lanes": [{ "key": "alpha", "agent": "scout" }]
+        },
+        "context": "fresh",
+        "timeoutMs": 1800000,
+        "deadlineAt": 1790199499998
+      },
+      "isError": false,
+      "timestamp": 1790197700011
+    },
+    {
+      "role": "custom",
+      "customType": "subagent-incremental-child-notify",
+      "content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running",
+      "display": false,
+      "details": {
+        "childKey": "alpha",
+        "outcome": "completed"
+      },
+      "timestamp": 1790197711046
+    },
+    {
+      "role": "custom",
+      "customType": "subagent-notify",
+      "content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl",
+      "display": false,
+      "timestamp": 1790197722046
+    }
+  ]
+}

packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts

@@ -1,6 +1,6 @@
-import { describe, expect, test } from "vitest";
+import { describe, expect, test, vi } from "vitest";
 import { fileURLToPath } from "node:url";
-import { mkdtemp, writeFile, rm } from "node:fs/promises";
+import { mkdtemp, writeFile, appendFile, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { createPiExtensionHost } from "../index.js";
@@ -45,6 +45,18 @@ function hasCompletedChild(events: AgentStreamEvent[]) {
   );
 }
 
+function subagentUpserts(events: AgentStreamEvent[]) {
+  return events.flatMap((event) =>
+    event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
+  );
+}
+
+function reportedUpsert(events: AgentStreamEvent[], id: string, status?: string) {
+  return subagentUpserts(events).some(
+    (event) => event.id === id && (status === undefined || event.status === status),
+  );
+}
+
 describe("pi-subagents adapter", () => {
   test("follows async status and transcript without bg_wait", async () => {
     const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
@@ -273,4 +285,867 @@ describe("pi-subagents adapter", () => {
       },
     ]);
   });
+
+  test("maps a reconstructed workflowScript run to a live spawn row and its children", async () => {
+    const fixture = readSubagentFixture(new URL("./fixtures/workflow.json", import.meta.url));
+    const events = await verifySubagentFixture(fixture);
+    const upserts = events
+      .filter((event) => event.event.type === "upsert")
+      .map((event) => (event.event.type === "upsert" ? event.event : null));
+    expect(upserts.map((event) => [event?.id, event?.status])).toEqual([
+      ["call_wf1", "running"],
+      ["call_wf1", "running"],
+      ["call_wf1:alpha", "completed"],
+      ["call_wf1", "completed"],
+    ]);
+    // The workflow call carries no agent/task, so the script file names the run.
+    expect(upserts[0]?.title).toBe("rigor-ui-test.js");
+    expect(upserts[1]?.liveness).toBe("poll");
+    expect(upserts[2]?.title).toBeUndefined();
+  });
+
+  test("maps a reported workflowScript run to its reviewer row and saved timeline", async () => {
+    // Reported in #5837: `subagent({ workflow: true })` reaches the adapter as `workflowScript`,
+    // not `workflow`, so a launch check keyed on `workflow` alone drops the whole run.
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-script-"));
+    const host = createPiExtensionHost();
+    try {
+      const sessionFile = join(dir, "child.jsonl");
+      await writeFile(
+        sessionFile,
+        JSON.stringify({
+          type: "message",
+          message: { role: "assistant", content: [{ type: "text", text: "README summary" }] },
+        }) + "\n",
+      );
+      const args = {
+        workflowScript:
+          "return await runs.run('review', { agent: 'reviewer', task: 'Read README.md' });",
+        async: false,
+      };
+      expect(
+        host.mapToolCall({
+          callId: "reported-workflow",
+          toolName: "subagent",
+          args,
+          status: "running",
+          result: null,
+        })?.subagents,
+      ).toEqual([
+        expect.objectContaining({ id: "reported-workflow", title: "workflow", status: "running" }),
+      ]);
+      const output = host.mapToolCall({
+        callId: "reported-workflow",
+        toolName: "subagent",
+        args,
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "reported-run",
+            results: [{ index: 0, agent: "reviewer", exitCode: 0, sessionFile }],
+          },
+        },
+      });
+      expect(output?.subagents).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ type: "upsert", title: "reviewer", status: "completed" }),
+        ]),
+      );
+      expect(hasChildTimeline(await output!.hydration)).toBe(true);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("keys workflow child rows by their childKey while the run reports progress", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      await expect.poll(() => reportedUpsert(received, "call-wf:beta")).toBe(true);
+      expect(
+        subagentUpserts(received).map((event) => [event.id, event.status, event.liveness]),
+      ).toEqual(
+        expect.arrayContaining([
+          ["call-wf:alpha", "running", "poll"],
+          ["call-wf:beta", "running", "poll"],
+          ["call-wf", "running", "poll"],
+        ]),
+      );
+
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "complete",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "complete" },
+          ],
+        }),
+      );
+      await expect.poll(() => reportedUpsert(received, "call-wf", "completed")).toBe(true);
+      expect(
+        subagentUpserts(received).findLast((event) => event.id === "call-wf:alpha")?.status,
+      ).toBe("completed");
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("keeps a missing status directory's row alive until the miss budget expires", async () => {
+    vi.useFakeTimers();
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-lost",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "single",
+            runId: "run-1",
+            asyncId: "run-1",
+            asyncDir: "/nonexistent",
+            results: [],
+          },
+        },
+      });
+      await vi.advanceTimersByTimeAsync(250 * 5);
+      // The run is alive even while its file is missing: every miss tick re-reports the row, or a
+      // turn end inside the miss budget would cancel live work.
+      const duringMisses = subagentUpserts(received);
+      expect(duringMisses.length).toBeGreaterThanOrEqual(5);
+      expect(duringMisses.every((event) => event.status === "running")).toBe(true);
+      expect(duringMisses.every((event) => event.liveness === "poll")).toBe(true);
+
+      // 60 s of missing ticks, the budget before a status directory counts as gone.
+      await vi.advanceTimersByTimeAsync(250 * 236);
+      // The launch row comes back through the caller; polling reports the give-up exactly once.
+      const afterGiveUp = subagentUpserts(received);
+      expect(afterGiveUp.findLast((event) => event.id === "call-lost")).toEqual({
+        type: "upsert",
+        id: "call-lost",
+        status: "canceled",
+        liveness: "poll",
+      });
+      expect(afterGiveUp.filter((event) => event.status === "canceled")).toHaveLength(1);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+    }
+  });
+
+  test("a completion notice stops polling so a stale status file cannot revert the card", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stale-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({ state: "running", steps: [{ agent: "scout", status: "running" }] }),
+      );
+      const statuses = () =>
+        received.flatMap((event) =>
+          event.type === "provider_subagent" && event.event.type === "upsert"
+            ? [event.event.status]
+            : [],
+        );
+      await expect.poll(() => statuses().length).toBeGreaterThan(1);
+
+      host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-notify",
+        content: [
+          "Background task completed: **scout**",
+          "",
+          `Retention-managed async directory: ${dir}`,
+        ].join("\n"),
+      });
+      const settled = statuses().length;
+
+      // The run's directory is still on disk with a running status; polling must stay off.
+      await new Promise((resolve) => setTimeout(resolve, 400));
+      expect(statuses().length).toBe(settled);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a bg_wait result keeps its verdict while the run's status names no child yet", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-settled-owner-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      // A single unnamed child takes the owner's id, so this result settles the launch row while the
+      // run itself keeps polling.
+      expect(
+        host.mapToolCall({
+          callId: "wait-1",
+          toolName: "bg_wait",
+          args: { runId: "run-1" },
+          status: "completed",
+          result: {
+            details: {
+              mode: "single",
+              results: [],
+              completions: [
+                { runId: "run-1", results: [{ agent: "scout", success: true, exitCode: 0 }] },
+              ],
+            },
+          },
+        })?.subagents,
+      ).toEqual([expect.objectContaining({ id: "call-1", status: "completed" })]);
+
+      // The queue has not started the child yet, so the status file names no step: that tick must
+      // not re-report the owner as running over the verdict it already has.
+      await writeFile(join(dir, "status.json"), JSON.stringify({ state: "running", steps: [] }));
+      const beforeTick = received.length;
+      await vi.advanceTimersByTimeAsync(250 * 2);
+
+      expect(subagentUpserts(received.slice(beforeTick))).toEqual([]);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a paused run settles as canceled", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-paused-"));
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-notify",
+          content: [
+            "Background task paused: **scout**",
+            "",
+            `Retention-managed async directory: ${dir}`,
+          ].join("\n"),
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-1", status: "canceled" }]);
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a paused workflow child notice settles its row as canceled", () => {
+    const host = createPiExtensionHost();
+    host.mapToolCall({
+      callId: "call-wf",
+      toolName: "subagent",
+      args: { workflow: "tmp/rigor-ui-test.js", async: true },
+      status: "completed",
+      result: {
+        details: {
+          mode: "workflow",
+          runId: "run-wf",
+          asyncId: "run-wf",
+          asyncDir: "/nonexistent",
+          results: [],
+        },
+      },
+    });
+    expect(
+      host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-incremental-child-notify",
+        content: "Workflow child paused (needs attention): **alpha**\nWorkflow run: run-wf",
+      })?.subagents,
+    ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "canceled" }]);
+  });
+
+  test("partial and rejected runs settle every child instead of leaving a spinner", async () => {
+    for (const [state, expected] of [
+      ["partial", ["completed", "completed", "canceled"]],
+      ["rejected", ["completed", "completed", "failed"]],
+    ] as const) {
+      vi.useFakeTimers();
+      const dir = await mkdtemp(join(tmpdir(), "paseo-pi-partial-"));
+      const received: AgentStreamEvent[] = [];
+      const host = createPiExtensionHost();
+      try {
+        host.mapToolCall({
+          callId: "call-wf",
+          toolName: "subagent",
+          args: { workflow: "tmp/rigor-ui-test.js", async: true },
+          status: "completed",
+          result: {
+            details: {
+              mode: "workflow",
+              runId: "run-wf",
+              asyncId: "run-wf",
+              asyncDir: dir,
+              results: [],
+            },
+          },
+        });
+        await writeFile(
+          join(dir, "status.json"),
+          JSON.stringify({
+            state,
+            steps: [
+              { agent: "scout", workflowKey: "alpha", status: "complete" },
+              // WorkflowNodeStatus spells the same outcome `completed`.
+              { agent: "auditor", workflowKey: "gamma", status: "completed" },
+              {
+                agent: "reviewer",
+                workflowKey: "beta",
+                status: state === "partial" ? "running" : "failed",
+              },
+            ],
+          }),
+        );
+        host.follow((event) => received.push(event));
+        await vi.advanceTimersByTimeAsync(250);
+
+        expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
+          expect.arrayContaining([
+            ["call-wf:alpha", expected[0]],
+            ["call-wf:gamma", expected[1]],
+            ["call-wf:beta", expected[2]],
+            ["call-wf", "canceled"],
+          ]),
+        );
+        expect(
+          subagentUpserts(received).filter((event) => event.status === "running"),
+        ).toHaveLength(0);
+
+        // The run is terminal, so the poll stops instead of re-reporting fresh rows.
+        const settled = received.length;
+        await vi.advanceTimersByTimeAsync(250 * 4);
+        expect(received.length).toBe(settled);
+      } finally {
+        host.close();
+        vi.useRealTimers();
+        await rm(dir, { recursive: true, force: true });
+      }
+    }
+  });
+
+  test("a child notice keeps its status while the run is still reporting", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-pin-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      // The stale step entry still says running; the child's notice is its verdict.
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      host.follow((event) => received.push(event));
+      await vi.advanceTimersByTimeAsync(250);
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      const beforeNotice = received.length;
+      await vi.advanceTimersByTimeAsync(250 * 3);
+      expect(
+        subagentUpserts(received.slice(beforeNotice)).filter(
+          (event) => event.id === "call-wf:alpha",
+        ),
+      ).toEqual([]);
+      // The rest of the run keeps re-reporting, so this is not a poll that simply stopped.
+      expect(reportedUpsert(received.slice(beforeNotice), "call-wf:beta")).toBe(true);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a child notice's verdict survives the run's terminal tick", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-terminal-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      host.follow((event) => received.push(event));
+      // The child's notice settles it before the run itself ends.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      // The run then ends without a verdict for alpha: its step entry still says running, which the
+      // terminal tick would read as canceled if the notice were not the child's verdict.
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "stopped",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "running" },
+            { agent: "reviewer", workflowKey: "beta", status: "stopped" },
+          ],
+        }),
+      );
+      await vi.advanceTimersByTimeAsync(250);
+
+      expect(
+        subagentUpserts(received).filter((event) => event.id === "call-wf:alpha" && event.status),
+      ).toEqual([expect.objectContaining({ id: "call-wf:alpha", status: "completed" })]);
+      // The child that never reported a verdict of its own is still settled by that tick.
+      expect(
+        subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
+      ).toBe("canceled");
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("gives up without rewriting a child that already settled", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-giveup-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      host.follow((event) => received.push(event));
+      await vi.advanceTimersByTimeAsync(250);
+      expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
+
+      // The workflow keeps running, so polling continues after the child's own notice.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      await rm(join(dir, "status.json"));
+      await vi.advanceTimersByTimeAsync(250 * 241);
+
+      expect(
+        subagentUpserts(received).some(
+          (event) => event.id === "call-wf:alpha" && event.status === "canceled",
+        ),
+      ).toBe(false);
+      expect(
+        subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
+      ).toBe("canceled");
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a run's final notice cancels the children it stopped with", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stopped-children-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "running" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      host.follow((event) => received.push(event));
+      await vi.advanceTimersByTimeAsync(250);
+      expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
+
+      // alpha reports its own verdict before the run stops.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      // The run then stops with beta unfinished. The notice ends the poll, so no later tick will
+      // ever report beta: it settles as canceled, never completed. alpha keeps its own verdict.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-notify",
+          content: [
+            "Background task stopped: **rigor-ui-test.js**",
+            "",
+            `Retention-managed async directory: ${dir}`,
+          ].join("\n"),
+        })?.subagents,
+      ).toEqual([
+        { type: "upsert", id: "call-wf", status: "canceled" },
+        { type: "upsert", id: "call-wf:beta", status: "canceled" },
+      ]);
+
+      // Polling is off; the status file still says running and must not bring the rows back.
+      const settled = received.length;
+      await vi.advanceTimersByTimeAsync(250 * 3);
+      expect(received.length).toBe(settled);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("names a child a notice settled before the first poll saw it", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-fast-child-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      // A fast child reports before any poll tick, so its row is born without a name.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [{ agent: "scout", workflowKey: "alpha", status: "running" }],
+        }),
+      );
+      host.follow((event) => received.push(event));
+      await vi.advanceTimersByTimeAsync(250);
+
+      // The poll leaves the settled status alone but still fills the name status.json carries.
+      const alphaUpserts = () =>
+        subagentUpserts(received).filter((event) => event.id === "call-wf:alpha");
+      expect(alphaUpserts()).toEqual([
+        {
+          type: "upsert",
+          id: "call-wf:alpha",
+          title: "scout",
+          status: "completed",
+          liveness: "poll",
+        },
+      ]);
+
+      // Reported once, not on every tick.
+      await vi.advanceTimersByTimeAsync(250 * 2);
+      expect(alphaUpserts()).toHaveLength(1);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a workflow that reported rows replaces its placeholder with them", () => {
+    const host = createPiExtensionHost();
+    const args = { workflow: true };
+    // The in-flight placeholder is poll-marked, so a turn end can settle it if its call never
+    // reports back.
+    expect(
+      host.mapToolCall({
+        callId: "fanout",
+        toolName: "subagent",
+        args,
+        status: "running",
+        result: null,
+      })?.subagents,
+    ).toEqual([expect.objectContaining({ id: "fanout", status: "running", liveness: "poll" })]);
+    expect(
+      host.mapToolCall({
+        callId: "fanout",
+        toolName: "subagent",
+        args,
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            results: [
+              { index: 0, agent: "scout", workflowKey: "alpha", success: true, exitCode: 0 },
+              { index: 1, agent: "reviewer", workflowKey: "beta", success: true, exitCode: 0 },
+            ],
+          },
+        },
+      })?.subagents,
+    ).toEqual([
+      { type: "remove", id: "fanout" },
+      expect.objectContaining({ id: "fanout:alpha", title: "scout", status: "completed" }),
+      expect.objectContaining({ id: "fanout:beta", title: "reviewer", status: "completed" }),
+    ]);
+  });
+  test("replays keyed async workflow child transcripts without a live poll", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-replay-"));
+    const host = createPiExtensionHost();
+    try {
+      const sessionFile = join(dir, "child.jsonl");
+      await writeFile(
+        sessionFile,
+        JSON.stringify({
+          type: "message",
+          message: {
+            role: "assistant",
+            content: [{ type: "text", text: "Saved reviewer answer" }],
+          },
+        }) + "\n",
+      );
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "complete",
+          steps: [{ agent: "reviewer", workflowKey: "review", status: "complete", sessionFile }],
+        }),
+      );
+      const output = host.mapToolCall({
+        callId: "replayed-workflow",
+        toolName: "subagent",
+        args: { workflowScript: "return await runs.all(items);", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "saved-run",
+            asyncId: "saved-run",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      const events = await output!.hydration;
+      expect(
+        events.some(
+          (event) =>
+            event.type === "provider_subagent" &&
+            event.event.type === "timeline" &&
+            event.event.id === "replayed-workflow:review",
+        ),
+      ).toBe(true);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("stops following a child whose completion arrived before its transcript was attached", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-finished-reader-"));
+    const host = createPiExtensionHost();
+    const received: AgentStreamEvent[] = [];
+    try {
+      const sessionFile = join(dir, "child.jsonl");
+      const message = (text: string) =>
+        JSON.stringify({
+          type: "message",
+          message: {
+            role: "assistant",
+            content: [{ type: "text", text }],
+          },
+        }) + "\n";
+      await writeFile(sessionFile, message("Finished child"));
+      host.follow((event) => received.push(event));
+      const spawn = host.mapToolCall({
+        callId: "fast-workflow",
+        toolName: "subagent",
+        args: { workflow: true, async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "fast-run",
+            asyncId: "fast-run",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      await spawn!.hydration;
+      const notice = host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-incremental-child-notify",
+        content: "Workflow child completed: **review**\nWorkflow run: fast-run",
+      });
+      await notice!.hydration;
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [{ agent: "reviewer", workflowKey: "review", status: "running", sessionFile }],
+        }),
+      );
+      await expect.poll(() => hasChildTimeline(received)).toBe(true);
+      const timelineCount = () =>
+        received.filter(
+          (event) => event.type === "provider_subagent" && event.event.type === "timeline",
+        ).length;
+      const count = timelineCount();
+      await appendFile(sessionFile, message("Later content must not be followed"));
+      await new Promise((resolve) => setTimeout(resolve, 600));
+      expect(timelineCount()).toBe(count);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
 });

packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts

@@ -2,12 +2,20 @@ import { z } from "zod";
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import { extractTextFromToolResult } from "../../tool-call-mapper.js";
+import type { PiAgentMessage } from "../../rpc-types.js";
+import type { ProviderSubagentInputEvent } from "../../../../provider-subagents/store.js";
 import type { PiExtension, PiExtensionToolCall, PiExtensionToolMapping } from "../contract.js";
 
+type CustomMessage = Extract<PiAgentMessage, { role: "custom" }>;
+type SubagentUpsert = Extract<ProviderSubagentInputEvent, { type: "upsert" }>;
+
 const Args = z
   .object({
     agent: z.string().trim().min(1).optional(),
     task: z.string().trim().min(1).optional(),
+    workflow: z.union([z.string().trim().min(1), z.literal(true)]).optional(),
+    // `subagent({ workflow: true })` launches the reply's script; Pi delivers it as this field.
+    workflowScript: z.string().trim().min(1).optional(),
     async: z.boolean().optional(),
     action: z.unknown().optional(),
   })
@@ -16,6 +24,7 @@ const Row = z
   .object({
     index: z.number().int().nonnegative().optional(),
     agent: z.string().trim().min(1),
+    workflowKey: z.string().trim().min(1).optional(),
     exitCode: z.number().optional(),
     success: z.boolean().optional(),
     sessionFile: z.string().trim().min(1).optional(),
@@ -41,6 +50,7 @@ const AsyncStatus = z
       z
         .object({
           agent: z.string().trim().min(1),
+          workflowKey: z.string().trim().min(1).optional(),
           status: z.string().optional(),
           sessionFile: z.string().trim().min(1).optional(),
         })
@@ -49,21 +59,116 @@ const AsyncStatus = z
   })
   .passthrough();
 
+/**
+ * Run states that will not change again: `paused` waits on the user, and `partial`/`rejected`
+ * still end the run even when some steps never produced a result of their own.
+ */
+const TERMINAL_RUN_STATES = new Set([
+  "complete",
+  "failed",
+  "stopped",
+  "paused",
+  "partial",
+  "rejected",
+]);
+/** The run's status file, or undefined while it is missing or not yet readable. */
+function readAsyncStatus(dir: string): z.infer<typeof AsyncStatus> | undefined {
+  try {
+    const parsed = AsyncStatus.safeParse(
+      JSON.parse(readFileSync(join(dir, "status.json"), "utf8")),
+    );
+    return parsed.success ? parsed.data : undefined;
+  } catch {
+    return undefined;
+  }
+}
+
+/** One async run the poll follows, keyed by its async directory. */
+interface AsyncRun {
+  owner: string;
+  polling: boolean;
+  misses: number;
+  deadline: number;
+  rows: Set<string>;
+}
+
+/**
+ * Missing ticks before a status directory counts as gone: 60 s at the 250 ms follow cadence. A live
+ * run can briefly lose its file while the concurrency queue starts it or retention rotates it.
+ */
+const POLL_MISS_LIMIT = 240;
+/**
+ * A run that never reaches a terminal state in its status file must not keep a card spinning
+ * forever. pi-subagents' own default run timeout is 30 minutes; this is the backstop.
+ */
+const POLL_TTL_MS = 6 * 60 * 60 * 1000;
+
 function asyncChildStatus(runState: string, stepStatus?: string) {
-  if (!["complete", "failed", "stopped"].includes(runState)) return "running";
-  if (runState === "stopped" || stepStatus === "stopped") return "canceled";
+  if (!TERMINAL_RUN_STATES.has(runState)) return "running";
+  // A partial or rejected run says nothing about its individual children: read their own outcome,
+  // and treat a step that never reached one as canceled rather than still running.
+  if (runState === "partial" || runState === "rejected") {
+    // Two vocabularies: status.json steps say `complete`, WorkflowNodeStatus says `completed`.
+    if (stepStatus === "complete" || stepStatus === "completed") return "completed";
+    if (stepStatus === "failed" || stepStatus === "error") return "failed";
+    return "canceled";
+  }
+  if (runState === "stopped" || runState === "paused" || stepStatus === "stopped")
+    return "canceled";
   if (runState === "failed" || stepStatus === "failed" || stepStatus === "error") return "failed";
   return "completed";
 }
 
-/** The `subagent-notify` header; a grouped notice reports every run as completed. */
+/** The status of the run itself, used for the row a launch opened before its children reported. */
+function asyncRunStatus(runState: string) {
+  if (runState === "complete") return "completed" as const;
+  if (runState === "failed") return "failed" as const;
+  return "canceled" as const;
+}
+
+/**
+ * The `subagent-notify` / `subagent-incremental-child-notify` header. A grouped notice reports
+ * every run as completed; a paused run is waiting on the user, which the track shows as canceled.
+ */
 function noticeStatus(header: string) {
   if (/^Background tasks completed \(\d+\):/.test(header)) return "completed";
   const outcome = header.match(
-    /^(?:Background task|Detached foreground task) (completed|failed|stopped): \*\*/,
+    /^(?:Background task|Detached foreground task|Workflow child) (completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*/,
   )?.[1];
-  if (outcome === "stopped") return "canceled";
-  return outcome as "completed" | "failed" | undefined;
+  if (!outcome) return undefined;
+  return outcome === "paused" || outcome === "stopped"
+    ? ("canceled" as const)
+    : (outcome as "completed" | "failed");
+}
+
+/**
+ * A workflow launch carries no `agent`/`task`: the children are declared by the script. Its label
+ * is the script file name, or the bare `workflow` of the reply-block form.
+ */
+function workflowLabel(workflow: string | true | undefined) {
+  if (workflow === undefined) return undefined;
+  if (workflow === true) return "workflow";
+  const segments = workflow.split(/[\\/]/).filter(Boolean);
+  return segments[segments.length - 1] ?? workflow;
+}
+
+/**
+ * A workflow launch carries no agent/task; its children are declared by the script. A reply-scoped
+ * script arrives as `workflowScript` instead of a `workflow` path, so it takes the generic label.
+ */
+function workflowTitle(args: { workflow?: string | true; workflowScript?: string }) {
+  const label = workflowLabel(args.workflow);
+  if (label !== undefined) return label;
+  return args.workflowScript === undefined ? undefined : "workflow";
+}
+
+/**
+ * A row keeps one id from launch to settle: the owner alone for a single unnamed child, otherwise
+ * `owner:key`. Workflow children carry their `workflowKey`, so progress updates, child notices and
+ * final results all land on the same row.
+ */
+function rowId(owner: string, count: number, key: string | undefined, index: number) {
+  return count === 1 && !key ? owner : `${owner}:${key ?? index}`;
 }
 
 export const piSubagents: PiExtension = {
@@ -71,13 +176,64 @@ export const piSubagents: PiExtension = {
   createSession: () => {
     const callsByRun = new Map<string, string>();
     const readSessions = new Set<string>();
+    /** Rows already reported terminal; a late give-up must not rewrite one of them. */
+    const settledRows = new Map<string, NonNullable<SubagentUpsert["status"]>>();
+    /** Rows whose poll already reported the step's agent. A settled row is still named once. */
+    const namedRows = new Set<string>();
+    const recordTerminal = (id: string, status: NonNullable<SubagentUpsert["status"]>) => {
+      if (status !== "running") settledRows.set(id, status);
+    };
     /** Keyed by async directory, which both status.json and the completion notice carry. */
-    const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
+    const asyncRuns = new Map<string, AsyncRun>();
     const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
       if (details.asyncDir && call.status !== "failed") {
-        asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true });
+        asyncRuns.set(details.asyncDir, {
+          owner: call.callId,
+          polling: true,
+          misses: 0,
+          deadline: Date.now() + POLL_TTL_MS,
+          rows: new Set<string>(),
+        });
       }
     };
+    /**
+     * The run is over and nothing will ever update its rows again, so settle them rather than leave a
+     * spinner. A row that already reported a terminal status keeps it: this is not its verdict.
+     */
+    const giveUpRun = (run: AsyncRun) => {
+      run.polling = false;
+      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
+      for (const id of new Set([run.owner, ...run.rows])) {
+        if (settledRows.has(id)) continue;
+        subagents.push({ type: "upsert", id, status: "canceled", liveness: "poll" });
+      }
+      return subagents;
+    };
+    /**
+     * The notice is the run's last word: it ends the poll, settles the launch row, and cancels the
+     * children the run stopped with. No later tick will report those, and nothing here proves they
+     * succeeded, so they settle as canceled. A child that already reported its own verdict keeps it.
+     */
+    const settleRunNotice = (run: AsyncRun, status: "completed" | "failed" | "canceled") => {
+      // A stale status.json must not revert this card either.
+      run.polling = false;
+      recordTerminal(run.owner, status);
+      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [
+        { type: "upsert", id: run.owner, status },
+      ];
+      for (const id of run.rows) {
+        if (settledRows.has(id)) continue;
+        subagents.push({ type: "upsert", id, status: "canceled" });
+      }
+      return subagents;
+    };
+    /**
+     * A row id can settle twice: a completion notice records a verdict in `settledRows`, and a
+     * `subagent_wait` result repeats the run's rows with the run's own verdict. Both are terminal, so
+     * a disagreement moves a row terminal-to-terminal (completed → failed) and never resurrects it;
+     * the owner row is what carries the run outcome.
+     */
+    // ponytail: child verdicts can diverge between notice and bg_wait result — unify when one source of truth exists
     const collectRows = (
       owner: string,
       rows: z.infer<typeof Row>[],
@@ -87,17 +243,19 @@ export const piSubagents: PiExtension = {
       const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
       const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
       for (const [index, row] of rows.entries()) {
-        const id = rows.length === 1 ? owner : `${owner}:${row.index ?? index}`;
+        const id = rowId(owner, rows.length, row.workflowKey, row.index ?? index);
+        const status =
+          failed || row.success === false || (row.success !== true && row.exitCode !== 0)
+            ? ("failed" as const)
+            : ("completed" as const);
+        recordTerminal(id, status);
         subagents.push({
           type: "upsert",
           id,
           title: row.agent,
           description,
           toolCallId: owner,
-          status:
-            failed || row.success === false || (row.success !== true && row.exitCode !== 0)
-              ? "failed"
-              : "completed",
+          status,
         });
         if (row.sessionFile && !readSessions.has(row.sessionFile)) {
           readSessions.add(row.sessionFile);
@@ -123,66 +281,169 @@ export const piSubagents: PiExtension = {
       }
       return subagents.length ? { subagents, childSessions } : undefined;
     };
-    const mapSpawn = (call: PiExtensionToolCall) => {
-      const args = Args.safeParse(call.args);
-      if (!args.success || args.data.action !== undefined || !args.data.agent || !args.data.task)
-        return undefined;
-      const detail = {
-        type: "sub_agent" as const,
-        subAgentType: args.data.agent,
-        description: args.data.task,
-        log: extractTextFromToolResult(call.result)?.trim() ?? "",
-      };
-      const base = {
-        type: "upsert" as const,
-        id: call.callId,
-        title: args.data.agent,
-        description: args.data.task,
-        toolCallId: call.callId,
-      };
-      if (call.status === "running")
-        return { detail, subagents: [{ ...base, status: "running" as const }] };
+    /** Project a saved status both when replaying a launch and when its live poll advances. */
+    const projectAsyncStatus = (run: AsyncRun, status: z.infer<typeof AsyncStatus>) => {
+      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
+      const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
+      const runTerminal = TERMINAL_RUN_STATES.has(status.state);
+      for (const [index, step] of status.steps.entries()) {
+        const id = rowId(run.owner, status.steps.length, step.workflowKey, index);
+        run.rows.add(id);
+        // A child's notice supplies its verdict even when its saved status is stale. Reuse that
+        // verdict when attaching a transcript or filling its name after the notice arrived.
+        const settledStatus = settledRows.get(id);
+        const childSession =
+          step.sessionFile && !readSessions.has(step.sessionFile)
+            ? { id, file: step.sessionFile }
+            : undefined;
+        if (!settledStatus) {
+          const childStatus = asyncChildStatus(status.state, step.status);
+          recordTerminal(id, childStatus);
+          namedRows.add(id);
+          subagents.push({
+            type: "upsert",
+            id,
+            title: step.agent,
+            status: childStatus,
+            liveness: "poll" as const,
+          });
+        } else if (!namedRows.has(id) || childSession) {
+          // Attach a late transcript with its known verdict, so the host reads it once instead of
+          // starting a follower that no later terminal update will release.
+          namedRows.add(id);
+          subagents.push({
+            type: "upsert",
+            id,
+            title: step.agent,
+            status: settledStatus,
+            liveness: "poll" as const,
+          });
+        }
+        if (childSession) {
+          readSessions.add(childSession.file);
+          childSessions.push(childSession);
+        }
+      }
+      if (runTerminal) run.polling = false;
+      // The launch row stands for the whole run, and a single unnamed child already took its
+      // id. Re-reporting it every tick is also what proves to the manager that this run is alive.
+      // A row that already settled keeps that verdict: a result that landed before the run's
+      // steps named its children must not be reverted to running by the next tick.
+      if (!run.rows.has(run.owner) && !settledRows.has(run.owner)) {
+        const ownerStatus = run.polling ? ("running" as const) : asyncRunStatus(status.state);
+        recordTerminal(run.owner, ownerStatus);
+        subagents.push({
+          type: "upsert",
+          id: run.owner,
+          status: ownerStatus,
+          liveness: "poll",
+        });
+      }
+      return { subagents, childSessions };
+    };
+    /** A settled spawn: Pi either reports the rows it ran, or an async directory the poll follows. */
+    const mapSettledSpawn = (
+      call: PiExtensionToolCall,
+      base: SubagentUpsert,
+      description: string | undefined,
+      asyncRequested: boolean,
+    ): Pick<PiExtensionToolMapping, "subagents" | "childSessions"> => {
+      const failed = call.status === "failed";
       const details = Details.safeParse(
         typeof call.result === "object" ? call.result?.details : null,
       );
-      if (!details.success || details.data.mode === "management")
-        return call.status === "failed"
-          ? { detail, subagents: [{ ...base, status: "failed" as const }] }
-          : { detail };
+      if (!details.success || details.data.mode === "management") {
+        if (!failed) return { subagents: [] };
+        recordTerminal(call.callId, "failed");
+        return { subagents: [{ ...base, status: "failed" as const }] };
+      }
       if (details.data.runId) callsByRun.set(details.data.runId, call.callId);
-      if (details.data.results.length === 0 && (details.data.asyncId || args.data.async)) {
+      if (details.data.results.length === 0) {
+        if (failed) {
+          recordTerminal(call.callId, "failed");
+          return { subagents: [{ ...base, status: "failed" as const }] };
+        }
+        if (!details.data.asyncId && !asyncRequested) {
+          // ponytail: inferred completion — revisit if pi can return success while a child outlives the call.
+          recordTerminal(call.callId, "completed");
+          return { subagents: [{ ...base, status: "completed" as const }] };
+        }
+        // The placeholder row is now backed by a per-tick poll, which is what proves it alive.
         rememberAsyncRun(details.data, call);
+        const dir = details.data.asyncDir;
+        const run = dir ? asyncRuns.get(dir) : undefined;
+        const status = dir ? readAsyncStatus(dir) : undefined;
+        if (run && status) return projectAsyncStatus(run, status);
+        return { subagents: [{ ...base, status: "running" as const, liveness: "poll" as const }] };
+      }
+      const collected = collectRows(call.callId, details.data.results, description, failed);
+      // A run that reported rows of its own never keeps the placeholder card it opened at launch.
+      if (!collected.subagents.some((event) => event.id === call.callId)) {
+        collected.subagents.unshift({ type: "remove", id: call.callId });
+      }
+      return collected;
+    };
+    const mapSpawn = (call: PiExtensionToolCall) => {
+      const args = Args.safeParse(call.args);
+      if (!args.success || args.data.action !== undefined) return undefined;
+      const single = Boolean(args.data.agent && args.data.task);
+      let workflow: string | undefined;
+      if (!single) workflow = workflowTitle(args.data);
+      if (!single && workflow === undefined) return undefined;
+      const base = {
+        type: "upsert" as const,
+        id: call.callId,
+        title: (single ? args.data.agent : workflow) ?? "workflow",
+        description: args.data.task,
+        toolCallId: call.callId,
+      };
+      const detail = {
+        type: "sub_agent" as const,
+        subAgentType: base.title,
+        description: args.data.task,
+        log: extractTextFromToolResult(call.result)?.trim() ?? "",
+      };
+      // The in-flight placeholder is poll-marked: a turn interrupted before this call reports
+      // never delivers a terminal mapping, and only a marked row can be settled by the sweep.
+      // ponytail: replay re-emits running before the settled rows land — one-pass client flip, upgrade when replay maps calls with their results first
+      if (call.status === "running")
         return {
           detail,
-          subagents: [
-            {
-              ...base,
-              status: call.status === "failed" ? ("failed" as const) : ("running" as const),
-            },
-          ],
+          subagents: [{ ...base, status: "running" as const, liveness: "poll" as const }],
         };
-      }
-      if (details.data.results.length === 0 && call.status === "failed")
-        return { detail, subagents: [{ ...base, status: "failed" as const }] };
       return {
         detail,
-        ...collectRows(call.callId, details.data.results, args.data.task, call.status === "failed"),
+        ...mapSettledSpawn(call, base, args.data.task, Boolean(args.data.async)),
       };
     };
+    const noticeText = (message: CustomMessage) =>
+      typeof message.content === "string"
+        ? message.content
+        : extractTextFromToolResult({ content: message.content });
+    const mapWorkflowChildNotice = (message: CustomMessage) => {
+      const lines = noticeText(message)?.split("\n") ?? [];
+      const status = noticeStatus(lines[0] ?? "");
+      const key = lines[0]?.match(
+        /^Workflow child (?:completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*(.+?)\*\*/,
+      )?.[1];
+      const runId = lines.find((line) => line.startsWith("Workflow run: "))?.slice(14);
+      const owner = runId ? callsByRun.get(runId) : undefined;
+      if (!status || !key || !owner) return undefined;
+      const id = `${owner}:${key}`;
+      recordTerminal(id, status);
+      return { subagents: [{ type: "upsert" as const, id, status }] };
+    };
     return {
       mapToolCall(call) {
         if (call.toolName === "bg_wait") return mapWait(call);
         if (call.toolName === "subagent") return mapSpawn(call);
         return undefined;
       },
-      // The notice is the only completion Pi persists, so replay settles async runs from it.
       mapCustomMessage(message) {
+        if (message.customType === "subagent-incremental-child-notify")
+          return mapWorkflowChildNotice(message);
         if (message.customType !== "subagent-notify") return undefined;
-        const text =
-          typeof message.content === "string"
-            ? message.content
-            : extractTextFromToolResult({ content: message.content });
-        const lines = text?.split("\n") ?? [];
+        const lines = noticeText(message)?.split("\n") ?? [];
         const status = noticeStatus(lines[0] ?? "");
         if (!status) return undefined;
         const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
@@ -192,8 +453,10 @@ export const piSubagents: PiExtension = {
         for (const line of lines) {
           const dir = line.match(/^Retention-managed async directory: (.+)$/)?.[1];
           if (dir) {
-            owner = asyncRuns.get(dir)?.owner;
-            if (owner) subagents.push({ type: "upsert", id: owner, status });
+            const run = asyncRuns.get(dir);
+            owner = run?.owner;
+            if (!run || !owner) continue;
+            subagents.push(...settleRunNotice(run, status));
             continue;
           }
           const file = owner ? line.match(/^Session file: (.+)$/)?.[1] : undefined;
@@ -209,36 +472,38 @@ export const piSubagents: PiExtension = {
       poll() {
         const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
         const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
+        const now = Date.now();
         for (const [dir, run] of asyncRuns) {
           if (!run.polling) continue;
-          let raw: unknown;
-          try {
-            raw = JSON.parse(readFileSync(join(dir, "status.json"), "utf8"));
-          } catch {
+          if (now >= run.deadline) {
+            subagents.push(...giveUpRun(run));
             continue;
           }
-          const parsed = AsyncStatus.safeParse(raw);
-          if (!parsed.success) continue;
-          const terminal = ["complete", "failed", "stopped"].includes(parsed.data.state);
-          for (const [index, step] of parsed.data.steps.entries()) {
-            const id = parsed.data.steps.length === 1 ? run.owner : `${run.owner}:${index}`;
-            const status = asyncChildStatus(parsed.data.state, step.status);
-            subagents.push({ type: "upsert", id, title: step.agent, status });
-            if (step.sessionFile && !readSessions.has(step.sessionFile)) {
-              readSessions.add(step.sessionFile);
-              childSessions.push({ id, file: step.sessionFile });
+          const status = readAsyncStatus(dir);
+          if (!status) {
+            run.misses += 1;
+            if (run.misses >= POLL_MISS_LIMIT) {
+              subagents.push(...giveUpRun(run));
+              continue;
             }
-          }
-          if (terminal) {
-            if (parsed.data.steps.length === 0) {
+            // The run is still alive even though its file is gone for now. Keep re-reporting its
+            // unsettled rows: silence would freeze their timestamp and a turn end would cancel a
+            // live run inside the miss budget.
+            for (const id of new Set([run.owner, ...run.rows])) {
+              if (settledRows.has(id)) continue;
               subagents.push({
                 type: "upsert",
-                id: run.owner,
-                status: parsed.data.state === "complete" ? "completed" : "failed",
+                id,
+                status: "running" as const,
+                liveness: "poll" as const,
               });
             }
-            run.polling = false;
+            continue;
           }
+          run.misses = 0;
+          const output = projectAsyncStatus(run, status);
+          subagents.push(...output.subagents);
+          childSessions.push(...output.childSessions);
         }
         return subagents.length || childSessions.length ? { subagents, childSessions } : undefined;
       },

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

@@ -6,7 +6,7 @@ import type { Logger } from "pino";
 
 import type { AgentMode, AgentProvider, AgentSessionConfig } from "../agent-sdk-types.js";
 import type { AgentManager } from "../agent-manager.js";
-import { ActiveTurnBehaviorSchema, AgentProfileSchema } from "@getpaseo/protocol/messages";
+import { AgentProfileSchema } from "@getpaseo/protocol/messages";
 import type { DaemonConfigStore } from "../../daemon-config-store.js";
 import {
   AgentFeatureSchema,
@@ -1209,9 +1209,6 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
     .object(legacyTopLevelCreateAgentInputSchema)
     .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.',
-    ),
     agentId: z.string(),
     prompt: z.string(),
     sessionMode: z.string().optional().describe("Optional mode to set before running the prompt."),
@@ -1958,55 +1955,28 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
       inputSchema: sendAgentPromptInputSchema,
       outputSchema: {
         success: z.boolean(),
-        disposition: z.enum(["turn_started", "out_of_band", "steered"]).optional(),
+        disposition: z.enum(["turn_started", "out_of_band"]).optional(),
         status: AgentStatusEnum,
         guidance: z.string().optional(),
       },
     },
-    async ({
-      agentId,
-      prompt,
-      sessionMode,
-      notifyOnFinish = Boolean(callerAgentId),
-      activeTurnBehavior = callerAgentId ? "steer" : "interrupt",
-    }) => {
+    async ({ agentId, prompt, sessionMode, notifyOnFinish = Boolean(callerAgentId) }) => {
       let stopNotification: (() => void) | undefined;
-      let acceptDispatch!: (result: PromptDispatchDisposition | false) => void;
-      const dispatchResult = new Promise<PromptDispatchDisposition | false>((resolve) => {
-        acceptDispatch = resolve;
-      });
-      const notifying = Boolean(callerAgentId && notifyOnFinish);
-      if (callerAgentId && notifyOnFinish) {
-        stopNotification = setupFinishNotification({
-          agentManager,
-          agentStorage,
-          childAgentId: agentId,
-          callerAgentId,
-          waitForTurnStart: false,
-          dispatchResult,
-          logger: childLogger,
-        });
-      }
-      function confirmDispatch(disposition: PromptDispatchDisposition): void {
-        // A new turn must ignore outcomes from the displaced turn. Its iterator
-        // has not started yet, so rearm here; steering keeps the early observer.
-        if (disposition !== "turn_started") {
-          acceptDispatch(disposition);
+      function armFinishNotification(disposition: PromptDispatchDisposition): void {
+        if (!callerAgentId || !notifyOnFinish) {
           return;
         }
-        acceptDispatch(disposition);
-        if (!callerAgentId || !notifyOnFinish) return;
-        stopNotification?.();
         stopNotification = setupFinishNotification({
           agentManager,
           agentStorage,
           childAgentId: agentId,
           callerAgentId,
-          waitForTurnStart: true,
+          waitForTurnStart: disposition === "turn_started",
           logger: childLogger,
         });
       }
 
+      const notifying = Boolean(callerAgentId && notifyOnFinish);
       const { disposition } = await sendPromptToAgent({
         agentManager,
         agentStorage,
@@ -2014,11 +1984,9 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
         prompt,
         source: callerAgentId ? { kind: "agent-message", agentId: callerAgentId } : undefined,
         sessionMode,
-        activeTurnBehavior,
-        onDispatch: confirmDispatch,
+        onDispatch: armFinishNotification,
         logger: childLogger,
       }).catch((error) => {
-        acceptDispatch(false);
         stopNotification?.();
         if (error instanceof AgentNotFoundError) {
           throw new PaseoToolInputError(error.message);

public-docs/cli.md

@@ -289,12 +289,9 @@ Use the recipient's agent ID from `paseo ls`, or [copy it from the agent's tab](
 ```bash
 paseo send <id> "now run the tests"
 paseo send <id> --image screenshot.png "what's wrong here?"
-paseo send <id> --no-wait "run this task"
-paseo send <id> --behavior interrupt "stop and do this instead"
+paseo send <id> --no-wait "queue this task"
 ```
 
-Agent-originated sends (`PASEO_AGENT_ID` set) default to steering the running turn. Other sends default to interrupting it. Use `--behavior steer` or `--behavior interrupt` to override the default. Steering uses the same fallback as the client: if steering is unavailable, the running turn is interrupted and replaced.
-
 ## Viewing logs
 
 ```bash

public-docs/orchestration-workflows.md

@@ -51,7 +51,7 @@ Agents can prompt each other by **agent ID**, the identifier Paseo uses to addre
 
 > Use Paseo to send this prompt to agent [paste agent ID here]: “Review the parser changes in your workspace and report any missing test cases.”
 
-The receiving agent gets the prompt in its existing conversation. Agent-originated sends default to steering a running turn, with interruption fallback when steering is unavailable. Set MCP `activeTurnBehavior` or CLI `--behavior` to `steer` or `interrupt` to override that default. To have it send a separate message back, give it the sender's agent ID too.
+The receiving agent gets the prompt in its existing conversation. To have it send a separate message back, give it the sender's agent ID too.
 
 Underneath, the agent calls `send_agent_prompt` with the recipient's `agentId` and a `prompt`. Agents can also discover IDs with `list_agents`; the CLI uses [`paseo send <id>`](/docs/cli#sending-messages). For another host, use the [remote CLI workflow](#work-on-another-machine).
 

skills/paseo/SKILL.md

@@ -63,7 +63,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` only for truly fire-and-forget agents.
 
-**`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`. Returns after dispatch acknowledgement. Agent-scoped calls default `notifyOnFinish` to true; top-level calls have no parent callback.
+**`send_agent_prompt`** — `{ agentId, prompt }`. Use for follow-ups to an existing agent. Returns 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 } }`.