test(acp): start hung-session deadlines after initialization

Wait for the fake agent to receive session/new before advancing its diagnostic deadline. Slow process startup must not turn a session-timeout assertion into an initialization timeout.

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.
 
 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.
 

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,157 @@
+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";
+
+const AGENT_ID = "00000000-0000-4000-8000-000000000001";
+async function drainSessionEvents(): Promise<void> {
+  for (let index = 0; index < 10_000; index++) await Promise.resolve();
+}
+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 Pi subagent output");
+}
+afterEach(() => vi.useRealTimers());
+describe("provider subagent lifecycle ownership", () => {
+  test(
+    "a delayed parent turn end preserves a Pi child until its provider settles it",
+    { timeout: 15_000 },
+    async () => {
+      // Run the real Pi client and adapter with an injected runtime.
+      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");
+
+        // Model delayed delivery of the parent turn end, with no provider cancellation.
+        // Only Date moves: the adapter's real polling interval has not expired its miss budget.
+        vi.useFakeTimers({ toFake: ["Date"] });
+        vi.setSystemTime(new Date(Date.now() + 60_000));
+        piSession.emit({ type: "agent_start" });
+        piSession.emit({ type: "agent_end" });
+        await drainSessionEvents();
+        expect(row()?.status).toBe("running");
+
+        // The provider's completion notice, rather than elapsed time in the manager, settles it.
+        piSession.emit({
+          type: "message_end",
+          message: {
+            role: "custom",
+            customType: "subagent-notify",
+            content: `Background task completed: **scout**\nRetention-managed async directory: ${join(workdir, "gone")}`,
+            display: true,
+          },
+        });
+        await drainSessionEvents();
+        expect(row()?.status).toBe("completed");
+      } 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("history replay preserves the async status supplied by Pi", async () => {
+    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,
+      });
+      pi.latestSession().messages = [
+        {
+          role: "assistant",
+          content: [
+            {
+              type: "toolCall",
+              id: "call-replay",
+              name: "subagent",
+              arguments: { workflowScript: "return await runs.all([]);", async: true },
+            },
+          ],
+        },
+        {
+          role: "toolResult",
+          toolCallId: "call-replay",
+          toolName: "subagent",
+          content: [],
+          details: {
+            mode: "workflow",
+            runId: "run-replay",
+            asyncId: "run-replay",
+            asyncDir: join(workdir, "gone"),
+            results: [],
+          },
+        },
+      ];
+      await manager.hydrateTimelineFromProvider(AGENT_ID, { force: true });
+      expect(manager.getProviderSubagent(AGENT_ID, "call-replay")?.status).toBe("running");
+    } finally {
+      await manager.closeAgent(AGENT_ID);
+      rmSync(workdir, { recursive: true, force: true });
+    }
+  });
+});

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/providers/claude/rewind-compaction.local.e2e.test.ts

@@ -1,86 +0,0 @@
-import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
-import path from "node:path";
-import { expect, test } from "vitest";
-import { realClaudeRewindSdk, revertClaudeConversation } from "./rewind.js";
-
-interface TranscriptEntry {
-  uuid: string;
-  sessionId: string;
-  subtype?: string;
-  forkedFrom?: { messageUuid: string };
-  compactMetadata?: {
-    preservedSegment: {
-      headUuid: string;
-      anchorUuid: string;
-      tailUuid: string;
-    };
-    preservedMessages: {
-      anchorUuid: string;
-      uuids: string[];
-      allUuids: string[];
-    };
-  };
-}
-
-function parseEntries(text: string): TranscriptEntry[] {
-  return text
-    .trim()
-    .split("\n")
-    .map((line) => JSON.parse(line) as TranscriptEntry);
-}
-
-// Fixture from PR #3938, commit 374b08f9d56d9dabc62f8e4e01acc73d1c619548.
-// This proves stale metadata, not the reported model context overflow.
-test("keeps compaction references attached to forked messages", async () => {
-  const root = await mkdtemp(path.join(process.cwd(), ".compacted-rewind-"));
-  try {
-    const fixture = await readFile(
-      new URL(
-        "./test-fixtures/issue-3837-compacted-session.jsonl",
-        import.meta.url
-      ),
-      "utf8"
-    );
-    const entries = parseEntries(fixture);
-    const sourceSessionId = entries[0]!.sessionId;
-    const transcriptPath = path.join(root, `${sourceSessionId}.jsonl`);
-    await writeFile(transcriptPath, fixture);
-    let forkId = sourceSessionId;
-    await revertClaudeConversation({
-      sdk: realClaudeRewindSdk,
-      sessionId: sourceSessionId,
-      transcriptPath,
-      messageId: entries.at(-1)!.uuid,
-      setSessionId: (id) => {
-        forkId = id;
-      },
-    });
-    const fork = parseEntries(
-      await readFile(path.join(root, `${forkId}.jsonl`), "utf8")
-    );
-    const ids = new Map(
-      fork.map((entry) => [entry.forkedFrom?.messageUuid, entry.uuid])
-    );
-    const remap = (id: string) => ids.get(id) ?? id;
-    const source = entries.find(
-      (entry) => entry.subtype === "compact_boundary"
-    )!.compactMetadata!;
-    const actual = fork.find((entry) => entry.subtype === "compact_boundary")!
-      .compactMetadata!;
-    expect(actual).toEqual({
-      ...source,
-      preservedSegment: {
-        headUuid: remap(source.preservedSegment.headUuid),
-        anchorUuid: remap(source.preservedSegment.anchorUuid),
-        tailUuid: remap(source.preservedSegment.tailUuid),
-      },
-      preservedMessages: {
-        anchorUuid: remap(source.preservedMessages.anchorUuid),
-        uuids: source.preservedMessages.uuids.map(remap),
-        allUuids: source.preservedMessages.allUuids.map(remap),
-      },
-    });
-  } finally {
-    await rm(root, { recursive: true, force: true });
-  }
-});

