From 9ae7a64aa71f6464350ca8a907a59510c15c2649 Mon Sep 17 00:00:00 2001 From: Kushal Bairagya Date: Sat, 17 Oct 2026 16:40:09 +0530 Subject: [PATCH] fix(pi): surface workflowScript subagents and finalize stuck rows --- docs/agent-lifecycle.md | 2 +- .../agent-manager.provider-subagents.test.ts | 658 +++++++++++++++ .../server/src/server/agent/agent-manager.ts | 184 ++++- .../agent/provider-subagents/store.test.ts | 17 + .../server/agent/provider-subagents/store.ts | 9 + .../pi-subagents/fixtures/workflow.json | 153 ++++ .../pi/extensions/pi-subagents/index.test.ts | 760 +++++++++++++++++- .../pi/extensions/pi-subagents/index.ts | 393 +++++++-- 8 files changed, 2093 insertions(+), 83 deletions(-) create mode 100644 packages/server/src/server/agent/agent-manager.provider-subagents.test.ts create mode 100644 packages/server/src/server/agent/providers/pi/extensions/pi-subagents/fixtures/workflow.json diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 378a0070ea..f5fbda4d48 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -210,7 +210,7 @@ The rows combine two kinds of children: parentAgentId === thisAgent.id AND !archivedAt ``` -- **Provider subagents** are child executions owned by Claude, Codex, or OpenCode. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent. +- **Provider subagents** are child executions owned by Claude, Codex, OpenCode, or Pi. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent. Pi rows that a poll keeps alive carry `liveness: "poll"`, and only a marked row can be settled by a sweep once its poll stops reporting — that gate is what stops an idle parent from leaving a card spinning and keeps other providers' rows untouched. Unmarked rows and every other provider settle their own rows. Clicking either kind opens a workspace tab. On compact layouts it opens in a [sheet](explorer-sidebar.md#sheets) over the subagents list instead; the sheet hides the composer and shows the agent's model and thinking as one line in its place (`PaneSubtitle`, the row provider subagents already use), and a Paseo subagent's sheet offers **Open as tab** (the full-screen glyph in the header) to get the composer back. A Paseo subagent tab is a normal interactive agent pane. A provider subagent tab is a read-only timeline pane with no composer, archive, detach, rewind, or fork actions. It shows its own direct children in a subagents track. Both panes use `AgentStreamView`, so message, reasoning, tool-call, and layout rendering stay identical. diff --git a/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts b/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts new file mode 100644 index 0000000000..35e66fa739 --- /dev/null +++ b/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts @@ -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 { + 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 { + 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 {} + + getPendingPermissions() { + return []; + } + + async respondToPermission(): Promise {} + + describePersistence(): AgentPersistenceHandle { + return { provider: this.provider, sessionId: this.id }; + } + + async interrupt(): Promise {} + + async close(): Promise {} +} + +class TestAgentClient implements AgentClient { + readonly capabilities = TEST_CAPABILITIES; + session?: TestAgentSession; + + constructor(readonly provider: AgentProvider) {} + + async createSession(config: AgentSessionConfig): Promise { + this.session = new TestAgentSession(config.provider); + return this.session; + } + + async resumeSession(): Promise { + this.session ??= new TestAgentSession(this.provider); + return this.session; + } + + async fetchCatalog(): Promise { + return { + models: [{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true }], + modes: [], + }; + } + + async isAvailable(): Promise { + return true; + } +} + +interface Harness { + manager: AgentManager; + session: TestAgentSession; + agentId: string; + cleanup: () => void; +} + +async function createHarness(provider: AgentProvider): Promise { + 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 { + for (let index = 0; index < 10_000; index++) await Promise.resolve(); +} + +async function pushEvent(harness: Harness, event: AgentStreamEvent): Promise { + 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 { + 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(); + } + }); +}); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index ff91ef85ef..05fe98cd02 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -726,6 +726,17 @@ function detachedAgentLabelPatch(labels: Record): 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(); private readonly timelineStore = new InMemoryAgentTimelineStore(); private readonly providerSubagents = new ProviderSubagentStore(); + private readonly providerSubagentSweepTimers = new Map>(); private readonly agentsAwaitingInitialSnapshotPersist = new Set(); private readonly sessionEventTails = new Map>(); private readonly steerEventBarriers = new Map(); @@ -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, + ): 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, + preexistingRunningIds: ReadonlySet, + ): 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 { + 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(); 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(); + // 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) { diff --git a/packages/server/src/server/agent/provider-subagents/store.test.ts b/packages/server/src/server/agent/provider-subagents/store.test.ts index dc71c0f5c6..4496571908 100644 --- a/packages/server/src/server/agent/provider-subagents/store.test.ts +++ b/packages/server/src/server/agent/provider-subagents/store.test.ts @@ -60,6 +60,23 @@ describe("ProviderSubagentStore", () => { expect(subagents.list("parent-b")).toHaveLength(1); }); + test("keeps a poll liveness marker across presentation-only upserts", () => { + const subagents = new ProviderSubagentStore(); + subagents.apply("parent-a", "pi", { + type: "upsert", + id: "child-1", + status: "running", + liveness: "poll", + }); + subagents.apply("parent-a", "pi", { type: "upsert", id: "child-1", title: "scout" }); + + expect(subagents.get("parent-a", "child-1")).toMatchObject({ + title: "scout", + status: "running", + liveness: "poll", + }); + }); + test("limits oversized provider child tool output before storage", () => { const subagents = new ProviderSubagentStore(); const output = "x".repeat(70 * 1024); diff --git a/packages/server/src/server/agent/provider-subagents/store.ts b/packages/server/src/server/agent/provider-subagents/store.ts index b04246054a..98862bdcb0 100644 --- a/packages/server/src/server/agent/provider-subagents/store.ts +++ b/packages/server/src/server/agent/provider-subagents/store.ts @@ -23,6 +23,11 @@ export interface ProviderSubagentDescriptor { toolCallId: string | null; cwd: string | null; subtitle: string | null; + /** + * Set when a live poll re-reports this row on every tick, so silence proves the poll is gone. + * Rows without it are event-driven and may stay quiet while their child still runs. + */ + liveness?: "poll"; } export type ProviderSubagentInputEvent = @@ -40,6 +45,8 @@ export type ProviderSubagentInputEvent = cwd?: string | null; subtitle?: string | null; parentSubagentId?: string | null; + /** Only a poll-backed reporter may set this; it is what makes a stale row sweepable. */ + liveness?: "poll"; timestamp?: string; } | { @@ -113,6 +120,7 @@ export class ProviderSubagentStore { this.timelines.initialize(key); } const timestamp = event.timestamp ?? new Date().toISOString(); + const liveness = event.liveness ?? previous?.liveness; const subagent: ProviderSubagentDescriptor = { id: event.id, parentAgentId, @@ -126,6 +134,7 @@ export class ProviderSubagentStore { cwd: stickyField(event.cwd, previous?.cwd), subtitle: stickyField(event.subtitle, previous?.subtitle), parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId), + ...(liveness === undefined ? {} : { liveness }), }; this.descriptors.set(key, subagent); return { type: "upsert", subagent }; diff --git a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/fixtures/workflow.json b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/fixtures/workflow.json new file mode 100644 index 0000000000..209c86102f --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/fixtures/workflow.json @@ -0,0 +1,153 @@ +{ + "provenance": { + "package": "pi-subagents", + "version": "0.78.0", + "sourceCommit": "reconstructed-workflow-launch", + "piVersion": "1.2.0", + "captureDate": "2026-10-17" + }, + "events": [ + { + "type": "tool_execution_start", + "toolCallId": "call_wf1", + "toolName": "subagent", + "args": { + "workflow": "tmp/rigor-ui-test.js", + "async": true + } + }, + { + "type": "tool_execution_end", + "toolCallId": "call_wf1", + "toolName": "subagent", + "result": { + "content": [ + { + "type": "text", + "text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)" + } + ], + "details": { + "mode": "workflow", + "runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "results": [], + "asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "workflowGraph": { + "lanes": [{ "key": "alpha", "agent": "scout" }] + }, + "context": "fresh", + "timeoutMs": 1800000, + "deadlineAt": 1790199499998 + } + }, + "isError": false + }, + { + "type": "message_end", + "message": { + "role": "custom", + "customType": "subagent-incremental-child-notify", + "content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running", + "display": false, + "details": { + "childKey": "alpha", + "outcome": "completed" + }, + "timestamp": 1790197711046 + } + }, + { + "type": "message_end", + "message": { + "role": "custom", + "customType": "subagent-notify", + "content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl", + "display": false, + "timestamp": 1790197722046 + } + } + ], + "messages": [ + { + "role": "assistant", + "content": [ + { + "type": "toolCall", + "id": "call_wf1", + "name": "subagent", + "arguments": { + "workflow": "tmp/rigor-ui-test.js", + "async": true + } + } + ], + "api": "openai-completions", + "provider": "openrouter", + "model": "google/gemini-3.6-flash", + "usage": { + "input": 3303, + "output": 114, + "cacheRead": 7053, + "cacheWrite": 0, + "reasoning": 81, + "totalTokens": 10470, + "cost": { + "input": 0.00247725, + "output": 0.00042750000000000004, + "cacheRead": 0.000528975, + "cacheWrite": 0, + "total": 0.003433725 + } + }, + "stopReason": "toolUse", + "timestamp": 1790197697608, + "responseId": "gen-1790197697-lxluaTrDkdaxjSXsO957", + "rawStopReason": "tool_calls" + }, + { + "role": "toolResult", + "toolCallId": "call_wf1", + "toolName": "subagent", + "content": [ + { + "type": "text", + "text": "Run fan-out: 1/64 used, 63 remaining\nAsync workflow: [eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1]\n\nThe async run is detached and running in the background.\nMission: 3d6d9bb6-1ea3-4b8d-9c56-6a1f0e3b6f88 (active)" + } + ], + "details": { + "mode": "workflow", + "runId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "results": [], + "asyncId": "eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "asyncDir": "/tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1", + "workflowGraph": { + "lanes": [{ "key": "alpha", "agent": "scout" }] + }, + "context": "fresh", + "timeoutMs": 1800000, + "deadlineAt": 1790199499998 + }, + "isError": false, + "timestamp": 1790197700011 + }, + { + "role": "custom", + "customType": "subagent-incremental-child-notify", + "content": "Workflow child completed: **alpha**\nWorkflow run: eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\nChild run: 4cb1c0a2-2f2e-4c31-9c53-0f0d2ad0f3f1\nOutput: /tmp/paseo-pi-extension-phase3/workspace/tmp/alpha.md\nStatus: workflow still running", + "display": false, + "details": { + "childKey": "alpha", + "outcome": "completed" + }, + "timestamp": 1790197711046 + }, + { + "role": "custom", + "customType": "subagent-notify", + "content": "Background task completed: **workflow**\n\nWorkflow receipt: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1/receipt.json\n\nRetention-managed async directory: /tmp/pi-subagents-uid-1000/async-subagent-runs/eac895df-6a71-4b4f-9f0a-2ba2b7f0f4a1\n\nSession file: /tmp/paseo-pi-extension-phase3/sessions-nico/2026-10-17T21-08-13-722Z_01a0d018/0f2c1b1e-9d1c-4b0e-8d5b-8e6f27a1c2d3/run-0/session.jsonl", + "display": false, + "timestamp": 1790197722046 + } + ] +} 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 8e9db54047..12eb85f66b 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,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" }), + ]); + }); }); 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 64e814af3f..2c84ff35d1 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 @@ -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; +type SubagentUpsert = Extract; + 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 | 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; +} + +/** + * 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(); const readSessions = new Set(); + /** Rows already reported terminal; a late give-up must not rewrite one of them. */ + const settledRows = new Set(); + /** Rows whose poll already reported the step's agent. A settled row is still named once. */ + const namedRows = new Set(); + 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(); + 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 }); + asyncRuns.set(details.asyncDir, { + owner: call.callId, + polling: true, + misses: 0, + deadline: Date.now() + POLL_TTL_MS, + rows: new Set(), + }); } }; + /** + * 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 = []; + 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 = [ + { 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[], @@ -87,17 +243,19 @@ export const piSubagents: PiExtension = { const subagents: NonNullable = []; const childSessions: NonNullable = []; 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 => { + 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 = []; @@ -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 = []; const childSessions: NonNullable = []; + 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;