From e9e479c6bac08ed1b9d74310c53cbcbca70aa48e Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 30 Sep 2026 13:56:22 +0100 Subject: [PATCH] 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 --- .../server/agent/providers/pi/agent.test.ts | 2 + .../src/server/agent/providers/pi/agent.ts | 16 ++- .../providers/pi/extensions/child-session.ts | 109 +++++++++----- .../agent/providers/pi/extensions/contract.ts | 23 ++- .../gotgenes-pi-subagents/index.test.ts | 90 ++++++++++++ .../extensions/gotgenes-pi-subagents/index.ts | 56 ++++++-- .../gotgenes-pi-subagents/runtime-bridge.ts | 33 +++++ .../agent/providers/pi/extensions/host.ts | 133 ++++++++++++++++-- .../providers/pi/extensions/index.test.ts | 50 +++++++ .../agent/providers/pi/extensions/index.ts | 4 + .../pi/extensions/pi-subagents/index.test.ts | 108 +++++++++++++- .../pi/extensions/pi-subagents/index.ts | 112 ++++++++++++++- .../tintinweb-pi-subagents/index.test.ts | 18 +++ .../tintinweb-pi-subagents/index.ts | 10 +- 14 files changed, 691 insertions(+), 73 deletions(-) create mode 100644 packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/runtime-bridge.ts diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index 969b221f02..0130574211 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/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; diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 5fd7b051ef..9a37c9adc7 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/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) || diff --git a/packages/server/src/server/agent/providers/pi/extensions/child-session.ts b/packages/server/src/server/agent/providers/pi/extensions/child-session.ts index d2dfa0e99b..2013b831be 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/child-session.ts +++ b/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 { + 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 { + 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; } diff --git a/packages/server/src/server/agent/providers/pi/extensions/contract.ts b/packages/server/src/server/agent/providers/pi/extensions/contract.ts index 7f9330eaeb..93129a0476 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/contract.ts +++ b/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; @@ -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, ): 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; } diff --git a/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.test.ts b/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.test.ts index 0e15438384..8e3d6e64e9 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.test.ts +++ b/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 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)[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)[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)), diff --git a/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.ts b/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.ts index bf7d2813fb..62db93c79e 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/index.ts +++ b/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(); const readSessions = new Set(); + const pendingFilesByAgent = new Map(); + 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), + }; + }, }; }, }; diff --git a/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/runtime-bridge.ts b/packages/server/src/server/agent/providers/pi/extensions/gotgenes-pi-subagents/runtime-bridge.ts new file mode 100644 index 0000000000..386fbb4705 --- /dev/null +++ b/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); + }); +`; diff --git a/packages/server/src/server/agent/providers/pi/extensions/host.ts b/packages/server/src/server/agent/providers/pi/extensions/host.ts index ae0f307393..8915e3df01 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/host.ts +++ b/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; +} + export class PiExtensionHost { private readonly sessions: Session[] = []; private remainingHydrationBytes: number; + private readonly followers = new Map(); + private onFollowEvent?: (event: AgentStreamEvent) => void; + private followTimer?: ReturnType; + 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 { + 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 { + 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(id: string, operation: string, call: () => T): T | undefined { try { return call(); @@ -80,6 +140,19 @@ export class PiExtensionHost { return undefined; } + mapRuntimeNotification(message: string): PiExtensionOutput | 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( 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 { + 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 { + 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 { diff --git a/packages/server/src/server/agent/providers/pi/extensions/index.test.ts b/packages/server/src/server/agent/providers/pi/extensions/index.test.ts index dd5d76601a..9afacae977 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/index.test.ts +++ b/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); diff --git a/packages/server/src/server/agent/providers/pi/extensions/index.ts b/packages/server/src/server/agent/providers/pi/extensions/index.ts index eb4ecda059..2ac68178da 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/index.ts +++ b/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"; diff --git a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts index 2f3293f8c1..8e9db54047 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts +++ b/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); }); diff --git a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts index 0ec0f0ab47..64e814af3f 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts +++ b/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(); const readSessions = new Set(); + /** Keyed by async directory, which both status.json and the completion notice carry. */ + const asyncRuns = new Map(); + const rememberAsyncRun = (details: z.infer, call: PiExtensionToolCall) => { + if (details.asyncDir && call.status !== "failed") { + asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true }); + } + }; const collectRows = ( owner: string, rows: z.infer[], @@ -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 = []; + const childSessions: NonNullable = []; + // 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 = []; + const childSessions: NonNullable = []; + 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; + }, }; }, }; diff --git a/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.test.ts b/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.test.ts index 62a8e5ea26..b464026ab1 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.test.ts +++ b/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)), diff --git a/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.ts b/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.ts index dcfe093328..2b06779afa 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/tintinweb-pi-subagents/index.ts +++ b/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: [