mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-20 00:02:54 +00:00
Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e26f0339d1 | ||
|
|
02216d601c | ||
|
|
e5f8c7ac25 | ||
|
|
b38dbf8ae6 | ||
|
|
1ea820ca17 | ||
|
|
9ae7a64aa7 |
@@ -210,7 +210,7 @@ The rows combine two kinds of children:
|
||||
parentAgentId === thisAgent.id AND !archivedAt
|
||||
```
|
||||
|
||||
- **Provider subagents** are child executions owned by Claude, Codex, or OpenCode. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
|
||||
- **Provider subagents** are child executions owned by Claude, Codex, OpenCode, or Pi. They are not inserted into `AgentManager` as managed agents. Providers emit a separate descriptor and timeline stream through `agent.provider_subagents.*`; the client keeps that state outside the normal agent store and merges only the presentation rows into the track. A descriptor's optional `parentSubagentId` identifies its direct provider-subagent parent; an absent value identifies a direct child of the managed agent.
|
||||
|
||||
Clicking either kind opens a workspace tab. On compact layouts it opens in a [sheet](explorer-sidebar.md#sheets) over the subagents list instead; the sheet hides the composer and shows the agent's model and thinking as one line in its place (`PaneSubtitle`, the row provider subagents already use), and a Paseo subagent's sheet offers **Open as tab** (the full-screen glyph in the header) to get the composer back. A Paseo subagent tab is a normal interactive agent pane. A provider subagent tab is a read-only timeline pane with no composer, archive, detach, rewind, or fork actions. It shows its own direct children in a subagents track. Both panes use `AgentStreamView`, so message, reasoning, tool-call, and layout rendering stay identical.
|
||||
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import pino from "pino";
|
||||
import { afterEach, describe, expect, test, vi } from "vitest";
|
||||
import { createTestLogger } from "../../test-utils/test-logger.js";
|
||||
import { AgentManager } from "./agent-manager.js";
|
||||
import { PiRpcAgentClient } from "./providers/pi/agent.js";
|
||||
import { FakePi } from "./providers/pi/test-utils/fake-pi.js";
|
||||
|
||||
const AGENT_ID = "00000000-0000-4000-8000-000000000001";
|
||||
async function drainSessionEvents(): Promise<void> {
|
||||
for (let index = 0; index < 10_000; index++) await Promise.resolve();
|
||||
}
|
||||
async function waitFor(condition: () => boolean): Promise<void> {
|
||||
for (let attempt = 0; attempt < 250; attempt++) {
|
||||
if (condition()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 20));
|
||||
}
|
||||
throw new Error("Timed out waiting for Pi subagent output");
|
||||
}
|
||||
afterEach(() => vi.useRealTimers());
|
||||
describe("provider subagent lifecycle ownership", () => {
|
||||
test(
|
||||
"a delayed parent turn end preserves a Pi child until its provider settles it",
|
||||
{ timeout: 15_000 },
|
||||
async () => {
|
||||
// Run the real Pi client and adapter with an injected runtime.
|
||||
const pi = new FakePi();
|
||||
const client = new PiRpcAgentClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
runtime: pi,
|
||||
// The injected runtime owns execution; use an existing executable for the real availability
|
||||
// probe so this test does not depend on a Pi installation or replace a client method.
|
||||
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
|
||||
});
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagent-poll-"));
|
||||
const manager = new AgentManager({
|
||||
clients: { pi: client },
|
||||
idFactory: () => AGENT_ID,
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
try {
|
||||
await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
const piSession = pi.latestSession();
|
||||
// An async spawn whose status directory never appears: every poll tick is a miss.
|
||||
piSession.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
});
|
||||
piSession.emit({
|
||||
type: "tool_execution_end",
|
||||
toolCallId: "call-1",
|
||||
toolName: "subagent",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
runId: "run-1",
|
||||
asyncId: "run-1",
|
||||
asyncDir: join(workdir, "gone"),
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
const row = () => manager.getProviderSubagent(AGENT_ID, "call-1");
|
||||
await waitFor(() => row()?.status === "running");
|
||||
|
||||
// Model delayed delivery of the parent turn end, with no provider cancellation.
|
||||
// Only Date moves: the adapter's real polling interval has not expired its miss budget.
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(new Date(Date.now() + 60_000));
|
||||
piSession.emit({ type: "agent_start" });
|
||||
piSession.emit({ type: "agent_end" });
|
||||
await drainSessionEvents();
|
||||
expect(row()?.status).toBe("running");
|
||||
|
||||
// The provider's completion notice, rather than elapsed time in the manager, settles it.
|
||||
piSession.emit({
|
||||
type: "message_end",
|
||||
message: {
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: `Background task completed: **scout**\nRetention-managed async directory: ${join(workdir, "gone")}`,
|
||||
display: true,
|
||||
},
|
||||
});
|
||||
await drainSessionEvents();
|
||||
expect(row()?.status).toBe("completed");
|
||||
} finally {
|
||||
try {
|
||||
await manager.closeAgent(AGENT_ID);
|
||||
} catch {
|
||||
// A createAgent failure leaves nothing to close; the workdir cleanup must still run.
|
||||
} finally {
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test("history replay preserves the async status supplied by Pi", async () => {
|
||||
const pi = new FakePi();
|
||||
const client = new PiRpcAgentClient({
|
||||
logger: pino({ level: "silent" }),
|
||||
runtime: pi,
|
||||
// The injected runtime owns execution; use an existing executable for the real availability
|
||||
// probe so this test does not depend on a Pi installation or replace a client method.
|
||||
runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } },
|
||||
});
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagent-poll-"));
|
||||
const manager = new AgentManager({
|
||||
clients: { pi: client },
|
||||
idFactory: () => AGENT_ID,
|
||||
logger: createTestLogger(),
|
||||
});
|
||||
try {
|
||||
await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
pi.latestSession().messages = [
|
||||
{
|
||||
role: "assistant",
|
||||
content: [
|
||||
{
|
||||
type: "toolCall",
|
||||
id: "call-replay",
|
||||
name: "subagent",
|
||||
arguments: { workflowScript: "return await runs.all([]);", async: true },
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
role: "toolResult",
|
||||
toolCallId: "call-replay",
|
||||
toolName: "subagent",
|
||||
content: [],
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-replay",
|
||||
asyncId: "run-replay",
|
||||
asyncDir: join(workdir, "gone"),
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
];
|
||||
await manager.hydrateTimelineFromProvider(AGENT_ID, { force: true });
|
||||
expect(manager.getProviderSubagent(AGENT_ID, "call-replay")?.status).toBe("running");
|
||||
} finally {
|
||||
await manager.closeAgent(AGENT_ID);
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -3,7 +3,8 @@ import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
|
||||
import { createTestLogger } from "../../../test-utils/test-logger.js";
|
||||
import { runProviderRefreshWithDeadline } from "../provider-refresh-deadline.js";
|
||||
@@ -203,16 +204,28 @@ describe("GenericACPAgentClient diagnostics", () => {
|
||||
});
|
||||
|
||||
test("reports a hung ACP session/new phase without failing the diagnostic", async () => {
|
||||
await withFakeACPAgent("hang-session", async (scriptPath, mode) => {
|
||||
await withFakeACPAgent("hang-session", async (scriptPath, mode, testDir) => {
|
||||
const tracePath = path.join(testDir, "phase-trace.jsonl");
|
||||
const client = new GenericACPAgentClient({
|
||||
logger: createTestLogger(),
|
||||
command: [process.execPath, scriptPath, mode],
|
||||
command: [process.execPath, scriptPath, mode, "", tracePath],
|
||||
providerId: "grok",
|
||||
label: "Grok",
|
||||
diagnosticPhaseTimeoutMs: TEST_ACP_TIMEOUT_MS,
|
||||
});
|
||||
|
||||
const { diagnostic } = await client.getDiagnostic();
|
||||
// Freeze phase deadlines until the child actually reaches session/new. Process startup
|
||||
// and initialize latency must not consume the session timeout this test exercises.
|
||||
vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
|
||||
let diagnostic: string;
|
||||
try {
|
||||
const pending = client.getDiagnostic();
|
||||
await waitForSessionNew(tracePath);
|
||||
await vi.advanceTimersByTimeAsync(TEST_ACP_TIMEOUT_MS);
|
||||
({ diagnostic } = await pending);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
|
||||
expect(diagnostic).toContain("Grok (ACP)");
|
||||
expect(diagnostic).toContain("Provider ID: grok");
|
||||
@@ -356,6 +369,14 @@ async function withTempDir(prefix: string, run: (testDir: string) => Promise<voi
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForSessionNew(tracePath: string): Promise<void> {
|
||||
const readyBy = Date.now() + 10_000;
|
||||
while (!(await readFile(tracePath, "utf8").catch(() => "")).includes("session/new")) {
|
||||
if (Date.now() >= readyBy) throw new Error("Fake ACP agent did not reach session/new");
|
||||
await delay(20);
|
||||
}
|
||||
}
|
||||
|
||||
async function expectProcessExit(pid: number): Promise<void> {
|
||||
const deadline = Date.now() + 2_000;
|
||||
while (Date.now() < deadline) {
|
||||
@@ -426,6 +447,7 @@ rl.on("line", (line) => {
|
||||
|
||||
if (message.method === "session/new") {
|
||||
if (mode === "hang-session") {
|
||||
if (initializeTracePath) fs.appendFileSync(initializeTracePath, "session/new\\n");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
@@ -2101,6 +2101,52 @@ describe("PiRpcAgentSession", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("fails an unfinished workflow on process exit while preserving detached runs", async () => {
|
||||
const { pi, session, events } = await createSession();
|
||||
onTestFinished(() => session.close());
|
||||
const updates: AgentStreamEvent[] = [];
|
||||
session.subscribe((event) => updates.push(event));
|
||||
await session.startTurn("run workflows");
|
||||
const runtime = pi.latestSession();
|
||||
runtime.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "detached",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all([]);", async: true },
|
||||
});
|
||||
runtime.emit({
|
||||
type: "tool_execution_end",
|
||||
toolCallId: "detached",
|
||||
toolName: "subagent",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "detached-run",
|
||||
asyncId: "detached-run",
|
||||
asyncDir: "/nonexistent/pi-exit-regression",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
runtime.emit({
|
||||
type: "tool_execution_start",
|
||||
toolCallId: "foreground",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all([]);", async: false },
|
||||
});
|
||||
runtime.emit({ type: "process_exit", error: "Pi exited" });
|
||||
|
||||
await expect(events.nextTurnFailure()).resolves.toMatchObject({ error: "Pi exited" });
|
||||
const statuses = (id: string) =>
|
||||
updates.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" && event.event.id === id
|
||||
? [event.event.status]
|
||||
: [],
|
||||
);
|
||||
expect(statuses("foreground")).toEqual(["running", "failed"]);
|
||||
expect(statuses("detached")).toEqual(["running", "running"]);
|
||||
});
|
||||
|
||||
test("fails an autonomous turn when the Pi process exits before settlement", async () => {
|
||||
const { pi, events } = await createSession();
|
||||
const fakeSession = pi.latestSession();
|
||||
|
||||
@@ -2108,6 +2108,15 @@ export class PiRpcAgentSession implements AgentSession {
|
||||
}
|
||||
|
||||
private handleProcessExit(error: string): void {
|
||||
for (const [toolCallId, toolCall] of this.activeToolCalls) {
|
||||
this.handleToolExecutionEnd({
|
||||
type: "tool_execution_end",
|
||||
toolCallId,
|
||||
toolName: toolCall.toolName,
|
||||
isError: true,
|
||||
result: { content: [{ type: "text", text: error }] },
|
||||
});
|
||||
}
|
||||
this.rejectAllExtensionResults(new Error(error));
|
||||
this.interruptingTurn = null;
|
||||
if (!this.activeTurnId && !this.activeTurnStarted) {
|
||||
|
||||
+153
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
+868
-2
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { mkdtemp, writeFile, rm } from "node:fs/promises";
|
||||
import { mkdtemp, writeFile, appendFile, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { createPiExtensionHost } from "../index.js";
|
||||
@@ -45,6 +45,18 @@ function hasCompletedChild(events: AgentStreamEvent[]) {
|
||||
);
|
||||
}
|
||||
|
||||
function subagentUpserts(events: AgentStreamEvent[]) {
|
||||
return events.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert" ? [event.event] : [],
|
||||
);
|
||||
}
|
||||
|
||||
function reportedUpsert(events: AgentStreamEvent[], id: string, status?: string) {
|
||||
return subagentUpserts(events).some(
|
||||
(event) => event.id === id && (status === undefined || event.status === status),
|
||||
);
|
||||
}
|
||||
|
||||
describe("pi-subagents adapter", () => {
|
||||
test("follows async status and transcript without bg_wait", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-async-"));
|
||||
@@ -273,4 +285,858 @@ describe("pi-subagents adapter", () => {
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("maps a reconstructed workflowScript run to a live spawn row and its children", async () => {
|
||||
const fixture = readSubagentFixture(new URL("./fixtures/workflow.json", import.meta.url));
|
||||
const events = await verifySubagentFixture(fixture);
|
||||
const upserts = events
|
||||
.filter((event) => event.event.type === "upsert")
|
||||
.map((event) => (event.event.type === "upsert" ? event.event : null));
|
||||
expect(upserts.map((event) => [event?.id, event?.status])).toEqual([
|
||||
["call_wf1", "running"],
|
||||
["call_wf1", "running"],
|
||||
["call_wf1:alpha", "completed"],
|
||||
["call_wf1", "completed"],
|
||||
]);
|
||||
// The workflow call carries no agent/task, so the script file names the run.
|
||||
expect(upserts[0]?.title).toBe("rigor-ui-test.js");
|
||||
expect(upserts[2]?.title).toBeUndefined();
|
||||
});
|
||||
|
||||
test("maps a reported workflowScript run to its reviewer row and saved timeline", async () => {
|
||||
// Reported in #5837: `subagent({ workflow: true })` reaches the adapter as `workflowScript`,
|
||||
// not `workflow`, so a launch check keyed on `workflow` alone drops the whole run.
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-script-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
await writeFile(
|
||||
sessionFile,
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: { role: "assistant", content: [{ type: "text", text: "README summary" }] },
|
||||
}) + "\n",
|
||||
);
|
||||
const args = {
|
||||
workflowScript:
|
||||
"return await runs.run('review', { agent: 'reviewer', task: 'Read README.md' });",
|
||||
async: false,
|
||||
};
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "reported-workflow",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "running",
|
||||
result: null,
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
expect.objectContaining({ id: "reported-workflow", title: "workflow", status: "running" }),
|
||||
]);
|
||||
const output = host.mapToolCall({
|
||||
callId: "reported-workflow",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "reported-run",
|
||||
results: [{ index: 0, agent: "reviewer", exitCode: 0, sessionFile }],
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(output?.subagents).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ type: "upsert", title: "reviewer", status: "completed" }),
|
||||
]),
|
||||
);
|
||||
expect(hasChildTimeline(await output!.hydration)).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("keys workflow child rows by their childKey while the run reports progress", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
|
||||
expect.arrayContaining([
|
||||
["call-wf:alpha", "running"],
|
||||
["call-wf:beta", "running"],
|
||||
["call-wf", "running"],
|
||||
]),
|
||||
);
|
||||
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "complete",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "complete" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => reportedUpsert(received, "call-wf", "completed")).toBe(true);
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:alpha")?.status,
|
||||
).toBe("completed");
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("keeps a missing status directory's row alive until the miss budget expires", async () => {
|
||||
vi.useFakeTimers();
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-lost",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
runId: "run-1",
|
||||
asyncId: "run-1",
|
||||
asyncDir: "/nonexistent",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(250 * 5);
|
||||
// A missing status file supplies no new outcome before the miss budget expires.
|
||||
const duringMisses = subagentUpserts(received);
|
||||
expect(duringMisses).toEqual([]);
|
||||
|
||||
// 60 s of missing ticks, the budget before a status directory counts as gone.
|
||||
await vi.advanceTimersByTimeAsync(250 * 236);
|
||||
// The launch row comes back through the caller; polling reports the give-up exactly once.
|
||||
const afterGiveUp = subagentUpserts(received);
|
||||
expect(afterGiveUp.findLast((event) => event.id === "call-lost")).toEqual({
|
||||
type: "upsert",
|
||||
id: "call-lost",
|
||||
status: "canceled",
|
||||
});
|
||||
expect(afterGiveUp.filter((event) => event.status === "canceled")).toHaveLength(1);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
test("a completion notice stops polling so a stale status file cannot revert the card", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stale-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({ state: "running", steps: [{ agent: "scout", status: "running" }] }),
|
||||
);
|
||||
const statuses = () =>
|
||||
received.flatMap((event) =>
|
||||
event.type === "provider_subagent" && event.event.type === "upsert"
|
||||
? [event.event.status]
|
||||
: [],
|
||||
);
|
||||
await expect.poll(() => statuses().length).toBeGreaterThan(1);
|
||||
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task completed: **scout**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
});
|
||||
const settled = statuses().length;
|
||||
|
||||
// The run's directory is still on disk with a running status; polling must stay off.
|
||||
await new Promise((resolve) => setTimeout(resolve, 400));
|
||||
expect(statuses().length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a bg_wait result keeps its verdict while the run's status names no child yet", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-settled-owner-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.follow((event) => received.push(event));
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
// A single unnamed child takes the owner's id, so this result settles the launch row while the
|
||||
// run itself keeps polling.
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "wait-1",
|
||||
toolName: "bg_wait",
|
||||
args: { runId: "run-1" },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "single",
|
||||
results: [],
|
||||
completions: [
|
||||
{ runId: "run-1", results: [{ agent: "scout", success: true, exitCode: 0 }] },
|
||||
],
|
||||
},
|
||||
},
|
||||
})?.subagents,
|
||||
).toEqual([expect.objectContaining({ id: "call-1", status: "completed" })]);
|
||||
|
||||
// The queue has not started the child yet, so the status file names no step: that tick must
|
||||
// not re-report the owner as running over the verdict it already has.
|
||||
await writeFile(join(dir, "status.json"), JSON.stringify({ state: "running", steps: [] }));
|
||||
const beforeTick = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 2);
|
||||
|
||||
expect(subagentUpserts(received.slice(beforeTick))).toEqual([]);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a paused run settles as canceled", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-paused-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-1",
|
||||
toolName: "subagent",
|
||||
args: { agent: "scout", task: "Inspect", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: { mode: "single", runId: "run-1", asyncId: "run-1", asyncDir: dir, results: [] },
|
||||
},
|
||||
});
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task paused: **scout**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-1", status: "canceled" }]);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a paused workflow child notice settles its row as canceled", () => {
|
||||
const host = createPiExtensionHost();
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: "/nonexistent",
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child paused (needs attention): **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "canceled" }]);
|
||||
});
|
||||
|
||||
test("partial and rejected runs settle every child instead of leaving a spinner", async () => {
|
||||
for (const [state, expected] of [
|
||||
["partial", ["completed", "completed", "canceled"]],
|
||||
["rejected", ["completed", "completed", "failed"]],
|
||||
] as const) {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-partial-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state,
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
// WorkflowNodeStatus spells the same outcome `completed`.
|
||||
{ agent: "auditor", workflowKey: "gamma", status: "completed" },
|
||||
{
|
||||
agent: "reviewer",
|
||||
workflowKey: "beta",
|
||||
status: state === "partial" ? "running" : "failed",
|
||||
},
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
|
||||
expect.arrayContaining([
|
||||
["call-wf:alpha", expected[0]],
|
||||
["call-wf:gamma", expected[1]],
|
||||
["call-wf:beta", expected[2]],
|
||||
["call-wf", "canceled"],
|
||||
]),
|
||||
);
|
||||
expect(
|
||||
subagentUpserts(received).filter((event) => event.status === "running"),
|
||||
).toHaveLength(0);
|
||||
|
||||
// The run is terminal, so the poll stops instead of re-reporting fresh rows.
|
||||
const settled = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 4);
|
||||
expect(received.length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
test("a child notice keeps its status while the run is still reporting", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-pin-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
// The stale step entry still says running; the child's notice is its verdict.
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
const beforeNotice = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 3);
|
||||
expect(
|
||||
subagentUpserts(received.slice(beforeNotice)).filter(
|
||||
(event) => event.id === "call-wf:alpha",
|
||||
),
|
||||
).toEqual([]);
|
||||
// The rest of the run keeps re-reporting, so this is not a poll that simply stopped.
|
||||
expect(reportedUpsert(received.slice(beforeNotice), "call-wf:beta")).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a child notice's verdict survives the run's terminal tick", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-notice-terminal-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
host.follow((event) => received.push(event));
|
||||
// The child's notice settles it before the run itself ends.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
// The run then ends without a verdict for alpha: its step entry still says running, which the
|
||||
// terminal tick would read as canceled if the notice were not the child's verdict.
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "stopped",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "running" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "stopped" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
expect(
|
||||
subagentUpserts(received).filter((event) => event.id === "call-wf:alpha" && event.status),
|
||||
).toEqual([expect.objectContaining({ id: "call-wf:alpha", status: "completed" })]);
|
||||
// The child that never reported a verdict of its own is still settled by that tick.
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
|
||||
).toBe("canceled");
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("gives up without rewriting a child that already settled", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-giveup-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "complete" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
|
||||
// The workflow keeps running, so polling continues after the child's own notice.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
await rm(join(dir, "status.json"));
|
||||
await vi.advanceTimersByTimeAsync(250 * 241);
|
||||
|
||||
expect(
|
||||
subagentUpserts(received).some(
|
||||
(event) => event.id === "call-wf:alpha" && event.status === "canceled",
|
||||
),
|
||||
).toBe(false);
|
||||
expect(
|
||||
subagentUpserts(received).findLast((event) => event.id === "call-wf:beta")?.status,
|
||||
).toBe("canceled");
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a run's final notice cancels the children it stopped with", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-stopped-children-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [
|
||||
{ agent: "scout", workflowKey: "alpha", status: "running" },
|
||||
{ agent: "reviewer", workflowKey: "beta", status: "running" },
|
||||
],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
expect(reportedUpsert(received, "call-wf:beta")).toBe(true);
|
||||
|
||||
// alpha reports its own verdict before the run stops.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
// The run then stops with beta unfinished. The notice ends the poll, so no later tick will
|
||||
// ever report beta: it settles as canceled, never completed. alpha keeps its own verdict.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-notify",
|
||||
content: [
|
||||
"Background task stopped: **rigor-ui-test.js**",
|
||||
"",
|
||||
`Retention-managed async directory: ${dir}`,
|
||||
].join("\n"),
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
{ type: "upsert", id: "call-wf", status: "canceled" },
|
||||
{ type: "upsert", id: "call-wf:beta", status: "canceled" },
|
||||
]);
|
||||
|
||||
// Polling is off; the status file still says running and must not bring the rows back.
|
||||
const settled = received.length;
|
||||
await vi.advanceTimersByTimeAsync(250 * 3);
|
||||
expect(received.length).toBe(settled);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("names a child a notice settled before the first poll saw it", async () => {
|
||||
vi.useFakeTimers();
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-fast-child-"));
|
||||
const received: AgentStreamEvent[] = [];
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
host.mapToolCall({
|
||||
callId: "call-wf",
|
||||
toolName: "subagent",
|
||||
args: { workflow: "tmp/rigor-ui-test.js", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "run-wf",
|
||||
asyncId: "run-wf",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
// A fast child reports before any poll tick, so its row is born without a name.
|
||||
expect(
|
||||
host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **alpha**\nWorkflow run: run-wf",
|
||||
})?.subagents,
|
||||
).toEqual([{ type: "upsert", id: "call-wf:alpha", status: "completed" }]);
|
||||
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [{ agent: "scout", workflowKey: "alpha", status: "running" }],
|
||||
}),
|
||||
);
|
||||
host.follow((event) => received.push(event));
|
||||
await vi.advanceTimersByTimeAsync(250);
|
||||
|
||||
// The poll leaves the settled status alone but still fills the name status.json carries.
|
||||
const alphaUpserts = () =>
|
||||
subagentUpserts(received).filter((event) => event.id === "call-wf:alpha");
|
||||
expect(alphaUpserts()).toEqual([
|
||||
{
|
||||
type: "upsert",
|
||||
id: "call-wf:alpha",
|
||||
title: "scout",
|
||||
status: "completed",
|
||||
},
|
||||
]);
|
||||
|
||||
// Reported once, not on every tick.
|
||||
await vi.advanceTimersByTimeAsync(250 * 2);
|
||||
expect(alphaUpserts()).toHaveLength(1);
|
||||
} finally {
|
||||
host.close();
|
||||
vi.useRealTimers();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("a workflow that reported rows replaces its placeholder with them", () => {
|
||||
const host = createPiExtensionHost();
|
||||
const args = { workflow: true };
|
||||
// The launch placeholder is replaced once the workflow supplies its child rows.
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "fanout",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "running",
|
||||
result: null,
|
||||
})?.subagents,
|
||||
).toEqual([expect.objectContaining({ id: "fanout", status: "running" })]);
|
||||
expect(
|
||||
host.mapToolCall({
|
||||
callId: "fanout",
|
||||
toolName: "subagent",
|
||||
args,
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
results: [
|
||||
{ index: 0, agent: "scout", workflowKey: "alpha", success: true, exitCode: 0 },
|
||||
{ index: 1, agent: "reviewer", workflowKey: "beta", success: true, exitCode: 0 },
|
||||
],
|
||||
},
|
||||
},
|
||||
})?.subagents,
|
||||
).toEqual([
|
||||
{ type: "remove", id: "fanout" },
|
||||
expect.objectContaining({ id: "fanout:alpha", title: "scout", status: "completed" }),
|
||||
expect.objectContaining({ id: "fanout:beta", title: "reviewer", status: "completed" }),
|
||||
]);
|
||||
});
|
||||
test("replays keyed async workflow child transcripts without a live poll", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-workflow-replay-"));
|
||||
const host = createPiExtensionHost();
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
await writeFile(
|
||||
sessionFile,
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text: "Saved reviewer answer" }],
|
||||
},
|
||||
}) + "\n",
|
||||
);
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "complete",
|
||||
steps: [{ agent: "reviewer", workflowKey: "review", status: "complete", sessionFile }],
|
||||
}),
|
||||
);
|
||||
const output = host.mapToolCall({
|
||||
callId: "replayed-workflow",
|
||||
toolName: "subagent",
|
||||
args: { workflowScript: "return await runs.all(items);", async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "saved-run",
|
||||
asyncId: "saved-run",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
const events = await output!.hydration;
|
||||
expect(
|
||||
events.some(
|
||||
(event) =>
|
||||
event.type === "provider_subagent" &&
|
||||
event.event.type === "timeline" &&
|
||||
event.event.id === "replayed-workflow:review",
|
||||
),
|
||||
).toBe(true);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("stops following a child whose completion arrived before its transcript was attached", async () => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paseo-pi-finished-reader-"));
|
||||
const host = createPiExtensionHost();
|
||||
const received: AgentStreamEvent[] = [];
|
||||
try {
|
||||
const sessionFile = join(dir, "child.jsonl");
|
||||
const message = (text: string) =>
|
||||
JSON.stringify({
|
||||
type: "message",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "text", text }],
|
||||
},
|
||||
}) + "\n";
|
||||
await writeFile(sessionFile, message("Finished child"));
|
||||
host.follow((event) => received.push(event));
|
||||
const spawn = host.mapToolCall({
|
||||
callId: "fast-workflow",
|
||||
toolName: "subagent",
|
||||
args: { workflow: true, async: true },
|
||||
status: "completed",
|
||||
result: {
|
||||
details: {
|
||||
mode: "workflow",
|
||||
runId: "fast-run",
|
||||
asyncId: "fast-run",
|
||||
asyncDir: dir,
|
||||
results: [],
|
||||
},
|
||||
},
|
||||
});
|
||||
await spawn!.hydration;
|
||||
const notice = host.mapCustomMessage({
|
||||
role: "custom",
|
||||
customType: "subagent-incremental-child-notify",
|
||||
content: "Workflow child completed: **review**\nWorkflow run: fast-run",
|
||||
});
|
||||
await notice!.hydration;
|
||||
await writeFile(
|
||||
join(dir, "status.json"),
|
||||
JSON.stringify({
|
||||
state: "running",
|
||||
steps: [{ agent: "reviewer", workflowKey: "review", status: "running", sessionFile }],
|
||||
}),
|
||||
);
|
||||
await expect.poll(() => hasChildTimeline(received)).toBe(true);
|
||||
const timelineCount = () =>
|
||||
received.filter(
|
||||
(event) => event.type === "provider_subagent" && event.event.type === "timeline",
|
||||
).length;
|
||||
const count = timelineCount();
|
||||
await appendFile(sessionFile, message("Later content must not be followed"));
|
||||
await new Promise((resolve) => setTimeout(resolve, 600));
|
||||
expect(timelineCount()).toBe(count);
|
||||
} finally {
|
||||
host.close();
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2,12 +2,20 @@ import { z } from "zod";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { extractTextFromToolResult } from "../../tool-call-mapper.js";
|
||||
import type { PiAgentMessage } from "../../rpc-types.js";
|
||||
import type { ProviderSubagentInputEvent } from "../../../../provider-subagents/store.js";
|
||||
import type { PiExtension, PiExtensionToolCall, PiExtensionToolMapping } from "../contract.js";
|
||||
|
||||
type CustomMessage = Extract<PiAgentMessage, { role: "custom" }>;
|
||||
type SubagentUpsert = Extract<ProviderSubagentInputEvent, { type: "upsert" }>;
|
||||
|
||||
const Args = z
|
||||
.object({
|
||||
agent: z.string().trim().min(1).optional(),
|
||||
task: z.string().trim().min(1).optional(),
|
||||
workflow: z.union([z.string().trim().min(1), z.literal(true)]).optional(),
|
||||
// `subagent({ workflow: true })` launches the reply's script; Pi delivers it as this field.
|
||||
workflowScript: z.string().trim().min(1).optional(),
|
||||
async: z.boolean().optional(),
|
||||
action: z.unknown().optional(),
|
||||
})
|
||||
@@ -16,6 +24,7 @@ const Row = z
|
||||
.object({
|
||||
index: z.number().int().nonnegative().optional(),
|
||||
agent: z.string().trim().min(1),
|
||||
workflowKey: z.string().trim().min(1).optional(),
|
||||
exitCode: z.number().optional(),
|
||||
success: z.boolean().optional(),
|
||||
sessionFile: z.string().trim().min(1).optional(),
|
||||
@@ -41,6 +50,7 @@ const AsyncStatus = z
|
||||
z
|
||||
.object({
|
||||
agent: z.string().trim().min(1),
|
||||
workflowKey: z.string().trim().min(1).optional(),
|
||||
status: z.string().optional(),
|
||||
sessionFile: z.string().trim().min(1).optional(),
|
||||
})
|
||||
@@ -49,21 +59,116 @@ const AsyncStatus = z
|
||||
})
|
||||
.passthrough();
|
||||
|
||||
/**
|
||||
* Run states that will not change again: `paused` waits on the user, and `partial`/`rejected`
|
||||
* still end the run even when some steps never produced a result of their own.
|
||||
*/
|
||||
const TERMINAL_RUN_STATES = new Set([
|
||||
"complete",
|
||||
"failed",
|
||||
"stopped",
|
||||
"paused",
|
||||
"partial",
|
||||
"rejected",
|
||||
]);
|
||||
/** The run's status file, or undefined while it is missing or not yet readable. */
|
||||
function readAsyncStatus(dir: string): z.infer<typeof AsyncStatus> | undefined {
|
||||
try {
|
||||
const parsed = AsyncStatus.safeParse(
|
||||
JSON.parse(readFileSync(join(dir, "status.json"), "utf8")),
|
||||
);
|
||||
return parsed.success ? parsed.data : undefined;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** One async run the poll follows, keyed by its async directory. */
|
||||
interface AsyncRun {
|
||||
owner: string;
|
||||
polling: boolean;
|
||||
misses: number;
|
||||
deadline: number;
|
||||
rows: Set<string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Missing ticks before a status directory counts as gone: 60 s at the 250 ms follow cadence. A live
|
||||
* run can briefly lose its file while the concurrency queue starts it or retention rotates it.
|
||||
*/
|
||||
const POLL_MISS_LIMIT = 240;
|
||||
/**
|
||||
* A run that never reaches a terminal state in its status file must not keep a card spinning
|
||||
* forever. pi-subagents' own default run timeout is 30 minutes; this is the backstop.
|
||||
*/
|
||||
const POLL_TTL_MS = 6 * 60 * 60 * 1000;
|
||||
|
||||
function asyncChildStatus(runState: string, stepStatus?: string) {
|
||||
if (!["complete", "failed", "stopped"].includes(runState)) return "running";
|
||||
if (runState === "stopped" || stepStatus === "stopped") return "canceled";
|
||||
if (!TERMINAL_RUN_STATES.has(runState)) return "running";
|
||||
// A partial or rejected run says nothing about its individual children: read their own outcome,
|
||||
// and treat a step that never reached one as canceled rather than still running.
|
||||
if (runState === "partial" || runState === "rejected") {
|
||||
// Two vocabularies: status.json steps say `complete`, WorkflowNodeStatus says `completed`.
|
||||
if (stepStatus === "complete" || stepStatus === "completed") return "completed";
|
||||
if (stepStatus === "failed" || stepStatus === "error") return "failed";
|
||||
return "canceled";
|
||||
}
|
||||
if (runState === "stopped" || runState === "paused" || stepStatus === "stopped")
|
||||
return "canceled";
|
||||
if (runState === "failed" || stepStatus === "failed" || stepStatus === "error") return "failed";
|
||||
return "completed";
|
||||
}
|
||||
|
||||
/** The `subagent-notify` header; a grouped notice reports every run as completed. */
|
||||
/** The status of the run itself, used for the row a launch opened before its children reported. */
|
||||
function asyncRunStatus(runState: string) {
|
||||
if (runState === "complete") return "completed" as const;
|
||||
if (runState === "failed") return "failed" as const;
|
||||
return "canceled" as const;
|
||||
}
|
||||
|
||||
/**
|
||||
* The `subagent-notify` / `subagent-incremental-child-notify` header. A grouped notice reports
|
||||
* every run as completed; a paused run is waiting on the user, which the track shows as canceled.
|
||||
*/
|
||||
function noticeStatus(header: string) {
|
||||
if (/^Background tasks completed \(\d+\):/.test(header)) return "completed";
|
||||
const outcome = header.match(
|
||||
/^(?:Background task|Detached foreground task) (completed|failed|stopped): \*\*/,
|
||||
/^(?:Background task|Detached foreground task|Workflow child) (completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*/,
|
||||
)?.[1];
|
||||
if (outcome === "stopped") return "canceled";
|
||||
return outcome as "completed" | "failed" | undefined;
|
||||
if (!outcome) return undefined;
|
||||
return outcome === "paused" || outcome === "stopped"
|
||||
? ("canceled" as const)
|
||||
: (outcome as "completed" | "failed");
|
||||
}
|
||||
|
||||
/**
|
||||
* A workflow launch carries no `agent`/`task`: the children are declared by the script. Its label
|
||||
* is the script file name, or the bare `workflow` of the reply-block form.
|
||||
*/
|
||||
function workflowLabel(workflow: string | true | undefined) {
|
||||
if (workflow === undefined) return undefined;
|
||||
if (workflow === true) return "workflow";
|
||||
const segments = workflow.split(/[\\/]/).filter(Boolean);
|
||||
return segments[segments.length - 1] ?? workflow;
|
||||
}
|
||||
|
||||
/**
|
||||
* A workflow launch carries no agent/task; its children are declared by the script. A reply-scoped
|
||||
* script arrives as `workflowScript` instead of a `workflow` path, so it takes the generic label.
|
||||
*/
|
||||
function workflowTitle(args: { workflow?: string | true; workflowScript?: string }) {
|
||||
const label = workflowLabel(args.workflow);
|
||||
if (label !== undefined) return label;
|
||||
return args.workflowScript === undefined ? undefined : "workflow";
|
||||
}
|
||||
|
||||
/**
|
||||
* A row keeps one id from launch to settle: the owner alone for a single unnamed child, otherwise
|
||||
* `owner:key`. Workflow children carry their `workflowKey`, so progress updates, child notices and
|
||||
* final results all land on the same row.
|
||||
*/
|
||||
function rowId(owner: string, count: number, key: string | undefined, index: number) {
|
||||
return count === 1 && !key ? owner : `${owner}:${key ?? index}`;
|
||||
}
|
||||
|
||||
export const piSubagents: PiExtension = {
|
||||
@@ -71,13 +176,64 @@ export const piSubagents: PiExtension = {
|
||||
createSession: () => {
|
||||
const callsByRun = new Map<string, string>();
|
||||
const readSessions = new Set<string>();
|
||||
/** Rows already reported terminal; a late give-up must not rewrite one of them. */
|
||||
const settledRows = new Map<string, NonNullable<SubagentUpsert["status"]>>();
|
||||
/** Rows whose poll already reported the step's agent. A settled row is still named once. */
|
||||
const namedRows = new Set<string>();
|
||||
const recordTerminal = (id: string, status: NonNullable<SubagentUpsert["status"]>) => {
|
||||
if (status !== "running") settledRows.set(id, status);
|
||||
};
|
||||
/** Keyed by async directory, which both status.json and the completion notice carry. */
|
||||
const asyncRuns = new Map<string, { owner: string; polling: boolean }>();
|
||||
const asyncRuns = new Map<string, AsyncRun>();
|
||||
const rememberAsyncRun = (details: z.infer<typeof Details>, call: PiExtensionToolCall) => {
|
||||
if (details.asyncDir && call.status !== "failed") {
|
||||
asyncRuns.set(details.asyncDir, { owner: call.callId, polling: true });
|
||||
asyncRuns.set(details.asyncDir, {
|
||||
owner: call.callId,
|
||||
polling: true,
|
||||
misses: 0,
|
||||
deadline: Date.now() + POLL_TTL_MS,
|
||||
rows: new Set<string>(),
|
||||
});
|
||||
}
|
||||
};
|
||||
/**
|
||||
* The run is over and nothing will ever update its rows again, so settle them rather than leave a
|
||||
* spinner. A row that already reported a terminal status keeps it: this is not its verdict.
|
||||
*/
|
||||
const giveUpRun = (run: AsyncRun) => {
|
||||
run.polling = false;
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
for (const id of new Set([run.owner, ...run.rows])) {
|
||||
if (settledRows.has(id)) continue;
|
||||
subagents.push({ type: "upsert", id, status: "canceled" });
|
||||
}
|
||||
return subagents;
|
||||
};
|
||||
/**
|
||||
* The notice is the run's last word: it ends the poll, settles the launch row, and cancels the
|
||||
* children the run stopped with. No later tick will report those, and nothing here proves they
|
||||
* succeeded, so they settle as canceled. A child that already reported its own verdict keeps it.
|
||||
*/
|
||||
const settleRunNotice = (run: AsyncRun, status: "completed" | "failed" | "canceled") => {
|
||||
// A stale status.json must not revert this card either.
|
||||
run.polling = false;
|
||||
recordTerminal(run.owner, status);
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [
|
||||
{ type: "upsert", id: run.owner, status },
|
||||
];
|
||||
for (const id of run.rows) {
|
||||
if (settledRows.has(id)) continue;
|
||||
subagents.push({ type: "upsert", id, status: "canceled" });
|
||||
}
|
||||
return subagents;
|
||||
};
|
||||
/**
|
||||
* A row id can settle twice: a completion notice records a verdict in `settledRows`, and a
|
||||
* `subagent_wait` result repeats the run's rows with the run's own verdict. Both are terminal, so
|
||||
* a disagreement moves a row terminal-to-terminal (completed → failed) and never resurrects it;
|
||||
* the owner row is what carries the run outcome.
|
||||
*/
|
||||
// ponytail: child verdicts can diverge between notice and bg_wait result — unify when one source of truth exists
|
||||
const collectRows = (
|
||||
owner: string,
|
||||
rows: z.infer<typeof Row>[],
|
||||
@@ -87,17 +243,19 @@ export const piSubagents: PiExtension = {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
for (const [index, row] of rows.entries()) {
|
||||
const id = rows.length === 1 ? owner : `${owner}:${row.index ?? index}`;
|
||||
const id = rowId(owner, rows.length, row.workflowKey, row.index ?? index);
|
||||
const status =
|
||||
failed || row.success === false || (row.success !== true && row.exitCode !== 0)
|
||||
? ("failed" as const)
|
||||
: ("completed" as const);
|
||||
recordTerminal(id, status);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: row.agent,
|
||||
description,
|
||||
toolCallId: owner,
|
||||
status:
|
||||
failed || row.success === false || (row.success !== true && row.exitCode !== 0)
|
||||
? "failed"
|
||||
: "completed",
|
||||
status,
|
||||
});
|
||||
if (row.sessionFile && !readSessions.has(row.sessionFile)) {
|
||||
readSessions.add(row.sessionFile);
|
||||
@@ -123,66 +281,164 @@ export const piSubagents: PiExtension = {
|
||||
}
|
||||
return subagents.length ? { subagents, childSessions } : undefined;
|
||||
};
|
||||
const mapSpawn = (call: PiExtensionToolCall) => {
|
||||
const args = Args.safeParse(call.args);
|
||||
if (!args.success || args.data.action !== undefined || !args.data.agent || !args.data.task)
|
||||
return undefined;
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: args.data.agent,
|
||||
description: args.data.task,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
};
|
||||
const base = {
|
||||
type: "upsert" as const,
|
||||
id: call.callId,
|
||||
title: args.data.agent,
|
||||
description: args.data.task,
|
||||
toolCallId: call.callId,
|
||||
};
|
||||
if (call.status === "running")
|
||||
return { detail, subagents: [{ ...base, status: "running" as const }] };
|
||||
/** Project a saved status both when replaying a launch and when its live poll advances. */
|
||||
const projectAsyncStatus = (run: AsyncRun, status: z.infer<typeof AsyncStatus>) => {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
const runTerminal = TERMINAL_RUN_STATES.has(status.state);
|
||||
for (const [index, step] of status.steps.entries()) {
|
||||
const id = rowId(run.owner, status.steps.length, step.workflowKey, index);
|
||||
run.rows.add(id);
|
||||
// A child's notice supplies its verdict even when its saved status is stale. Reuse that
|
||||
// verdict when attaching a transcript or filling its name after the notice arrived.
|
||||
const settledStatus = settledRows.get(id);
|
||||
const childSession =
|
||||
step.sessionFile && !readSessions.has(step.sessionFile)
|
||||
? { id, file: step.sessionFile }
|
||||
: undefined;
|
||||
if (!settledStatus) {
|
||||
const childStatus = asyncChildStatus(status.state, step.status);
|
||||
recordTerminal(id, childStatus);
|
||||
namedRows.add(id);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: step.agent,
|
||||
status: childStatus,
|
||||
});
|
||||
} else if (!namedRows.has(id) || childSession) {
|
||||
// Attach a late transcript with its known verdict, so the host reads it once instead of
|
||||
// starting a follower that no later terminal update will release.
|
||||
namedRows.add(id);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id,
|
||||
title: step.agent,
|
||||
status: settledStatus,
|
||||
});
|
||||
}
|
||||
if (childSession) {
|
||||
readSessions.add(childSession.file);
|
||||
childSessions.push(childSession);
|
||||
}
|
||||
}
|
||||
if (runTerminal) run.polling = false;
|
||||
// The launch row stands for the whole run, and a single unnamed child already took its
|
||||
// id.
|
||||
// A row that already settled keeps that verdict: a result that landed before the run's
|
||||
// steps named its children must not be reverted to running by the next tick.
|
||||
if (!run.rows.has(run.owner) && !settledRows.has(run.owner)) {
|
||||
const ownerStatus = run.polling ? ("running" as const) : asyncRunStatus(status.state);
|
||||
recordTerminal(run.owner, ownerStatus);
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id: run.owner,
|
||||
status: ownerStatus,
|
||||
});
|
||||
}
|
||||
return { subagents, childSessions };
|
||||
};
|
||||
/** A settled spawn: Pi either reports the rows it ran, or an async directory the poll follows. */
|
||||
const mapSettledSpawn = (
|
||||
call: PiExtensionToolCall,
|
||||
base: SubagentUpsert,
|
||||
description: string | undefined,
|
||||
asyncRequested: boolean,
|
||||
): Pick<PiExtensionToolMapping, "subagents" | "childSessions"> => {
|
||||
const failed = call.status === "failed";
|
||||
const details = Details.safeParse(
|
||||
typeof call.result === "object" ? call.result?.details : null,
|
||||
);
|
||||
if (!details.success || details.data.mode === "management")
|
||||
return call.status === "failed"
|
||||
? { detail, subagents: [{ ...base, status: "failed" as const }] }
|
||||
: { detail };
|
||||
if (!details.success || details.data.mode === "management") {
|
||||
if (!failed) return { subagents: [] };
|
||||
recordTerminal(call.callId, "failed");
|
||||
return { subagents: [{ ...base, status: "failed" as const }] };
|
||||
}
|
||||
if (details.data.runId) callsByRun.set(details.data.runId, call.callId);
|
||||
if (details.data.results.length === 0 && (details.data.asyncId || args.data.async)) {
|
||||
if (details.data.results.length === 0) {
|
||||
if (failed) {
|
||||
recordTerminal(call.callId, "failed");
|
||||
return { subagents: [{ ...base, status: "failed" as const }] };
|
||||
}
|
||||
if (!details.data.asyncId && !asyncRequested) {
|
||||
// ponytail: inferred completion — revisit if pi can return success while a child outlives the call.
|
||||
recordTerminal(call.callId, "completed");
|
||||
return { subagents: [{ ...base, status: "completed" as const }] };
|
||||
}
|
||||
// Follow the async run from its saved status or later poll updates.
|
||||
rememberAsyncRun(details.data, call);
|
||||
const dir = details.data.asyncDir;
|
||||
const run = dir ? asyncRuns.get(dir) : undefined;
|
||||
const status = dir ? readAsyncStatus(dir) : undefined;
|
||||
if (run && status) return projectAsyncStatus(run, status);
|
||||
return { subagents: [{ ...base, status: "running" as const }] };
|
||||
}
|
||||
const collected = collectRows(call.callId, details.data.results, description, failed);
|
||||
// A run that reported rows of its own never keeps the placeholder card it opened at launch.
|
||||
if (!collected.subagents.some((event) => event.id === call.callId)) {
|
||||
collected.subagents.unshift({ type: "remove", id: call.callId });
|
||||
}
|
||||
return collected;
|
||||
};
|
||||
const mapSpawn = (call: PiExtensionToolCall) => {
|
||||
const args = Args.safeParse(call.args);
|
||||
if (!args.success || args.data.action !== undefined) return undefined;
|
||||
const single = Boolean(args.data.agent && args.data.task);
|
||||
let workflow: string | undefined;
|
||||
if (!single) workflow = workflowTitle(args.data);
|
||||
if (!single && workflow === undefined) return undefined;
|
||||
const base = {
|
||||
type: "upsert" as const,
|
||||
id: call.callId,
|
||||
title: (single ? args.data.agent : workflow) ?? "workflow",
|
||||
description: args.data.task,
|
||||
toolCallId: call.callId,
|
||||
};
|
||||
const detail = {
|
||||
type: "sub_agent" as const,
|
||||
subAgentType: base.title,
|
||||
description: args.data.task,
|
||||
log: extractTextFromToolResult(call.result)?.trim() ?? "",
|
||||
};
|
||||
// ponytail: replay re-emits running before the settled rows land — one-pass client flip, upgrade when replay maps calls with their results first
|
||||
if (call.status === "running")
|
||||
return {
|
||||
detail,
|
||||
subagents: [
|
||||
{
|
||||
...base,
|
||||
status: call.status === "failed" ? ("failed" as const) : ("running" as const),
|
||||
},
|
||||
],
|
||||
subagents: [{ ...base, status: "running" as const }],
|
||||
};
|
||||
}
|
||||
if (details.data.results.length === 0 && call.status === "failed")
|
||||
return { detail, subagents: [{ ...base, status: "failed" as const }] };
|
||||
return {
|
||||
detail,
|
||||
...collectRows(call.callId, details.data.results, args.data.task, call.status === "failed"),
|
||||
...mapSettledSpawn(call, base, args.data.task, Boolean(args.data.async)),
|
||||
};
|
||||
};
|
||||
const noticeText = (message: CustomMessage) =>
|
||||
typeof message.content === "string"
|
||||
? message.content
|
||||
: extractTextFromToolResult({ content: message.content });
|
||||
const mapWorkflowChildNotice = (message: CustomMessage) => {
|
||||
const lines = noticeText(message)?.split("\n") ?? [];
|
||||
const status = noticeStatus(lines[0] ?? "");
|
||||
const key = lines[0]?.match(
|
||||
/^Workflow child (?:completed|failed|paused|stopped)(?: \([^)]*\))?: \*\*(.+?)\*\*/,
|
||||
)?.[1];
|
||||
const runId = lines.find((line) => line.startsWith("Workflow run: "))?.slice(14);
|
||||
const owner = runId ? callsByRun.get(runId) : undefined;
|
||||
if (!status || !key || !owner) return undefined;
|
||||
const id = `${owner}:${key}`;
|
||||
recordTerminal(id, status);
|
||||
return { subagents: [{ type: "upsert" as const, id, status }] };
|
||||
};
|
||||
return {
|
||||
mapToolCall(call) {
|
||||
if (call.toolName === "bg_wait") return mapWait(call);
|
||||
if (call.toolName === "subagent") return mapSpawn(call);
|
||||
return undefined;
|
||||
},
|
||||
// The notice is the only completion Pi persists, so replay settles async runs from it.
|
||||
mapCustomMessage(message) {
|
||||
if (message.customType === "subagent-incremental-child-notify")
|
||||
return mapWorkflowChildNotice(message);
|
||||
if (message.customType !== "subagent-notify") return undefined;
|
||||
const text =
|
||||
typeof message.content === "string"
|
||||
? message.content
|
||||
: extractTextFromToolResult({ content: message.content });
|
||||
const lines = text?.split("\n") ?? [];
|
||||
const lines = noticeText(message)?.split("\n") ?? [];
|
||||
const status = noticeStatus(lines[0] ?? "");
|
||||
if (!status) return undefined;
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
@@ -192,8 +448,10 @@ export const piSubagents: PiExtension = {
|
||||
for (const line of lines) {
|
||||
const dir = line.match(/^Retention-managed async directory: (.+)$/)?.[1];
|
||||
if (dir) {
|
||||
owner = asyncRuns.get(dir)?.owner;
|
||||
if (owner) subagents.push({ type: "upsert", id: owner, status });
|
||||
const run = asyncRuns.get(dir);
|
||||
owner = run?.owner;
|
||||
if (!run || !owner) continue;
|
||||
subagents.push(...settleRunNotice(run, status));
|
||||
continue;
|
||||
}
|
||||
const file = owner ? line.match(/^Session file: (.+)$/)?.[1] : undefined;
|
||||
@@ -209,36 +467,26 @@ export const piSubagents: PiExtension = {
|
||||
poll() {
|
||||
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
|
||||
const childSessions: NonNullable<PiExtensionToolMapping["childSessions"]> = [];
|
||||
const now = Date.now();
|
||||
for (const [dir, run] of asyncRuns) {
|
||||
if (!run.polling) continue;
|
||||
let raw: unknown;
|
||||
try {
|
||||
raw = JSON.parse(readFileSync(join(dir, "status.json"), "utf8"));
|
||||
} catch {
|
||||
if (now >= run.deadline) {
|
||||
subagents.push(...giveUpRun(run));
|
||||
continue;
|
||||
}
|
||||
const parsed = AsyncStatus.safeParse(raw);
|
||||
if (!parsed.success) continue;
|
||||
const terminal = ["complete", "failed", "stopped"].includes(parsed.data.state);
|
||||
for (const [index, step] of parsed.data.steps.entries()) {
|
||||
const id = parsed.data.steps.length === 1 ? run.owner : `${run.owner}:${index}`;
|
||||
const status = asyncChildStatus(parsed.data.state, step.status);
|
||||
subagents.push({ type: "upsert", id, title: step.agent, status });
|
||||
if (step.sessionFile && !readSessions.has(step.sessionFile)) {
|
||||
readSessions.add(step.sessionFile);
|
||||
childSessions.push({ id, file: step.sessionFile });
|
||||
const status = readAsyncStatus(dir);
|
||||
if (!status) {
|
||||
run.misses += 1;
|
||||
if (run.misses >= POLL_MISS_LIMIT) {
|
||||
subagents.push(...giveUpRun(run));
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (terminal) {
|
||||
if (parsed.data.steps.length === 0) {
|
||||
subagents.push({
|
||||
type: "upsert",
|
||||
id: run.owner,
|
||||
status: parsed.data.state === "complete" ? "completed" : "failed",
|
||||
});
|
||||
}
|
||||
run.polling = false;
|
||||
}
|
||||
run.misses = 0;
|
||||
const output = projectAsyncStatus(run, status);
|
||||
subagents.push(...output.subagents);
|
||||
childSessions.push(...output.childSessions);
|
||||
}
|
||||
return subagents.length || childSessions.length ? { subagents, childSessions } : undefined;
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user