packages/server/src/server/agent/providers/claude/test-fixtures/issue-3837-compacted-session.jsonl

@@ -1,6 +0,0 @@
-{"type":"user","uuid":"10000000-0000-4000-8000-000000000002","parentUuid":null,"isSidechain":false,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z","message":{"role":"user","content":"Before compaction"}}
-{"type":"assistant","uuid":"10000000-0000-4000-8000-000000000003","parentUuid":"10000000-0000-4000-8000-000000000002","isSidechain":false,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z","message":{"role":"assistant","content":[{"type":"text","text":"Done"}]}}
-{"type":"system","subtype":"stop_hook_summary","uuid":"10000000-0000-4000-8000-000000000004","parentUuid":"10000000-0000-4000-8000-000000000003","isSidechain":false,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z"}
-{"type":"system","subtype":"compact_boundary","uuid":"10000000-0000-4000-8000-000000000005","parentUuid":null,"logicalParentUuid":"10000000-0000-4000-8000-000000000004","isSidechain":false,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z","compactMetadata":{"trigger":"auto","preTokens":900000,"preservedSegment":{"headUuid":"10000000-0000-4000-8000-000000000003","anchorUuid":"10000000-0000-4000-8000-000000000006","tailUuid":"10000000-0000-4000-8000-000000000004"},"preservedMessages":{"anchorUuid":"10000000-0000-4000-8000-000000000006","uuids":["10000000-0000-4000-8000-000000000003","10000000-0000-4000-8000-000000000004"],"allUuids":["10000000-0000-4000-8000-000000000003","10000000-0000-4000-8000-000000000008","10000000-0000-4000-8000-000000000004"]},"postTokens":10000}}
-{"type":"user","uuid":"10000000-0000-4000-8000-000000000006","parentUuid":"10000000-0000-4000-8000-000000000005","isSidechain":false,"isCompactSummary":true,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z","message":{"role":"user","content":"Compacted summary"}}
-{"type":"user","uuid":"10000000-0000-4000-8000-000000000007","parentUuid":"10000000-0000-4000-8000-000000000006","isSidechain":false,"sessionId":"10000000-0000-4000-8000-000000000001","timestamp":"2026-08-27T08:00:00.000Z","message":{"role":"user","content":"Continue after compaction"}}

packages/server/src/server/agent/providers/generic-acp-agent.diagnostic.test.ts

@@ -3,7 +3,8 @@ import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import path from "node:path";
 
-import { describe, expect, test } from "vitest";
+import { describe, expect, test, vi } from "vitest";
+import { setTimeout as delay } from "node:timers/promises";
 
 import { createTestLogger } from "../../../test-utils/test-logger.js";
 import { runProviderRefreshWithDeadline } from "../provider-refresh-deadline.js";
@@ -203,16 +204,28 @@ describe("GenericACPAgentClient diagnostics", () => {
   });
 
   test("reports a hung ACP session/new phase without failing the diagnostic", async () => {
-    await withFakeACPAgent("hang-session", async (scriptPath, mode) => {
+    await withFakeACPAgent("hang-session", async (scriptPath, mode, testDir) => {
+      const tracePath = path.join(testDir, "phase-trace.jsonl");
       const client = new GenericACPAgentClient({
         logger: createTestLogger(),
-        command: [process.execPath, scriptPath, mode],
+        command: [process.execPath, scriptPath, mode, "", tracePath],
         providerId: "grok",
         label: "Grok",
         diagnosticPhaseTimeoutMs: TEST_ACP_TIMEOUT_MS,
       });
 
-      const { diagnostic } = await client.getDiagnostic();
+      // Freeze phase deadlines until the child actually reaches session/new. Process startup
+      // and initialize latency must not consume the session timeout this test exercises.
+      vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+      let diagnostic: string;
+      try {
+        const pending = client.getDiagnostic();
+        await waitForSessionNew(tracePath);
+        await vi.advanceTimersByTimeAsync(TEST_ACP_TIMEOUT_MS);
+        ({ diagnostic } = await pending);
+      } finally {
+        vi.useRealTimers();
+      }
 
       expect(diagnostic).toContain("Grok (ACP)");
       expect(diagnostic).toContain("Provider ID: grok");
@@ -356,6 +369,14 @@ async function withTempDir(prefix: string, run: (testDir: string) => Promise<voi
   }
 }
 
+async function waitForSessionNew(tracePath: string): Promise<void> {
+  const readyBy = Date.now() + 10_000;
+  while (!(await readFile(tracePath, "utf8").catch(() => "")).includes("session/new")) {
+    if (Date.now() >= readyBy) throw new Error("Fake ACP agent did not reach session/new");
+    await delay(20);
+  }
+}
+
 async function expectProcessExit(pid: number): Promise<void> {
   const deadline = Date.now() + 2_000;
   while (Date.now() < deadline) {
@@ -426,6 +447,7 @@ rl.on("line", (line) => {
 
   if (message.method === "session/new") {
     if (mode === "hang-session") {
+      if (initializeTracePath) fs.appendFileSync(initializeTracePath, "session/new\\n");
       return;
     }
 

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -2101,6 +2101,52 @@ describe("PiRpcAgentSession", () => {
     });
   });
 
+  test("fails an unfinished workflow on process exit while preserving detached runs", async () => {
+    const { pi, session, events } = await createSession();
+    onTestFinished(() => session.close());
+    const updates: AgentStreamEvent[] = [];
+    session.subscribe((event) => updates.push(event));
+    await session.startTurn("run workflows");
+    const runtime = pi.latestSession();
+    runtime.emit({
+      type: "tool_execution_start",
+      toolCallId: "detached",
+      toolName: "subagent",
+      args: { workflowScript: "return await runs.all([]);", async: true },
+    });
+    runtime.emit({
+      type: "tool_execution_end",
+      toolCallId: "detached",
+      toolName: "subagent",
+      result: {
+        details: {
+          mode: "workflow",
+          runId: "detached-run",
+          asyncId: "detached-run",
+          asyncDir: "/nonexistent/pi-exit-regression",
+          results: [],
+        },
+      },
+    });
+    runtime.emit({
+      type: "tool_execution_start",
+      toolCallId: "foreground",
+      toolName: "subagent",
+      args: { workflowScript: "return await runs.all([]);", async: false },
+    });
+    runtime.emit({ type: "process_exit", error: "Pi exited" });
+
+    await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Pi exited" });
+    const statuses = (id: string) =>
+      updates.flatMap((event) =>
+        event.type === "provider_subagent" && event.event.type === "upsert" && event.event.id === id
+          ? [event.event.status]
+          : [],
+      );
+    expect(statuses("foreground")).toEqual(["running", "failed"]);
+    expect(statuses("detached")).toEqual(["running", "running"]);
+  });
+
   test("fails an autonomous turn when the Pi process exits before settlement", async () => {
     const { pi, events } = await createSession();
     const fakeSession = pi.latestSession();

packages/server/src/server/agent/providers/pi/agent.ts

@@ -2108,6 +2108,15 @@ export class PiRpcAgentSession implements AgentSession {
   }
 
   private handleProcessExit(error: string): void {
+    for (const [toolCallId, toolCall] of this.activeToolCalls) {
+      this.handleToolExecutionEnd({
+        type: "tool_execution_end",
+        toolCallId,
+        toolName: toolCall.toolName,
+        isError: true,
+        result: { content: [{ type: "text", text: error }] },
+      });
+    }
     this.rejectAllExtensionResults(new Error(error));
     this.interruptingTurn = null;
     if (!this.activeTurnId && !this.activeTurnStarted) {

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,858 @@ 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[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])).toEqual(
+        expect.arrayContaining([
+          ["call-wf:alpha", "running"],
+          ["call-wf:beta", "running"],
+          ["call-wf", "running"],
+        ]),
+      );
+
+      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);
+      // A missing status file supplies no new outcome before the miss budget expires.
+      const duringMisses = subagentUpserts(received);
+      expect(duringMisses).toEqual([]);
+
+      // 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",
+      });
+      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",
+        },
+      ]);
+
+      // 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 launch placeholder is replaced once the workflow supplies its child rows.
+    expect(
+      host.mapToolCall({
+        callId: "fanout",
+        toolName: "subagent",
+        args,
+        status: "running",
+        result: null,
+      })?.subagents,
+    ).toEqual([expect.objectContaining({ id: "fanout", status: "running" })]);
+    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" });
+      }
+      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,164 @@ 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,
+          });
+        } 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,
+          });
+        }
+        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.
+      // 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,
+        });
+      }
+      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 }] };
+        }
+        // Follow the async run from its saved status or later poll updates.
         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 }] };
+      }
+      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() ?? "",
+      };
+      // 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 }],
         };
-      }
-      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 +448,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 +467,26 @@ 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;
             }
+            continue;
           }
-          if (terminal) {
-            if (parsed.data.steps.length === 0) {
-              subagents.push({
-                type: "upsert",
-                id: run.owner,
-                status: parsed.data.state === "complete" ? "completed" : "failed",
-              });
-            }
-            run.polling = false;
-          }
+          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 } }`.