fix(pi): surface workflowScript subagents and finalize stuck rows

2026-10-30 13:13:25 +00:00 · 2026-10-17 18:46:18 +05:30
parent a0aeca967c
commit 9ae7a64aa7
8 changed files with 2093 additions and 83 deletions
@@ -210,7 +210,7 @@ The rows combine two kinds of children:
 parentAgentId === thisAgent.id  AND  !archivedAt
 ```

- **Provider subagents** are child executions owned by Claude, Codex, or OpenCode. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
+- **Provider subagents** are child executions owned by Claude, Codex, OpenCode, or Pi. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent. Pi rows that a poll keeps alive carry `liveness: "poll"`, and only a marked row can be settled by a sweep once its poll stops reporting — that gate is what stops an idle parent from leaving a card spinning and keeps other providers' rows untouched. Unmarked rows and every other provider settle their own rows.

 Clicking either kind opens a workspace tab. On compact layouts it opens in a [sheet](explorer-sidebar.md#sheets) over the subagents list instead; the sheet hides the composer and shows the agent's model and thinking as one line in its place (`PaneSubtitle`, the row provider subagents already use), and a Paseo subagent's sheet offers **Open as tab** (the full-screen glyph in the header) to get the composer back. A Paseo subagent tab is a normal interactive agent pane. A provider subagent tab is a read-only timeline pane with no composer, archive, detach, rewind, or fork actions. It shows its own direct children in a subagents track. Both panes use `AgentStreamView`, so message, reasoning, tool-call, and layout rendering stay identical.

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

+/**
+ * How recently a poll-backed Pi subagent row must have been updated to count as alive. Only the
+ * pi-subagents extension polls: it re-upserts every live row on each 250 ms follow tick and marks
+ * those upserts `liveness: "poll"`, so a marked row untouched for a whole window can never change
+ * again. The other Pi extensions (gotgenes/tintinweb pi-subagents and friends) are event-driven
+ * and may stay silent while a background child still runs, and Claude
+ * (providers/claude/agent.ts:4779,4805) and OpenCode (providers/opencode/v2/children.ts:64-68)
+ * settle their own rows at turn or child end. No row without the marker is ever swept.
+ */
+const PI_SUBAGENT_LIVENESS_MS = 5_000;
+
 export interface RetiredInternalAgent {
  record: ArchivedStoredAgentRecord;
  lastMessage: string | null;
@@ -743,6 +754,7 @@ export class AgentManager {
  private readonly agents = new Map<string, LiveManagedAgent>();
  private readonly timelineStore = new InMemoryAgentTimelineStore();
  private readonly providerSubagents = new ProviderSubagentStore();
+  private readonly providerSubagentSweepTimers = new Map<string, ReturnType<typeof setTimeout>>();
  private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
  private readonly sessionEventTails = new Map<string, Promise<void>>();
  private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
@@ -1812,15 +1824,157 @@ export class AgentManager {
      if (subagent.status !== "running") {
        continue;
      }
-      const event = this.providerSubagents.apply(parentAgentId, subagent.provider, {
-        type: "upsert",
-        id: subagent.id,
-        status: "canceled",
-      });
-      this.dispatch({ type: "provider_subagent", event });
+      this.applyProviderSubagentStatus(parentAgentId, subagent.provider, subagent.id, "canceled");
    }
  }

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

+    // Rows the store already reports running are the live session's: a rewind replays history
+    // without recreating the session, so their poll keeps running.
+    const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
    for (const event of this.providerSubagents.deleteParent(agent.id)) {
      if (broadcast) {
        this.dispatch({ type: "provider_subagent", event });
      }
    }
+    const replayedRunningIds = new Set<string>();
    for (const event of providerSubagentEvents) {
      const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
+      if (event.event.type === "upsert" && event.event.status === "running") {
+        replayedRunningIds.add(event.event.id);
+      }
      if (broadcast) {
        this.dispatch({ type: "provider_subagent", event: update });
      }
@@ -4171,6 +4332,7 @@ export class AgentManager {
        });
      }
    }
+    this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
    this.touchUpdatedAt(agent);
    this.emitState(agent);
  }
@@ -4217,8 +4379,15 @@ export class AgentManager {
      row: AgentTimelineRow;
    }> = [];
    const providerSubagentEvents: AgentManagerEvent[] = [];
+    const replayedRunningIds = new Set<string>();
+    // A row the store still reports running is not a fresh resume: it belongs to the live
+    // session's poll.
+    const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
    for (const event of historySubagentEvents) {
      const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
+      if (event.event.type === "upsert" && event.event.status === "running") {
+        replayedRunningIds.add(event.event.id);
+      }
      const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
      if (deferredBroadcast) {
        providerSubagentEvents.push(managerEvent);
@@ -4245,6 +4414,7 @@ export class AgentManager {
    agent.historyPrimed = true;

    if (typeof broadcast !== "function" || !broadcast()) {
+      this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
      return;
    }
    for (const event of providerSubagentEvents) {
@@ -4257,6 +4427,7 @@ export class AgentManager {
        timestamp: row.timestamp,
      });
    }
+    this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
  }

  private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
@@ -4342,6 +4513,7 @@ export class AgentManager {
        if (isForegroundEvent) {
          this.finalizeForegroundTurn(agent, eventTurnId);
        }
+        this.finalizeUnprovenProviderSubagents(agent.id);
      }

      if (flags.shouldDispatchEvent) {
@@ -60,6 +60,23 @@ describe("ProviderSubagentStore", () => {
    expect(subagents.list("parent-b")).toHaveLength(1);
  });

+  test("keeps a poll liveness marker across presentation-only upserts", () => {
+    const subagents = new ProviderSubagentStore();
+    subagents.apply("parent-a", "pi", {
+      type: "upsert",
+      id: "child-1",
+      status: "running",
+      liveness: "poll",
+    });
+    subagents.apply("parent-a", "pi", { type: "upsert", id: "child-1", title: "scout" });
+
+    expect(subagents.get("parent-a", "child-1")).toMatchObject({
+      title: "scout",
+      status: "running",
+      liveness: "poll",
+    });
+  });
+
  test("limits oversized provider child tool output before storage", () => {
    const subagents = new ProviderSubagentStore();
    const output = "x".repeat(70 * 1024);
@@ -23,6 +23,11 @@ export interface ProviderSubagentDescriptor {
  toolCallId: string | null;
  cwd: string | null;
  subtitle: string | null;
+  /**
+   * Set when a live poll re-reports this row on every tick, so silence proves the poll is gone.
+   * Rows without it are event-driven and may stay quiet while their child still runs.
+   */
+  liveness?: "poll";
 }

 export type ProviderSubagentInputEvent =
@@ -40,6 +45,8 @@ export type ProviderSubagentInputEvent =
      cwd?: string | null;
      subtitle?: string | null;
      parentSubagentId?: string | null;
+      /** Only a poll-backed reporter may set this; it is what makes a stale row sweepable. */
+      liveness?: "poll";
      timestamp?: string;
    }
  | {
@@ -113,6 +120,7 @@ export class ProviderSubagentStore {
      this.timelines.initialize(key);
    }
    const timestamp = event.timestamp ?? new Date().toISOString();
+    const liveness = event.liveness ?? previous?.liveness;
    const subagent: ProviderSubagentDescriptor = {
      id: event.id,
      parentAgentId,
@@ -126,6 +134,7 @@ export class ProviderSubagentStore {
      cwd: stickyField(event.cwd, previous?.cwd),
      subtitle: stickyField(event.subtitle, previous?.subtitle),
      parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId),
+      ...(liveness === undefined ? {} : { liveness }),
    };
    this.descriptors.set(key, subagent);
    return { type: "upsert", subagent };
@@ -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
+    }
+  ]
+}
@@ -1,4 +1,4 @@
-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 { tmpdir } from "node:os";
@@ -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,750 @@ describe("pi-subagents adapter", () => {
      },
    ]);
  });
+
+  test("maps a reconstructed workflowScript run to a live spawn row and its children", async () => {
+    const fixture = readSubagentFixture(new URL("./fixtures/workflow.json", import.meta.url));
+    const events = await verifySubagentFixture(fixture);
+    const upserts = events
+      .filter((event) => event.event.type === "upsert")
+      .map((event) => (event.event.type === "upsert" ? event.event : null));
+    expect(upserts.map((event) => [event?.id, event?.status])).toEqual([
+      ["call_wf1", "running"],
+      ["call_wf1", "running"],
+      ["call_wf1:alpha", "completed"],
+      ["call_wf1", "completed"],
+    ]);
+    // The workflow call carries no agent/task, so the script file names the run.
+    expect(upserts[0]?.title).toBe("rigor-ui-test.js");
+    expect(upserts[1]?.liveness).toBe("poll");
+    expect(upserts[2]?.title).toBeUndefined();
+  });
+
+  test("maps a reported workflowScript run to its reviewer row and saved timeline", async () => {
+    // Reported in #5837: `subagent({ workflow: true })` reaches the adapter as `workflowScript`,
+    // not `workflow`, so a launch check keyed on `workflow` alone drops the whole run.
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-script-"));
+    const host = createPiExtensionHost();
+    try {
+      const sessionFile = join(dir, "child.jsonl");
+      await writeFile(
+        sessionFile,
+        JSON.stringify({
+          type: "message",
+          message: { role: "assistant", content: [{ type: "text", text: "README summary" }] },
+        }) + "\n",
+      );
+      const args = {
+        workflowScript:
+          "return await runs.run('review', { agent: 'reviewer', task: 'Read README.md' });",
+        async: false,
+      };
+      expect(
+        host.mapToolCall({
+          callId: "reported-workflow",
+          toolName: "subagent",
+          args,
+          status: "running",
+          result: null,
+        })?.subagents,
+      ).toEqual([
+        expect.objectContaining({ id: "reported-workflow", title: "workflow", status: "running" }),
+      ]);
+      const output = host.mapToolCall({
+        callId: "reported-workflow",
+        toolName: "subagent",
+        args,
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "reported-run",
+            results: [{ index: 0, agent: "reviewer", exitCode: 0, sessionFile }],
+          },
+        },
+      });
+      expect(output?.subagents).toEqual(
+        expect.arrayContaining([
+          expect.objectContaining({ type: "upsert", title: "reviewer", status: "completed" }),
+        ]),
+      );
+      expect(hasChildTimeline(await output!.hydration)).toBe(true);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("keys workflow child rows by their childKey while the run reports progress", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      await expect.poll(() => reportedUpsert(received, "call-wf:beta")).toBe(true);
+      expect(
+        subagentUpserts(received).map((event) => [event.id, event.status, event.liveness]),
+      ).toEqual(
+        expect.arrayContaining([
+          ["call-wf:alpha", "running", "poll"],
+          ["call-wf:beta", "running", "poll"],
+          ["call-wf", "running", "poll"],
+        ]),
+      );
+
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "complete",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "complete" },
+          ],
+        }),
+      );
+      await expect.poll(() => reportedUpsert(received, "call-wf", "completed")).toBe(true);
+      expect(
+        subagentUpserts(received).findLast((event) => event.id === "call-wf:alpha")?.status,
+      ).toBe("completed");
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("keeps a missing status directory's row alive until the miss budget expires", async () => {
+    vi.useFakeTimers();
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-lost",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "single",
+            runId: "run-1",
+            asyncId: "run-1",
+            asyncDir: "/nonexistent",
+            results: [],
+          },
+        },
+      });
+      await vi.advanceTimersByTimeAsync(250 * 5);
+      // The run is alive even while its file is missing: every miss tick re-reports the row, or a
+      // turn end inside the miss budget would cancel live work.
+      const duringMisses = subagentUpserts(received);
+      expect(duringMisses.length).toBeGreaterThanOrEqual(5);
+      expect(duringMisses.every((event) => event.status === "running")).toBe(true);
+      expect(duringMisses.every((event) => event.liveness === "poll")).toBe(true);
+
+      // 60 s of missing ticks, the budget before a status directory counts as gone.
+      await vi.advanceTimersByTimeAsync(250 * 236);
+      // The launch row comes back through the caller; polling reports the give-up exactly once.
+      const afterGiveUp = subagentUpserts(received);
+      expect(afterGiveUp.findLast((event) => event.id === "call-lost")).toEqual({
+        type: "upsert",
+        id: "call-lost",
+        status: "canceled",
+        liveness: "poll",
+      });
+      expect(afterGiveUp.filter((event) => event.status === "canceled")).toHaveLength(1);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+    }
+  });
+
+  test("a completion notice stops polling so a stale status file cannot revert the card", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stale-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({ state: "running", steps: [{ agent: "scout", status: "running" }] }),
+      );
+      const statuses = () =>
+        received.flatMap((event) =>
+          event.type === "provider_subagent" && event.event.type === "upsert"
+            ? [event.event.status]
+            : [],
+        );
+      await expect.poll(() => statuses().length).toBeGreaterThan(1);
+
+      host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-notify",
+        content: [
+          "Background task completed: **scout**",
+          "",
+          `Retention-managed async directory: ${dir}`,
+        ].join("\n"),
+      });
+      const settled = statuses().length;
+
+      // The run's directory is still on disk with a running status; polling must stay off.
+      await new Promise((resolve) => setTimeout(resolve, 400));
+      expect(statuses().length).toBe(settled);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a bg_wait result keeps its verdict while the run's status names no child yet", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-settled-owner-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.follow((event) => received.push(event));
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      // A single unnamed child takes the owner's id, so this result settles the launch row while the
+      // run itself keeps polling.
+      expect(
+        host.mapToolCall({
+          callId: "wait-1",
+          toolName: "bg_wait",
+          args: { runId: "run-1" },
+          status: "completed",
+          result: {
+            details: {
+              mode: "single",
+              results: [],
+              completions: [
+                { runId: "run-1", results: [{ agent: "scout", success: true, exitCode: 0 }] },
+              ],
+            },
+          },
+        })?.subagents,
+      ).toEqual([expect.objectContaining({ id: "call-1", status: "completed" })]);
+
+      // The queue has not started the child yet, so the status file names no step: that tick must
+      // not re-report the owner as running over the verdict it already has.
+      await writeFile(join(dir, "status.json"), JSON.stringify({ state: "running", steps: [] }));
+      const beforeTick = received.length;
+      await vi.advanceTimersByTimeAsync(250 * 2);
+
+      expect(subagentUpserts(received.slice(beforeTick))).toEqual([]);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a paused run settles as canceled", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-paused-"));
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { agent: "scout", task: "Inspect", async: true },
+        status: "completed",
+        result: {
+          details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
+        },
+      });
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-notify",
+          content: [
+            "Background task paused: **scout**",
+            "",
+            `Retention-managed async directory: ${dir}`,
+          ].join("\n"),
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-1", status: "canceled" }]);
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a paused workflow child notice settles its row as canceled", () => {
+    const host = createPiExtensionHost();
+    host.mapToolCall({
+      callId: "call-wf",
+      toolName: "subagent",
+      args: { workflow: "tmp/rigor-ui-test.js", async: true },
+      status: "completed",
+      result: {
+        details: {
+          mode: "workflow",
+          runId: "run-wf",
+          asyncId: "run-wf",
+          asyncDir: "/nonexistent",
+          results: [],
+        },
+      },
+    });
+    expect(
+      host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-incremental-child-notify",
+        content: "Workflow child paused (needs attention): **alpha**\nWorkflow run: run-wf",
+      })?.subagents,
+    ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "canceled" }]);
+  });
+
+  test("partial and rejected runs settle every child instead of leaving a spinner", async () => {
+    for (const [state, expected] of [
+      ["partial", ["completed", "completed", "canceled"]],
+      ["rejected", ["completed", "completed", "failed"]],
+    ] as const) {
+      vi.useFakeTimers();
+      const dir = await mkdtemp(join(tmpdir(), "paseo-pi-partial-"));
+      const received: AgentStreamEvent[] = [];
+      const host = createPiExtensionHost();
+      try {
+        host.mapToolCall({
+          callId: "call-wf",
+          toolName: "subagent",
+          args: { workflow: "tmp/rigor-ui-test.js", async: true },
+          status: "completed",
+          result: {
+            details: {
+              mode: "workflow",
+              runId: "run-wf",
+              asyncId: "run-wf",
+              asyncDir: dir,
+              results: [],
+            },
+          },
+        });
+        await writeFile(
+          join(dir, "status.json"),
+          JSON.stringify({
+            state,
+            steps: [
+              { agent: "scout", workflowKey: "alpha", status: "complete" },
+              // WorkflowNodeStatus spells the same outcome `completed`.
+              { agent: "auditor", workflowKey: "gamma", status: "completed" },
+              {
+                agent: "reviewer",
+                workflowKey: "beta",
+                status: state === "partial" ? "running" : "failed",
+              },
+            ],
+          }),
+        );
+        host.follow((event) => received.push(event));
+        await vi.advanceTimersByTimeAsync(250);
+
+        expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
+          expect.arrayContaining([
+            ["call-wf:alpha", expected[0]],
+            ["call-wf:gamma", expected[1]],
+            ["call-wf:beta", expected[2]],
+            ["call-wf", "canceled"],
+          ]),
+        );
+        expect(
+          subagentUpserts(received).filter((event) => event.status === "running"),
+        ).toHaveLength(0);
+
+        // The run is terminal, so the poll stops instead of re-reporting fresh rows.
+        const settled = received.length;
+        await vi.advanceTimersByTimeAsync(250 * 4);
+        expect(received.length).toBe(settled);
+      } finally {
+        host.close();
+        vi.useRealTimers();
+        await rm(dir, { recursive: true, force: true });
+      }
+    }
+  });
+
+  test("a child notice keeps its status while the run is still reporting", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-pin-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      // The stale step entry still says running; the child's notice is its verdict.
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "complete" },
+            { agent: "reviewer", workflowKey: "beta", status: "running" },
+          ],
+        }),
+      );
+      host.follow((event) => received.push(event));
+      await vi.advanceTimersByTimeAsync(250);
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      const beforeNotice = received.length;
+      await vi.advanceTimersByTimeAsync(250 * 3);
+      expect(
+        subagentUpserts(received.slice(beforeNotice)).filter(
+          (event) => event.id === "call-wf:alpha",
+        ),
+      ).toEqual([]);
+      // The rest of the run keeps re-reporting, so this is not a poll that simply stopped.
+      expect(reportedUpsert(received.slice(beforeNotice), "call-wf:beta")).toBe(true);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a child notice's verdict survives the run's terminal tick", async () => {
+    vi.useFakeTimers();
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-terminal-"));
+    const received: AgentStreamEvent[] = [];
+    const host = createPiExtensionHost();
+    try {
+      host.mapToolCall({
+        callId: "call-wf",
+        toolName: "subagent",
+        args: { workflow: "tmp/rigor-ui-test.js", async: true },
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            runId: "run-wf",
+            asyncId: "run-wf",
+            asyncDir: dir,
+            results: [],
+          },
+        },
+      });
+      host.follow((event) => received.push(event));
+      // The child's notice settles it before the run itself ends.
+      expect(
+        host.mapCustomMessage({
+          role: "custom",
+          customType: "subagent-incremental-child-notify",
+          content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
+        })?.subagents,
+      ).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
+
+      // The run then ends without a verdict for alpha: its step entry still says running, which the
+      // terminal tick would read as canceled if the notice were not the child's verdict.
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "stopped",
+          steps: [
+            { agent: "scout", workflowKey: "alpha", status: "running" },
+            { agent: "reviewer", workflowKey: "beta", status: "stopped" },
+          ],
+        }),
+      );
+      await vi.advanceTimersByTimeAsync(250);
+
+      expect(
+        subagentUpserts(received).filter((event) => event.id === "call-wf:alpha" && event.status),
+      ).toEqual([]);
+      // 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", liveness: "poll" },
+      ]);
+
+      // Reported once, not on every tick.
+      await vi.advanceTimersByTimeAsync(250 * 2);
+      expect(alphaUpserts()).toHaveLength(1);
+    } finally {
+      host.close();
+      vi.useRealTimers();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+
+  test("a workflow that reported rows replaces its placeholder with them", () => {
+    const host = createPiExtensionHost();
+    const args = { workflow: true };
+    // The in-flight placeholder is poll-marked, so a turn end can settle it if its call never
+    // reports back.
+    expect(
+      host.mapToolCall({
+        callId: "fanout",
+        toolName: "subagent",
+        args,
+        status: "running",
+        result: null,
+      })?.subagents,
+    ).toEqual([expect.objectContaining({ id: "fanout", status: "running", liveness: "poll" })]);
+    expect(
+      host.mapToolCall({
+        callId: "fanout",
+        toolName: "subagent",
+        args,
+        status: "completed",
+        result: {
+          details: {
+            mode: "workflow",
+            results: [
+              { index: 0, agent: "scout", workflowKey: "alpha", success: true, exitCode: 0 },
+              { index: 1, agent: "reviewer", workflowKey: "beta", success: true, exitCode: 0 },
+            ],
+          },
+        },
+      })?.subagents,
+    ).toEqual([
+      { type: "remove", id: "fanout" },
+      expect.objectContaining({ id: "fanout:alpha", title: "scout", status: "completed" }),
+      expect.objectContaining({ id: "fanout:beta", title: "reviewer", status: "completed" }),
+    ]);
+  });
 });
@@ -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 Set<string>();
+    /** 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: string) => {
+      if (status !== "running") settledRows.add(id);
+    };
    /** Keyed by async directory, which both status.json and the completion notice carry. */
-    const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
+    const asyncRuns = new Map<string, AsyncRun>();
    const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
      if (details.asyncDir && call.status !== "failed") {
-        asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true });
+        asyncRuns.set(details.asyncDir, {
+          owner: call.callId,
+          polling: true,
+          misses: 0,
+          deadline: Date.now() + POLL_TTL_MS,
+          rows: new Set<string>(),
+        });
      }
    };
+    /**
+     * The run is over and nothing will ever update its rows again, so settle them rather than leave a
+     * spinner. A row that already reported a terminal status keeps it: this is not its verdict.
+     */
+    const giveUpRun = (run: AsyncRun) => {
+      run.polling = false;
+      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
+      for (const id of new Set([run.owner, ...run.rows])) {
+        if (settledRows.has(id)) continue;
+        subagents.push({ type: "upsert", id, status: "canceled", liveness: "poll" });
+      }
+      return subagents;
+    };
+    /**
+     * The notice is the run's last word: it ends the poll, settles the launch row, and cancels the
+     * children the run stopped with. No later tick will report those, and nothing here proves they
+     * succeeded, so they settle as canceled. A child that already reported its own verdict keeps it.
+     */
+    const settleRunNotice = (run: AsyncRun, status: "completed" | "failed" | "canceled") => {
+      // A stale status.json must not revert this card either.
+      run.polling = false;
+      recordTerminal(run.owner, status);
+      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [
+        { type: "upsert", id: run.owner, status },
+      ];
+      for (const id of run.rows) {
+        if (settledRows.has(id)) continue;
+        subagents.push({ type: "upsert", id, status: "canceled" });
+      }
+      return subagents;
+    };
+    /**
+     * A row id can settle twice: a completion notice records a verdict in `settledRows`, and a
+     * `subagent_wait` result repeats the run's rows with the run's own verdict. Both are terminal, so
+     * a disagreement moves a row terminal-to-terminal (completed → failed) and never resurrects it;
+     * the owner row is what carries the run outcome.
+     */
+    // ponytail: child verdicts can diverge between notice and bg_wait result — unify when one source of truth exists
    const collectRows = (
      owner: string,
      rows: z.infer<typeof Row>[],
@@ -87,17 +243,19 @@ export const piSubagents: PiExtension = {
      const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
      const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
      for (const [index, row] of rows.entries()) {
-        const id = rows.length === 1 ? owner : `${owner}:${row.index ?? index}`;
+        const id = rowId(owner, rows.length, row.workflowKey, row.index ?? index);
+        const status =
+          failed || row.success === false || (row.success !== true && row.exitCode !== 0)
+            ? ("failed" as const)
+            : ("completed" as const);
+        recordTerminal(id, status);
        subagents.push({
          type: "upsert",
          id,
          title: row.agent,
          description,
          toolCallId: owner,
-          status:
-            failed || row.success === false || (row.success !== true && row.exitCode !== 0)
-              ? "failed"
-              : "completed",
+          status,
        });
        if (row.sessionFile && !readSessions.has(row.sessionFile)) {
          readSessions.add(row.sessionFile);
@@ -123,66 +281,105 @@ 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 }] };
+    /** A settled spawn: Pi either reports the rows it ran, or an async directory the poll follows. */
+    const mapSettledSpawn = (
+      call: PiExtensionToolCall,
+      base: SubagentUpsert,
+      description: string | undefined,
+      asyncRequested: boolean,
+    ): Pick<PiExtensionToolMapping, "subagents" | "childSessions"> => {
+      const failed = call.status === "failed";
      const details = Details.safeParse(
        typeof call.result === "object" ? call.result?.details : null,
      );
-      if (!details.success || details.data.mode === "management")
-        return call.status === "failed"
-          ? { detail, subagents: [{ ...base, status: "failed" as const }] }
-          : { detail };
+      if (!details.success || details.data.mode === "management") {
+        if (!failed) return { subagents: [] };
+        recordTerminal(call.callId, "failed");
+        return { subagents: [{ ...base, status: "failed" as const }] };
+      }
      if (details.data.runId) callsByRun.set(details.data.runId, call.callId);
-      if (details.data.results.length === 0 && (details.data.asyncId || args.data.async)) {
+      if (details.data.results.length === 0) {
+        if (failed) {
+          recordTerminal(call.callId, "failed");
+          return { subagents: [{ ...base, status: "failed" as const }] };
+        }
+        if (!details.data.asyncId && !asyncRequested) {
+          // ponytail: inferred completion — revisit if pi can return success while a child outlives the call.
+          recordTerminal(call.callId, "completed");
+          return { subagents: [{ ...base, status: "completed" as const }] };
+        }
+        // The placeholder row is now backed by a per-tick poll, which is what proves it alive.
        rememberAsyncRun(details.data, call);
+        return { subagents: [{ ...base, status: "running" as const, liveness: "poll" as const }] };
+      }
+      const collected = collectRows(call.callId, details.data.results, description, failed);
+      // A run that reported rows of its own never keeps the placeholder card it opened at launch.
+      if (!collected.subagents.some((event) => event.id === call.callId)) {
+        collected.subagents.unshift({ type: "remove", id: call.callId });
+      }
+      return collected;
+    };
+    const mapSpawn = (call: PiExtensionToolCall) => {
+      const args = Args.safeParse(call.args);
+      if (!args.success || args.data.action !== undefined) return undefined;
+      const single = Boolean(args.data.agent && args.data.task);
+      let workflow: string | undefined;
+      if (!single) workflow = workflowTitle(args.data);
+      if (!single && workflow === undefined) return undefined;
+      const base = {
+        type: "upsert" as const,
+        id: call.callId,
+        title: (single ? args.data.agent : workflow) ?? "workflow",
+        description: args.data.task,
+        toolCallId: call.callId,
+      };
+      const detail = {
+        type: "sub_agent" as const,
+        subAgentType: base.title,
+        description: args.data.task,
+        log: extractTextFromToolResult(call.result)?.trim() ?? "",
+      };
+      // The in-flight placeholder is poll-marked: a turn interrupted before this call reports
+      // never delivers a terminal mapping, and only a marked row can be settled by the sweep.
+      // ponytail: replay re-emits running before the settled rows land — one-pass client flip, upgrade when replay maps calls with their results first
+      if (call.status === "running")
        return {
          detail,
-          subagents: [
-            {
-              ...base,
-              status: call.status === "failed" ? ("failed" as const) : ("running" as const),
-            },
-          ],
+          subagents: [{ ...base, status: "running" as const, liveness: "poll" as const }],
        };
-      }
-      if (details.data.results.length === 0 && call.status === "failed")
-        return { detail, subagents: [{ ...base, status: "failed" as const }] };
      return {
        detail,
-        ...collectRows(call.callId, details.data.results, args.data.task, call.status === "failed"),
+        ...mapSettledSpawn(call, base, args.data.task, Boolean(args.data.async)),
      };
    };
+    const noticeText = (message: CustomMessage) =>
+      typeof message.content === "string"
+        ? message.content
+        : extractTextFromToolResult({ content: message.content });
+    const mapWorkflowChildNotice = (message: CustomMessage) => {
+      const lines = noticeText(message)?.split("\n") ?? [];
+      const status = noticeStatus(lines[0] ?? "");
+      const key = lines[0]?.match(
+        /^Workflow child (?:completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*(.+?)\*\*/,
+      )?.[1];
+      const runId = lines.find((line) => line.startsWith("Workflow run: "))?.slice(14);
+      const owner = runId ? callsByRun.get(runId) : undefined;
+      if (!status || !key || !owner) return undefined;
+      const id = `${owner}:${key}`;
+      recordTerminal(id, status);
+      return { subagents: [{ type: "upsert" as const, id, status }] };
+    };
    return {
      mapToolCall(call) {
        if (call.toolName === "bg_wait") return mapWait(call);
        if (call.toolName === "subagent") return mapSpawn(call);
        return undefined;
      },
-      // The notice is the only completion Pi persists, so replay settles async runs from it.
      mapCustomMessage(message) {
+        if (message.customType === "subagent-incremental-child-notify")
+          return mapWorkflowChildNotice(message);
        if (message.customType !== "subagent-notify") return undefined;
-        const text =
-          typeof message.content === "string"
-            ? message.content
-            : extractTextFromToolResult({ content: message.content });
-        const lines = text?.split("\n") ?? [];
+        const lines = noticeText(message)?.split("\n") ?? [];
        const status = noticeStatus(lines[0] ?? "");
        if (!status) return undefined;
        const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
@@ -192,8 +389,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,35 +408,79 @@ 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 });
+          const status = readAsyncStatus(dir);
+          if (!status) {
+            run.misses += 1;
+            if (run.misses >= POLL_MISS_LIMIT) {
+              subagents.push(...giveUpRun(run));
+              continue;
+            }
+            // The run is still alive even though its file is gone for now. Keep re-reporting its
+            // unsettled rows: silence would freeze their timestamp and a turn end would cancel a
+            // live run inside the miss budget.
+            for (const id of new Set([run.owner, ...run.rows])) {
+              if (settledRows.has(id)) continue;
+              subagents.push({
+                type: "upsert",
+                id,
+                status: "running" as const,
+                liveness: "poll" as const,
+              });
+            }
+            continue;
+          }
+          run.misses = 0;
+          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 notice is this child's verdict, so a settled row gets no further status write
+            // from the poll. While the run is going, a stale step entry must not revert it to
+            // running; on the terminal tick, the run's own outcome must not overwrite it.
+            if (!settledRows.has(id)) {
+              const childStatus = asyncChildStatus(status.state, step.status);
+              recordTerminal(id, childStatus);
+              namedRows.add(id);
+              subagents.push({
+                type: "upsert",
+                id,
+                title: step.agent,
+                status: childStatus,
+                liveness: "poll" as const,
+              });
+            } else if (!namedRows.has(id)) {
+              // A child notice can settle a row before any poll saw it, and that notice carries no
+              // agent name. Omit `status`: the store keeps the settled verdict and only takes the
+              // title the run's status file names.
+              namedRows.add(id);
+              subagents.push({ type: "upsert", id, title: step.agent, liveness: "poll" as const });
+            }
            if (step.sessionFile && !readSessions.has(step.sessionFile)) {
              readSessions.add(step.sessionFile);
              childSessions.push({ id, file: step.sessionFile });
            }
          }
-          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;
+          if (runTerminal) run.polling = false;
+          // The launch row stands for the whole run, and a single unnamed child already took its
+          // id. Re-reporting it every tick is also what proves to the manager that this run is alive.
+          // A row that already settled keeps that verdict: a result that landed before the run's
+          // steps named its children must not be reverted to running by the next tick.
+          if (!run.rows.has(run.owner) && !settledRows.has(run.owner)) {
+            const ownerStatus = run.polling ? ("running" as const) : asyncRunStatus(status.state);
+            recordTerminal(run.owner, ownerStatus);
+            subagents.push({
+              type: "upsert",
+              id: run.owner,
+              status: ownerStatus,
+              liveness: "poll",
+            });
          }
        }
        return subagents.length || childSessions.length ? { subagents, childSessions } : undefined;