fix(pi): stream running subagent transcripts (#5755)

Follow Pi child session files while subagents run for gotgenes, tintinweb, and nicobailon pi-subagents. gotgenes reports the child file through a runtime bridge that captures the parent's service at session start; nico async runs settle from status.json live and from the persisted subagent-notify notice on replay.

Closes #5752

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

@@ -165,11 +165,13 @@ async function loadPaseoExtensionListeners(
   const extension = (await import(pathToFileURL(extensionPath).href)) as {
     default: (piApi: {
       on: (event: string, listener: PaseoExtensionListener) => void;
+      events: { on: () => void };
       registerCommand: () => void;
     }) => void;
   };
   extension.default({
     on: (event, listener) => listeners.set(event, listener),
+    events: { on: () => undefined },
     registerCommand: () => undefined,
   });
   return listeners;

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

@@ -65,7 +65,11 @@ import {
 } from "./history-mapper.js";
 import { materializeProviderImage } from "../provider-image-output.js";
 import { PiCliRuntime } from "./cli-runtime.js";
-import { createPiExtensionHost, type PiExtensionEventOutput } from "./extensions/index.js";
+import {
+  createPiExtensionHost,
+  piExtensionRuntimeBridge,
+  type PiExtensionEventOutput,
+} from "./extensions/index.js";
 import { revertPiConversation } from "./rewind.js";
 import { listPiImportableSessions, readPiImportSessionConfig } from "./session-descriptor.js";
 import type { PiRuntime, PiRuntimeSession, PiStartSessionInput } from "./runtime.js";
@@ -644,6 +648,7 @@ function createPiPaseoExtensionFile(systemPrompt?: string): PiTempFile {
 
 	export default function paseoIntegration(pi) {
 	  const submittedUserMessages = [];
+	  ${piExtensionRuntimeBridge}
 
 	  function emitSubmittedUserEntries(ctx) {
 	    const entries = ctx.sessionManager.getEntries();
@@ -1168,6 +1173,9 @@ export class PiRpcAgentSession implements AgentSession {
     this.extensionTimeoutMs = options.extensionTimeoutMs ?? DEFAULT_PI_EXTENSION_RESULT_TIMEOUT_MS;
     this.logger = options.logger;
     this.extensionHost = createPiExtensionHost(this.logger);
+    this.extensionHost.follow((event) => {
+      if (!this.closeController.signal.aborted) this.emit(event);
+    });
     this.usagePoller = new PiUsagePoller({
       scheduler: options.usagePollScheduler,
       readStats: () => this.runtimeSession.getSessionStats(),
@@ -1523,6 +1531,7 @@ export class PiRpcAgentSession implements AgentSession {
     }
     this.closed = true;
     this.closeController.abort();
+    this.extensionHost.close();
     this.usagePoller.close();
     try {
       await this.runtimeSession.close();
@@ -1936,6 +1945,11 @@ export class PiRpcAgentSession implements AgentSession {
   ): void {
     const message = optionalString(event.message);
     if (event.method === "notify" && message) {
+      const extensionMapping = this.extensionHost.mapRuntimeNotification(message);
+      if (extensionMapping) {
+        this.emitExtensionOutput(extensionMapping, this.currentTurnIdForEvent());
+        return;
+      }
       if (
         this.handleSubmittedUserEntryMarker(message) ||
         this.handleEntryCaptureMarker(message) ||

packages/server/src/server/agent/providers/pi/extensions/child-session.ts

@@ -3,57 +3,96 @@ import { limitAgentTimelineItemContent } from "../../../agent-timeline-content.j
 import type { ProviderSubagentInputEvent } from "../../../provider-subagents/store.js";
 import { PiHistoryMapper } from "../history-mapper.js";
 import type { PiAgentMessage } from "../rpc-types.js";
+import { extractTextFromToolResult, type PiToolResult } from "../tool-call-mapper.js";
 
 const MAX_BYTES = 2 * 1024 * 1024;
 const MAX_ITEMS = 200;
 
-/** Child session files are immutable once the completed result exposes their path. */
-export async function mapPiChildSession(
+export function outputFileFromToolResult(result: PiToolResult): string | undefined {
+  return extractTextFromToolResult(result)?.match(/^Output file:\s*(\S+)$/m)?.[1];
+}
+
+/** Reads a finished child transcript once, up to `maxBytes`. */
+export function mapPiChildSession(
   id: string,
   file: string,
   maxBytes = MAX_BYTES,
 ): Promise<ProviderSubagentInputEvent[]> {
+  return new PiChildSessionFollower(id, file, maxBytes).readNew();
+}
+
+function mapChildLine(
+  mapper: PiHistoryMapper,
+  id: string,
+  line: string,
+): ProviderSubagentInputEvent[] {
+  let entry: { message?: PiAgentMessage; timestamp?: string };
   try {
-    const handle = await open(file, "r");
-    try {
-      const fileSize = (await handle.stat()).size;
-      const size = Math.min(fileSize, MAX_BYTES, maxBytes);
-      if (!size) return [];
-      const buffer = Buffer.alloc(size);
-      const { bytesRead } = await handle.read(buffer, 0, size, 0);
-      const text = buffer.toString("utf8", 0, bytesRead);
-      const completeText = bytesRead < fileSize ? text.slice(0, text.lastIndexOf("\n")) : text;
-      return parseChildTimeline(id, completeText);
-    } finally {
-      await handle.close();
-    }
+    entry = JSON.parse(line);
   } catch {
     return [];
   }
+  if (!entry.message || typeof entry.message !== "object" || !("role" in entry.message)) return [];
+  return mapper.mapMessages([entry.message]).flatMap((mapped) =>
+    mapped.type === "timeline"
+      ? [
+          {
+            type: "timeline" as const,
+            id,
+            item: limitAgentTimelineItemContent(mapped.item),
+            ...(entry.timestamp ? { timestamp: entry.timestamp } : {}),
+          },
+        ]
+      : [],
+  );
 }
 
-function parseChildTimeline(id: string, text: string): ProviderSubagentInputEvent[] {
-  const mapper = new PiHistoryMapper("pi");
-  const events: ProviderSubagentInputEvent[] = [];
-  for (const line of text.split("\n")) {
-    if (!line || events.length >= MAX_ITEMS) break;
-    let entry: { message?: PiAgentMessage; timestamp?: string };
+/** Incrementally reads one append-only child transcript. A partial JSONL row is held until newline. */
+export class PiChildSessionFollower {
+  private offset = 0;
+  private pending = Buffer.alloc(0);
+  private readonly mapper = new PiHistoryMapper("pi");
+  private items = 0;
+  private readonly maxBytes: number;
+
+  constructor(
+    private readonly id: string,
+    private readonly file: string,
+    maxBytes = MAX_BYTES,
+  ) {
+    this.maxBytes = Math.min(maxBytes, MAX_BYTES);
+  }
+
+  async readNew(): Promise<ProviderSubagentInputEvent[]> {
+    if (this.offset >= this.maxBytes || this.items >= MAX_ITEMS) return [];
     try {
-      entry = JSON.parse(line);
+      const handle = await open(this.file, "r");
+      try {
+        const size = Math.min((await handle.stat()).size, this.maxBytes) - this.offset;
+        if (size <= 0) return [];
+        const buffer = Buffer.alloc(size);
+        const { bytesRead } = await handle.read(buffer, 0, size, this.offset);
+        this.offset += bytesRead;
+        const data = Buffer.concat([this.pending, buffer.subarray(0, bytesRead)]);
+        const lastNewline = data.lastIndexOf(10);
+        if (lastNewline < 0) {
+          this.pending = data;
+          return [];
+        }
+        this.pending = data.subarray(lastNewline + 1);
+        const events: ProviderSubagentInputEvent[] = [];
+        for (const line of data.subarray(0, lastNewline).toString("utf8").split("\n")) {
+          if (this.items >= MAX_ITEMS) break;
+          const mapped = mapChildLine(this.mapper, this.id, line).slice(0, MAX_ITEMS - this.items);
+          this.items += mapped.length;
+          events.push(...mapped);
+        }
+        return events;
+      } finally {
+        await handle.close();
+      }
     } catch {
-      continue;
-    }
-    if (!entry.message || typeof entry.message !== "object" || !("role" in entry.message)) continue;
-    for (const mapped of mapper.mapMessages([entry.message])) {
-      if (mapped.type !== "timeline") continue;
-      events.push({
-        type: "timeline",
-        id,
-        item: limitAgentTimelineItemContent(mapped.item),
-        ...(entry.timestamp ? { timestamp: entry.timestamp } : {}),
-      });
-      if (events.length >= MAX_ITEMS) break;
+      return [];
     }
   }
-  return events;
 }

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

@@ -22,13 +22,21 @@ export interface PiExtensionToolMapping {
   detail?: ToolCallDetail;
   timeline?: AgentTimelineItem[];
   subagents?: ProviderSubagentInputEvent[];
-  /** Completed Pi child sessions to hydrate through the normal Pi history mapper. */
-  childSessions?: Array<{ id: string; file: string }>;
+  childSessions?: PiChildSessionFile[];
 }
 
 export interface PiExtensionCustomMapping {
   subagents: ProviderSubagentInputEvent[];
-  childSessions?: Array<{ id: string; file: string }>;
+  childSessions?: PiChildSessionFile[];
+}
+
+/**
+ * A Pi child's append-only session file, mapped through the normal Pi history mapper. Report it as
+ * soon as it is known; a live session follows it until the child's upsert reaches a final status.
+ */
+export interface PiChildSessionFile {
+  id: string;
+  file: string;
 }
 
 export type PiExtensionDialog = Extract<PiRuntimeEvent, { type: "extension_ui_request" }>;
@@ -48,6 +56,10 @@ export interface PiExtensionUiReply {
 
 export interface PiExtensionSession {
   mapToolCall?(call: PiExtensionToolCall): PiExtensionToolMapping | undefined;
+  /** Claims a `ctx.ui.notify` message sent by this extension's `runtimeBridge`. */
+  mapRuntimeNotification?(message: string): PiExtensionCustomMapping | undefined;
+  /** Called on the live follow cadence to report state Pi only writes to disk. */
+  poll?(): PiExtensionCustomMapping | undefined;
   mapCustomMessage?(
     message: Extract<PiAgentMessage, { role: "custom" }>,
   ): PiExtensionCustomMapping | undefined;
@@ -62,5 +74,10 @@ export interface PiExtensionSession {
 
 export interface PiExtension {
   id: string;
+  /**
+   * JavaScript statements run inside Paseo's Pi integration extension, with `pi` in scope. Reports
+   * state that Pi's RPC stream omits through `ctx.ui.notify`, which reaches `mapRuntimeNotification`.
+   */
+  runtimeBridge?: string;
   createSession(): PiExtensionSession;
 }

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

@@ -3,8 +3,98 @@ import { fileURLToPath } from "node:url";
 import { createPiExtensionHost } from "../index.js";
 import { readSubagentFixture, verifySubagentFixture } from "../subagent-fixture-test.js";
 import { streamPiHistory } from "../../history-mapper.js";
+import { GOTGENES_CHILD_SESSION_MARKER, gotgenesRuntimeBridge } from "./runtime-bridge.js";
+
+const SERVICE_KEY = Symbol.for("@gotgenes/pi-subagents:service");
+
+function loadRuntimeBridge() {
+  const listeners = new Map<string, (event: unknown, ctx?: unknown) => void>();
+  const pi = {
+    on: (name: string, listener: (event: unknown, ctx?: unknown) => void) =>
+      listeners.set(name, listener),
+    events: {
+      on: (name: string, listener: (event: unknown) => void) => listeners.set(name, listener),
+    },
+  };
+  new Function("pi", gotgenesRuntimeBridge)(pi);
+  return (name: string, event: unknown, ctx?: unknown) => listeners.get(name)?.(event, ctx);
+}
+
+function publishService(agents: Array<{ id: string; outputFile?: string }>): void {
+  (globalThis as Record<symbol, unknown>)[SERVICE_KEY] = { listAgents: () => agents };
+}
 
 describe("@gotgenes/pi-subagents adapter", () => {
+  test("reports a child's file after the child republishes the gotgenes service", async () => {
+    const notifications: string[] = [];
+    const emit = loadRuntimeBridge();
+    const file = "/sessions/parent/tasks/2026-09-30_child-session.jsonl";
+    try {
+      publishService([{ id: "native-1", outputFile: file }]);
+      emit(
+        "session_start",
+        {},
+        { ui: { notify: (message: string) => notifications.push(message) } },
+      );
+      // Pi loads gotgenes again inside the child, and that copy replaces the global service.
+      publishService([]);
+      emit("subagents:child:session-created", { sessionId: "child-session" });
+
+      await expect.poll(() => notifications).toHaveLength(1);
+      const host = createPiExtensionHost();
+      host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { subagent_type: "Explore", prompt: "Inspect" },
+        status: "completed",
+        result: { details: { agentId: "native-1", status: "background" } },
+      });
+      expect(host.mapRuntimeNotification(notifications[0]!)?.childSessions).toEqual([
+        { id: "call-1", file },
+      ]);
+    } finally {
+      delete (globalThis as Record<symbol, unknown>)[SERVICE_KEY];
+    }
+  });
+
+  test("accepts a live child path before or after the spawn result", () => {
+    for (const early of [true, false]) {
+      const host = createPiExtensionHost();
+      const marker = `${GOTGENES_CHILD_SESSION_MARKER} ${JSON.stringify({ agentId: "native-1", file: "/tmp/child.jsonl" })}`;
+      if (early) expect(host.mapRuntimeNotification(marker)?.subagents).toEqual([]);
+      const spawn = host.mapToolCall({
+        callId: "call-1",
+        toolName: "subagent",
+        args: { subagent_type: "Explore", prompt: "Inspect" },
+        status: "completed",
+        result: { details: { agentId: "native-1", status: "background" } },
+      });
+      const path = early
+        ? spawn?.childSessions
+        : host.mapRuntimeNotification(marker)?.childSessions;
+      expect(path).toEqual([{ id: "call-1", file: "/tmp/child.jsonl" }]);
+    }
+  });
+
+  test("accepts a running subagent-update without a status field", () => {
+    const host = createPiExtensionHost();
+    host.mapToolCall({
+      callId: "call-1",
+      toolName: "subagent",
+      args: { subagent_type: "Explore", prompt: "Inspect" },
+      status: "completed",
+      result: { details: { agentId: "native-1", status: "background" } },
+    });
+    const update = host.mapCustomMessage({
+      role: "custom",
+      customType: "subagent-update",
+      content: "progress",
+      details: { id: "native-1", description: "Inspect", message: "progress" },
+    });
+    expect(update?.subagents).toEqual([
+      { type: "upsert", id: "call-1", description: "Inspect", status: "running" },
+    ]);
+  });
   test("maps captured foreground lifecycle live and on replay", async () => {
     const events = await verifySubagentFixture(
       readSubagentFixture(new URL("./fixtures/foreground.json", import.meta.url)),

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

@@ -1,5 +1,7 @@
 import { z } from "zod";
 import { extractTextFromToolResult } from "../../tool-call-mapper.js";
+import { outputFileFromToolResult } from "../child-session.js";
+import { GOTGENES_CHILD_SESSION_MARKER, gotgenesRuntimeBridge } from "./runtime-bridge.js";
 import type { PiExtension, PiExtensionToolCall } from "../contract.js";
 
 const SpawnArgs = z
@@ -22,11 +24,15 @@ const Details = z
 const Notification = z
   .object({
     id: z.string().trim().min(1),
-    status: z.string(),
+    status: z.string().optional(),
     description: z.string().optional(),
     outputFile: z.string().trim().min(1).optional(),
   })
   .passthrough();
+const LiveSession = z.object({
+  agentId: z.string().trim().min(1),
+  file: z.string().trim().min(1),
+});
 const status = (value: string): "running" | "completed" | "failed" | "canceled" => {
   if (value === "completed") return "completed";
   if (value === "error") return "failed";
@@ -36,9 +42,16 @@ const status = (value: string): "running" | "completed" | "failed" | "canceled"
 
 export const gotgenesPiSubagents: PiExtension = {
   id: "@gotgenes/pi-subagents",
+  runtimeBridge: gotgenesRuntimeBridge,
   createSession: () => {
     const callsByAgent = new Map<string, string>();
     const readSessions = new Set<string>();
+    const pendingFilesByAgent = new Map<string, string>();
+    const childSession = (id: string, file: string | undefined) => {
+      if (!file || readSessions.has(file)) return [];
+      readSessions.add(file);
+      return [{ id, file }];
+    };
     const mapSpawn = (call: PiExtensionToolCall) => {
       const args = SpawnArgs.safeParse(call.args);
       if (!args.success) return undefined;
@@ -75,6 +88,10 @@ export const gotgenesPiSubagents: PiExtension = {
             }
           : { detail };
       callsByAgent.set(details.data.agentId, call.callId);
+      const file =
+        outputFileFromToolResult(call.result) ?? pendingFilesByAgent.get(details.data.agentId);
+      pendingFilesByAgent.delete(details.data.agentId);
+      const childSessions = childSession(call.callId, file);
       return {
         detail,
         subagents: [
@@ -87,6 +104,7 @@ export const gotgenesPiSubagents: PiExtension = {
             status: status(details.data.status),
           },
         ],
+        childSessions,
       };
     };
     const mapFollowup = (call: PiExtensionToolCall) => {
@@ -104,11 +122,7 @@ export const gotgenesPiSubagents: PiExtension = {
       };
       if (call.status === "running" || !details.success) return { detail };
       const file = details.data.transcriptPath;
-      const childSessions =
-        file && status(details.data.status) !== "running" && !readSessions.has(file)
-          ? [{ id, file }]
-          : [];
-      if (childSessions.length && file) readSessions.add(file);
+      const childSessions = childSession(id, file);
       return {
         detail,
         subagents: [{ type: "upsert" as const, id, status: status(details.data.status) }],
@@ -135,23 +149,41 @@ export const gotgenesPiSubagents: PiExtension = {
         const id = callsByAgent.get(details.data.id);
         if (!id) return undefined;
         const file = details.data.outputFile;
-        const childSessions =
-          file && status(details.data.status) !== "running" && !readSessions.has(file)
-            ? [{ id, file }]
-            : [];
-        if (childSessions.length && file) readSessions.add(file);
+        const childSessions = childSession(id, file);
         return {
           subagents: [
             {
               type: "upsert",
               id,
               description: details.data.description,
-              status: status(details.data.status),
+              status: details.data.status ? status(details.data.status) : "running",
             },
           ],
           childSessions,
         };
       },
+      mapRuntimeNotification(message) {
+        const prefix = `${GOTGENES_CHILD_SESSION_MARKER} `;
+        if (!message.startsWith(prefix)) return undefined;
+        let raw: unknown;
+        try {
+          raw = JSON.parse(message.slice(prefix.length));
+        } catch {
+          return { subagents: [] };
+        }
+        const parsed = LiveSession.safeParse(raw);
+        if (!parsed.success) return { subagents: [] };
+        const { agentId, file } = parsed.data;
+        const id = callsByAgent.get(agentId);
+        if (!id) {
+          pendingFilesByAgent.set(agentId, file);
+          return { subagents: [] };
+        }
+        return {
+          subagents: [{ type: "upsert", id, status: "running" }],
+          childSessions: childSession(id, file),
+        };
+      },
     };
   },
 };

packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/runtime-bridge.ts

@@ -0,0 +1,33 @@
+export const GOTGENES_CHILD_SESSION_MARKER = "PASEO_GOTGENES_CHILD_SESSION";
+
+/**
+ * Runs inside Pi. The spawn result omits the child's session file, so the parent's public service
+ * is the live source. Each child session loads its own copy of gotgenes, which republishes the
+ * global service and deletes it on exit, so the parent's service is captured at session_start.
+ */
+export const gotgenesRuntimeBridge = `
+  let gotgenesParent;
+  pi.on("session_start", (_event, ctx) => {
+    gotgenesParent = { ctx, service: globalThis[Symbol.for("@gotgenes/pi-subagents:service")] };
+  });
+  pi.events.on("subagents:child:session-created", (event) => {
+    const parent = gotgenesParent;
+    let attempts = 0;
+    // The record gets its session file after the child's extensions bind.
+    const report = () => {
+      const record = parent?.service?.listAgents().find((agent) =>
+        agent.outputFile?.includes(event.sessionId),
+      );
+      if (record) {
+        parent.ctx.ui.notify(
+          "${GOTGENES_CHILD_SESSION_MARKER} " +
+            JSON.stringify({ agentId: record.id, file: record.outputFile }),
+          "info",
+        );
+      } else if (++attempts < 240) {
+        setTimeout(report, 250);
+      }
+    };
+    setTimeout(report, 0);
+  });
+`;

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

@@ -5,7 +5,7 @@ import type {
   AgentStreamEvent,
 } from "../../../agent-sdk-types.js";
 import type { PiAgentMessage } from "../rpc-types.js";
-import { mapPiChildSession } from "./child-session.js";
+import { mapPiChildSession, PiChildSessionFollower } from "./child-session.js";
 import type {
   PiExtension,
   PiExtensionCustomMapping,
@@ -30,9 +30,19 @@ interface Session {
   adapter: PiExtensionSession;
 }
 
+interface Follower {
+  reader: PiChildSessionFollower;
+  /** Reads run in order so a final read cannot overtake a tick. */
+  pending: Promise<void>;
+}
+
 export class PiExtensionHost {
   private readonly sessions: Session[] = [];
   private remainingHydrationBytes: number;
+  private readonly followers = new Map<string, Follower>();
+  private onFollowEvent?: (event: AgentStreamEvent) => void;
+  private followTimer?: ReturnType<typeof setInterval>;
+  private pollingExtensions = false;
 
   constructor(
     extensions: readonly PiExtension[],
@@ -47,6 +57,56 @@ export class PiExtensionHost {
     }
   }
 
+  /** Live sessions follow child files and adapter polls; replay hosts read each file once. */
+  follow(onEvent: (event: AgentStreamEvent) => void): void {
+    this.onFollowEvent = onEvent;
+    this.followTimer ??= setInterval(() => {
+      for (const [id, follower] of this.followers) void this.readFollower(id, follower);
+      void this.pollExtensions();
+    }, 250);
+  }
+
+  close(): void {
+    if (this.followTimer) clearInterval(this.followTimer);
+    this.followTimer = undefined;
+    this.onFollowEvent = undefined;
+    this.followers.clear();
+  }
+
+  private readFollower(id: string, follower: Follower): Promise<void> {
+    follower.pending = follower.pending
+      .then(async () => {
+        const events = await follower.reader.readNew();
+        for (const event of events) {
+          this.onFollowEvent?.({ type: "provider_subagent", provider: "pi", event });
+        }
+        return undefined;
+      })
+      .catch((error) => {
+        this.logger?.warn(
+          { err: error, operation: "follow", childId: id },
+          "Pi child follow failed",
+        );
+      });
+    return follower.pending;
+  }
+
+  private async pollExtensions(): Promise<void> {
+    if (this.pollingExtensions || !this.onFollowEvent) return;
+    this.pollingExtensions = true;
+    try {
+      for (const { id, adapter } of this.sessions) {
+        const mapping = this.safe(id, "poll", () => adapter.poll?.());
+        if (!mapping) continue;
+        const output = this.prepare(id, mapping);
+        for (const event of output.events) this.onFollowEvent?.(event);
+        for (const event of await output.hydration) this.onFollowEvent?.(event);
+      }
+    } finally {
+      this.pollingExtensions = false;
+    }
+  }
+
   private safe<T>(id: string, operation: string, call: () => T): T | undefined {
     try {
       return call();
@@ -80,6 +140,19 @@ export class PiExtensionHost {
     return undefined;
   }
 
+  mapRuntimeNotification(message: string): PiExtensionOutput<PiExtensionCustomMapping> | undefined {
+    for (const { id, adapter } of this.sessions) {
+      const mapping = this.safe(id, "mapRuntimeNotification", () =>
+        adapter.mapRuntimeNotification?.(message),
+      );
+      if (mapping) {
+        const prepared = this.safe(id, "prepare", () => this.prepare(id, mapping));
+        if (prepared) return prepared;
+      }
+    }
+    return undefined;
+  }
+
   private prepare<T extends PiExtensionToolMapping | PiExtensionCustomMapping>(
     id: string,
     mapping: T,
@@ -92,25 +165,57 @@ export class PiExtensionHost {
         (event): AgentStreamEvent => ({ type: "provider_subagent", provider: "pi", event }),
       ),
     ];
-    const hydration = Promise.all(
-      (mapping.childSessions ?? []).map(async ({ id: childId, file }) => {
+    const hydration = (
+      this.onFollowEvent ? this.followChildren(mapping) : this.readChildren(mapping)
+    ).catch((error): AgentStreamEvent[] => {
+      this.logger?.warn(
+        { err: error, extensionId: id, operation: "hydrate" },
+        "Pi extension adapter failed",
+      );
+      return [];
+    });
+    return { ...mapping, events, hydration };
+  }
+
+  private async followChildren(
+    mapping: PiExtensionToolMapping | PiExtensionCustomMapping,
+  ): Promise<AgentStreamEvent[]> {
+    for (const { id, file } of mapping.childSessions ?? []) {
+      if (!this.followers.has(id)) {
+        this.followers.set(id, {
+          reader: new PiChildSessionFollower(id, file),
+          pending: Promise.resolve(),
+        });
+      }
+    }
+    const finished = (mapping.subagents ?? []).flatMap((event) =>
+      event.type === "upsert" && event.status && event.status !== "running" ? [event.id] : [],
+    );
+    const reads = [
+      ...new Set([...(mapping.childSessions ?? []).map(({ id }) => id), ...finished]),
+    ].flatMap((id) => {
+      const follower = this.followers.get(id);
+      return follower ? [this.readFollower(id, follower)] : [];
+    });
+    for (const id of finished) this.followers.delete(id);
+    await Promise.all(reads);
+    return [];
+  }
+
+  private async readChildren(
+    mapping: PiExtensionToolMapping | PiExtensionCustomMapping,
+  ): Promise<AgentStreamEvent[]> {
+    const groups = await Promise.all(
+      (mapping.childSessions ?? []).map(async ({ id, file }) => {
         const bytes = Math.min(this.remainingHydrationBytes, 2 * 1024 * 1024);
         if (bytes <= 0) return [];
         this.remainingHydrationBytes -= bytes;
-        return (await this.readChildSession(childId, file, bytes)).map(
+        return (await this.readChildSession(id, file, bytes)).map(
           (event): AgentStreamEvent => ({ type: "provider_subagent", provider: "pi", event }),
         );
       }),
-    )
-      .then((groups) => groups.flat())
-      .catch((error): AgentStreamEvent[] => {
-        this.logger?.warn(
-          { err: error, extensionId: id, operation: "hydrate" },
-          "Pi extension adapter failed",
-        );
-        return [];
-      });
-    return { ...mapping, events, hydration };
+    );
+    return groups.flat();
   }
 
   onToolStart(call: PiExtensionToolCall, provider = "pi"): AgentPermissionRequest | undefined {

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

@@ -1,4 +1,7 @@
 import { describe, expect, test } from "vitest";
+import { mkdtemp, appendFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { createPiExtensionHost } from "./index.js";
 import { PiHistoryMapper } from "../history-mapper.js";
 import { parseToolArgs, parseToolResult } from "../tool-call-mapper.js";
@@ -30,6 +33,53 @@ const throwingAdapter: PiExtension = {
 };
 
 describe("Pi extension host", () => {
+  test("streams appended child rows before the child completes", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-live-child-"));
+    const file = join(dir, "child.jsonl");
+    const received: string[] = [];
+    const host = new PiExtensionHost([
+      {
+        id: "child",
+        createSession: () => ({
+          mapToolCall: () => ({
+            subagents: [{ type: "upsert", id: "child-1", status: "running" }],
+            childSessions: [{ id: "child-1", file }],
+          }),
+        }),
+      },
+    ]);
+    try {
+      host.follow((event) => {
+        if (event.type === "provider_subagent" && event.event.type === "timeline") {
+          received.push(JSON.stringify(event.event.item));
+        }
+      });
+      host.mapToolCall({
+        callId: "x",
+        toolName: "child",
+        args: {},
+        status: "completed",
+        result: null,
+      });
+      await appendFile(
+        file,
+        '{"type":"message","message":{"role":"assistant","content":[{"type":"text","text":"BEGIN"}]}}\n',
+      );
+      await expect.poll(() => received.length).toBe(1);
+      await appendFile(
+        file,
+        '{"type":"message","message":{"role":"assistant","content":[{"type":"text","text":"DO',
+      );
+      await new Promise((resolve) => setTimeout(resolve, 350));
+      expect(received).toHaveLength(1);
+      await appendFile(file, 'NE"}]}}\n');
+      await expect.poll(() => received.length).toBe(2);
+      expect(received.join(" ")).toContain("DONE");
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
   test("declines a throwing adapter and preserves generic history", async () => {
     const host = createPiExtensionHost(undefined, [throwingAdapter]);
     const mapper = new PiHistoryMapper("pi", [], {}, host);

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

@@ -3,6 +3,10 @@ import { piExtensions } from "./registry.js";
 import type { Logger } from "pino";
 import type { PiExtension } from "./contract.js";
 
+export const piExtensionRuntimeBridge = piExtensions
+  .flatMap((extension) => extension.runtimeBridge ?? [])
+  .join("\n");
+
 export type { PiExtensionHost } from "./host.js";
 export type { PiExtensionEventOutput } from "./host.js";
 

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

@@ -1,5 +1,8 @@
 import { describe, expect, test } from "vitest";
 import { fileURLToPath } from "node:url";
+import { mkdtemp, writeFile, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
 import { createPiExtensionHost } from "../index.js";
 import {
   parseToolArgs,
@@ -27,7 +30,100 @@ function mapToolDetail(toolCall: PiTrackedToolCall, result: PiToolResult) {
   return mapping(toolCall, result)?.detail;
 }
 
+function hasChildTimeline(events: AgentStreamEvent[]) {
+  return events.some(
+    (event) => event.type === "provider_subagent" && event.event.type === "timeline",
+  );
+}
+
+function hasCompletedChild(events: AgentStreamEvent[]) {
+  return events.some(
+    (event) =>
+      event.type === "provider_subagent" &&
+      event.event.type === "upsert" &&
+      event.event.status === "completed",
+  );
+}
+
 describe("pi-subagents adapter", () => {
+  test("follows async status and transcript without bg_wait", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
+    const file = join(dir, "child.jsonl");
+    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(
+        file,
+        '{"type":"message","message":{"role":"assistant","content":[{"type":"text","text":"BEGIN"}]}}\n',
+      );
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "running",
+          steps: [{ agent: "scout", status: "running", sessionFile: file }],
+        }),
+      );
+      await expect.poll(() => hasChildTimeline(received)).toBe(true);
+      await writeFile(
+        join(dir, "status.json"),
+        JSON.stringify({
+          state: "complete",
+          steps: [{ agent: "scout", status: "complete", sessionFile: file }],
+        }),
+      );
+      await expect.poll(() => hasCompletedChild(received)).toBe(true);
+    } finally {
+      host.close();
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
+  test("settles an async run and reads its transcript from the completion notice on replay", async () => {
+    const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
+    const file = join(dir, "child.jsonl");
+    try {
+      await writeFile(
+        file,
+        '{"type":"message","message":{"role":"assistant","content":[{"type":"text","text":"DONE"}]}}\n',
+      );
+      const host = createPiExtensionHost();
+      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: [] },
+        },
+      });
+      const notice = host.mapCustomMessage({
+        role: "custom",
+        customType: "subagent-notify",
+        content: [
+          "Background task completed: **scout**",
+          "",
+          "Session file: /tmp/not-the-session-line.jsonl",
+          "",
+          `Retention-managed async directory: ${dir}`,
+          "",
+          `Session file: ${file}`,
+        ].join("\n"),
+      });
+      expect(notice?.subagents).toEqual([{ type: "upsert", id: "call-1", status: "completed" }]);
+      expect(hasChildTimeline(await notice!.hydration)).toBe(true);
+    } finally {
+      await rm(dir, { recursive: true, force: true });
+    }
+  });
   test("maps completed subagent calls with task input to sub-agent detail", () => {
     const toolCall = parseToolArgs("subagent", {
       agent: "reviewer",
@@ -74,11 +170,11 @@ describe("pi-subagents adapter", () => {
     expect(events.filter((event) => event.event.type === "timeline").length).toBeGreaterThan(0);
   });
 
-  test("keeps captured async runs running until structured completion", async () => {
+  test("settles captured async runs from the completion notice live and on replay", async () => {
     const fixture = readSubagentFixture(new URL("./fixtures/background.json", import.meta.url));
     const events = await verifySubagentFixture(fixture);
     expect(events.findLast((event) => event.event.type === "upsert")?.event).toEqual(
-      expect.objectContaining({ status: "running" }),
+      expect.objectContaining({ status: "completed" }),
     );
   });
 
@@ -102,7 +198,13 @@ describe("pi-subagents adapter", () => {
     const upserts = events
       .filter((event) => event.event.type === "upsert")
       .map((event) => (event.event.type === "upsert" ? event.event : null));
-    expect(upserts.map((event) => event?.status)).toEqual(["running", "running", "completed"]);
+    // bg_wait settles the child; the later completion notice repeats it.
+    expect(upserts.map((event) => event?.status)).toEqual([
+      "running",
+      "running",
+      "completed",
+      "completed",
+    ]);
     expect(new Set(upserts.map((event) => event?.id)).size).toBe(1);
     expect(events.filter((event) => event.event.type === "timeline").length).toBeGreaterThan(0);
   });

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

@@ -1,4 +1,6 @@
 import { z } from "zod";
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
 import { extractTextFromToolResult } from "../../tool-call-mapper.js";
 import type { PiExtension, PiExtensionToolCall, PiExtensionToolMapping } from "../contract.js";
 
@@ -27,16 +29,55 @@ const Details = z
     mode: z.string(),
     runId: z.string().optional(),
     asyncId: z.string().optional(),
+    asyncDir: z.string().trim().min(1).optional(),
     results: z.array(Row),
     completions: z.array(Completion).optional(),
   })
   .passthrough();
+const AsyncStatus = z
+  .object({
+    state: z.string(),
+    steps: z.array(
+      z
+        .object({
+          agent: z.string().trim().min(1),
+          status: z.string().optional(),
+          sessionFile: z.string().trim().min(1).optional(),
+        })
+        .passthrough(),
+    ),
+  })
+  .passthrough();
+
+function asyncChildStatus(runState: string, stepStatus?: string) {
+  if (!["complete", "failed", "stopped"].includes(runState)) return "running";
+  if (runState === "stopped" || 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. */
+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): \*\*/,
+  )?.[1];
+  if (outcome === "stopped") return "canceled";
+  return outcome as "completed" | "failed" | undefined;
+}
 
 export const piSubagents: PiExtension = {
   id: "pi-subagents",
   createSession: () => {
     const callsByRun = new Map<string, string>();
     const readSessions = new Set<string>();
+    /** Keyed by async directory, which both status.json and the completion notice carry. */
+    const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
+    const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
+      if (details.asyncDir && call.status !== "failed") {
+        asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true });
+      }
+    };
     const collectRows = (
       owner: string,
       rows: z.infer<typeof Row>[],
@@ -109,7 +150,8 @@ export const piSubagents: PiExtension = {
           ? { detail, subagents: [{ ...base, status: "failed" as const }] }
           : { detail };
       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 && (details.data.asyncId || args.data.async)) {
+        rememberAsyncRun(details.data, call);
         return {
           detail,
           subagents: [
@@ -119,6 +161,7 @@ export const piSubagents: PiExtension = {
             },
           ],
         };
+      }
       if (details.data.results.length === 0 && call.status === "failed")
         return { detail, subagents: [{ ...base, status: "failed" as const }] };
       return {
@@ -132,6 +175,73 @@ export const piSubagents: PiExtension = {
         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-notify") return undefined;
+        const text =
+          typeof message.content === "string"
+            ? message.content
+            : extractTextFromToolResult({ content: message.content });
+        const lines = text?.split("\n") ?? [];
+        const status = noticeStatus(lines[0] ?? "");
+        if (!status) return undefined;
+        const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
+        const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
+        // Each run's session line follows its async directory line; earlier lines are child output.
+        let owner: string | undefined;
+        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 });
+            continue;
+          }
+          const file = owner ? line.match(/^Session file: (.+)$/)?.[1] : undefined;
+          if (!owner || !file) continue;
+          if (!readSessions.has(file)) {
+            readSessions.add(file);
+            childSessions.push({ id: owner, file });
+          }
+          owner = undefined;
+        }
+        return subagents.length ? { subagents, childSessions } : undefined;
+      },
+      poll() {
+        const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
+        const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
+        for (const [dir, run] of asyncRuns) {
+          if (!run.polling) continue;
+          let raw: unknown;
+          try {
+            raw = JSON.parse(readFileSync(join(dir, "status.json"), "utf8"));
+          } catch {
+            continue;
+          }
+          const parsed = AsyncStatus.safeParse(raw);
+          if (!parsed.success) continue;
+          const terminal = ["complete", "failed", "stopped"].includes(parsed.data.state);
+          for (const [index, step] of parsed.data.steps.entries()) {
+            const id = parsed.data.steps.length === 1 ? run.owner : `${run.owner}:${index}`;
+            const status = asyncChildStatus(parsed.data.state, step.status);
+            subagents.push({ type: "upsert", id, title: step.agent, status });
+            if (step.sessionFile && !readSessions.has(step.sessionFile)) {
+              readSessions.add(step.sessionFile);
+              childSessions.push({ id, file: step.sessionFile });
+            }
+          }
+          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;
+          }
+        }
+        return subagents.length || childSessions.length ? { subagents, childSessions } : undefined;
+      },
     };
   },
 };

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

@@ -4,6 +4,24 @@ import { createPiExtensionHost } from "../index.js";
 import { readSubagentFixture, verifySubagentFixture } from "../subagent-fixture-test.js";
 
 describe("@tintinweb/pi-subagents adapter", () => {
+  test("exposes the background output file while the child is running", () => {
+    const mapping = createPiExtensionHost().mapToolCall({
+      callId: "call-1",
+      toolName: "Agent",
+      args: { subagent_type: "Explore", prompt: "Inspect" },
+      status: "completed",
+      result: {
+        details: { agentId: "native-1", status: "background" },
+        content: [
+          { type: "text", text: "Agent started in background.\nOutput file: /tmp/child.output\n" },
+        ],
+      },
+    });
+    expect(mapping?.subagents).toEqual([
+      expect.objectContaining({ id: "call-1", status: "running" }),
+    ]);
+    expect(mapping?.childSessions).toEqual([{ id: "call-1", file: "/tmp/child.output" }]);
+  });
   test("maps captured foreground lifecycle live and on replay", async () => {
     const events = await verifySubagentFixture(
       readSubagentFixture(new URL("./fixtures/foreground.json", import.meta.url)),

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

@@ -1,5 +1,6 @@
 import { z } from "zod";
 import { extractTextFromToolResult } from "../../tool-call-mapper.js";
+import { outputFileFromToolResult } from "../child-session.js";
 import type { PiExtension, PiExtensionToolCall } from "../contract.js";
 
 const SpawnArgs = z
@@ -74,6 +75,9 @@ export const tintinwebPiSubagents: PiExtension = {
             }
           : { detail };
       callsByAgent.set(details.data.agentId, call.callId);
+      const file = outputFileFromToolResult(call.result);
+      const childSessions = file && !readSessions.has(file) ? [{ id: call.callId, file }] : [];
+      if (file) readSessions.add(file);
       return {
         detail,
         subagents: [
@@ -86,6 +90,7 @@ export const tintinwebPiSubagents: PiExtension = {
             status: status(details.data.status),
           },
         ],
+        childSessions,
       };
     };
     const mapFollowup = (call: PiExtensionToolCall) => {
@@ -125,10 +130,7 @@ export const tintinwebPiSubagents: PiExtension = {
         const id = callsByAgent.get(details.data.id);
         if (!id) return undefined;
         const file = details.data.outputFile;
-        const childSessions =
-          file && status(details.data.status) !== "running" && !readSessions.has(file)
-            ? [{ id, file }]
-            : [];
+        const childSessions = file && !readSessions.has(file) ? [{ id, file }] : [];
         if (childSessions.length && file) readSessions.add(file);
         return {
           subagents: [