puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-10-24 02:02:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(pi): keep workflow lifecycle in the provider adapter

Browse Source 
Remove the shared manager sweep and poll marker from the workflow fix. Turn-end timing and history replay do not prove that a Pi child was canceled; preserve its provider status until the adapter supplies an outcome.
This commit is contained in:
paseo-bot[bot]
2026-10-17 18:13:17 +02:00
parent b38dbf8ae6
commit e5f8c7ac25
7 changed files with 94 additions and 822 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/agent-lifecycle.md
+1 -1
View File
@@ -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, 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.
- **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.
packages/server/src/server/agent/agent-manager.provider-subagents.test.ts
+74 -578
View File
@@ -3,430 +3,29 @@ import { tmpdir } from "node:os";
import { join } from "node:path";
import pino from "pino";
import { afterEach, describe, expect, test, vi } from "vitest";
import { createTestLogger } from "../../test-utils/test-logger.js";
import { AgentManager } from "./agent-manager.js";
import { PiRpcAgentClient } from "./providers/pi/agent.js";
import { FakePi } from "./providers/pi/test-utils/fake-pi.js";
import type {
AgentCapabilityFlags,
AgentClient,
AgentPersistenceHandle,
AgentPromptInput,
AgentProvider,
AgentRunResult,
AgentSession,
AgentSessionConfig,
AgentStreamEvent,
ProviderCatalog,
} from "./agent-sdk-types.js";
import type { ProviderSubagentInputEvent } from "./provider-subagents/store.js";
/**
* Stuck-running provider subagent cards are a lifecycle problem, not a presentation one: a row is
* born `running` and only a provider update can settle it. These cover the manager's two finalize
* points — the turn boundary and history replay — and, above all, that neither touches a row whose
* run can still report.
*/
const TEST_CAPABILITIES: AgentCapabilityFlags = {
supportsStreaming: true,
supportsSessionPersistence: true,
supportsDynamicModes: false,
supportsMcpServers: false,
supportsReasoningStream: false,
supportsToolInvocations: false,
};
const AGENT_ID = "00000000-0000-4000-8000-000000000001";
/** Long enough that no poll could have re-reported the row inside it. */
const QUIET_FOR_MS = 60_000;
class TestAgentSession implements AgentSession {
readonly capabilities = TEST_CAPABILITIES;
readonly id = "test-session";
private historyEvents: AgentStreamEvent[] = [];
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
constructor(readonly provider: AgentProvider) {}
async run(): Promise<AgentRunResult> {
return { sessionId: this.id, finalText: "", timeline: [] };
}
async startTurn(_prompt: AgentPromptInput): Promise<{ turnId: string }> {
return { turnId: "turn-1" };
}
subscribe(callback: (event: AgentStreamEvent) => void): () => void {
this.subscribers.add(callback);
return () => {
this.subscribers.delete(callback);
};
}
setHistory(events: AgentStreamEvent[]): void {
this.historyEvents = [...events];
}
pushEvent(event: AgentStreamEvent): void {
for (const callback of this.subscribers) callback(event);
}
async *streamHistory(): AsyncGenerator<AgentStreamEvent> {
for (const event of this.historyEvents) yield event;
}
async getRuntimeInfo() {
return { provider: this.provider, sessionId: this.id, model: null, modeId: null };
}
async getAvailableModes() {
return [];
}
async getCurrentMode() {
return null;
}
async setMode(): Promise<void> {}
getPendingPermissions() {
return [];
}
async respondToPermission(): Promise<void> {}
describePersistence(): AgentPersistenceHandle {
return { provider: this.provider, sessionId: this.id };
}
async interrupt(): Promise<void> {}
async close(): Promise<void> {}
}
class TestAgentClient implements AgentClient {
readonly capabilities = TEST_CAPABILITIES;
session?: TestAgentSession;
constructor(readonly provider: AgentProvider) {}
async createSession(config: AgentSessionConfig): Promise<AgentSession> {
this.session = new TestAgentSession(config.provider);
return this.session;
}
async resumeSession(): Promise<AgentSession> {
this.session ??= new TestAgentSession(this.provider);
return this.session;
}
async fetchCatalog(): Promise<ProviderCatalog> {
return {
models: [{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true }],
modes: [],
};
}
async isAvailable(): Promise<boolean> {
return true;
}
}
interface Harness {
manager: AgentManager;
session: TestAgentSession;
agentId: string;
cleanup: () => void;
}
async function createHarness(provider: AgentProvider): Promise<Harness> {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagents-"));
const client = new TestAgentClient(provider);
const manager = new AgentManager({
clients: { [provider]: client },
idFactory: () => AGENT_ID,
logger: createTestLogger(),
});
await manager.createAgent({ provider, cwd: workdir }, AGENT_ID, { workspaceId: undefined });
if (!client.session) throw new Error("Test client did not create a session");
return {
manager,
session: client.session,
agentId: AGENT_ID,
cleanup: () => rmSync(workdir, { recursive: true, force: true }),
};
}
function subagentEvent(
provider: AgentProvider,
event: ProviderSubagentInputEvent,
): AgentStreamEvent {
return { type: "provider_subagent", provider, event };
}
function turnCompleted(provider: AgentProvider): AgentStreamEvent {
return { type: "turn_completed", provider, turnId: "turn-1" };
}
async function drainSessionEvents(): Promise<void> {
for (let index = 0; index < 10_000; index++) await Promise.resolve();
}
async function pushEvent(harness: Harness, event: AgentStreamEvent): Promise<void> {
harness.session.pushEvent(event);
await drainSessionEvents();
}
function status(harness: Harness, subagentId: string) {
return harness.manager.getProviderSubagent(harness.agentId, subagentId)?.status;
}
/** Moves the clock past the liveness window without touching timers the manager runs on. */
function advanceClock(ms: number): void {
vi.useFakeTimers({ toFake: ["Date"] });
vi.setSystemTime(new Date(Date.now() + ms));
}
/** Waits on real time: the Pi adapter polls on a real interval. */
async function waitFor(condition: () => boolean): Promise<void> {
for (let attempt = 0; attempt < 250; attempt++) {
if (condition()) return;
await new Promise((resolve) => setTimeout(resolve, 20));
}
throw new Error("Timed out waiting for the Pi adapter poll");
throw new Error("Timed out waiting for Pi subagent output");
}
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();
}
});
afterEach(() => vi.useRealTimers());
describe("provider subagent lifecycle ownership", () => {
test(
"a turn end during a status-file miss streak leaves the live poll's row running",
"a delayed parent turn end preserves a Pi child until its provider settles it",
{ 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.
// Run the real Pi client and adapter with an injected runtime.
const pi = new FakePi();
const client = new PiRpcAgentClient({
logger: pino({ level: "silent" }),
@@ -468,18 +67,29 @@ describe("provider subagent finalize", () => {
},
});
const row = () => manager.getProviderSubagent(AGENT_ID, "call-1");
await waitFor(() => row()?.status === "running" && row()?.liveness === "poll");
await waitFor(() => row()?.status === "running");
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");
}
// 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);
@@ -492,170 +102,56 @@ describe("provider subagent finalize", () => {
},
);
test("a turn end leaves another provider's quiet row alone: it has no poll to prove life", async () => {
const harness = await createHarness("claude");
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 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,
await manager.createAgent({ provider: "pi", cwd: workdir }, AGENT_ID, {
workspaceId: undefined,
});
expect(harness.manager.getAgent(harness.agentId)?.historyPrimed).toBe(false);
await harness.manager.hydrateTimelineFromProvider(harness.agentId, { broadcast: true });
expect(status(harness, "call-prime")).toBe("canceled");
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 {
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();
await manager.closeAgent(AGENT_ID);
rmSync(workdir, { recursive: true, force: true });
}
});
});
packages/server/src/server/agent/agent-manager.ts
+6 -178
View File
@@ -726,17 +726,6 @@ function detachedAgentLabelPatch(labels: Record<string, string>): AgentLabelPatc
const RETIRED_INTERNAL_AGENT_TTL_MS = 10 * 60 * 1000;
const RETIRED_INTERNAL_AGENT_LIMIT = 500;
/**
* How recently a poll-backed Pi subagent row must have been updated to count as alive. Only the
* pi-subagents extension polls: it re-upserts every live row on each 250 ms follow tick and marks
* those upserts `liveness: "poll"`, so a marked row untouched for a whole window can never change
* again. The other Pi extensions (gotgenes/tintinweb pi-subagents and friends) are event-driven
* and may stay silent while a background child still runs, and Claude
* (providers/claude/agent.ts:4779,4805) and OpenCode (providers/opencode/v2/children.ts:64-68)
* settle their own rows at turn or child end. No row without the marker is ever swept.
*/
const PI_SUBAGENT_LIVENESS_MS = 5_000;
export interface RetiredInternalAgent {
record: ArchivedStoredAgentRecord;
lastMessage: string | null;
@@ -754,7 +743,6 @@ export class AgentManager {
private readonly agents = new Map<string, LiveManagedAgent>();
private readonly timelineStore = new InMemoryAgentTimelineStore();
private readonly providerSubagents = new ProviderSubagentStore();
private readonly providerSubagentSweepTimers = new Map<string, ReturnType<typeof setTimeout>>();
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
@@ -1824,157 +1812,15 @@ export class AgentManager {
if (subagent.status !== "running") {
continue;
}
this.applyProviderSubagentStatus(parentAgentId, subagent.provider, subagent.id, "canceled");
const event = this.providerSubagents.apply(parentAgentId, subagent.provider, {
type: "upsert",
id: subagent.id,
status: "canceled",
});
this.dispatch({ type: "provider_subagent", event });
}
}
private applyProviderSubagentStatus(
parentAgentId: string,
provider: AgentProvider,
subagentId: string,
status: ProviderSubagentDescriptor["status"],
): void {
const event = this.providerSubagents.apply(parentAgentId, provider, {
type: "upsert",
id: subagentId,
status,
});
this.dispatch({ type: "provider_subagent", event });
}
/**
* Poll-backed Pi subagent rows are born `running` and leave that state only through a provider
* update. A row whose poll is gone would keep its spinner forever, so a turn boundary settles
* only the rows that prove they had a poll and stopped reporting, as canceled — never completed,
* because nothing here proves they succeeded. A row a live poll keeps re-reporting is fresh and
* left alone, and a row without the marker belongs to an event-driven reporter that may still be
* running its child, so it is never swept.
*
* Gate: the poll marker plus no report inside the liveness window. The window is wall clock, so an
* event-loop stall, a GC pause, or a resume longer than it makes a live run's row look silent; this
* path is the only one that reads time — the replay settle (finalizeReplayedProviderSubagents) gates
* on the store's pre-replay state instead.
*
* A turn end can arrive inside that window, and for a spawn call a stop interrupted before it
* reported there is no later turn end to look again: the placeholder would stay `running` while the
* parent sits idle. So a row still inside the window schedules one re-check for the moment its
* silence becomes proof, instead of waiting for an event that may never come.
*
* That re-check is scoped to the rows it was armed for. A row born after it was armed belongs to a
* turn that is still running — a synchronous `subagent` call is poll-marked at spawn and only gets
* its poll when the result lands — so the timer must not judge it: that turn's own end sweeps it.
*/
// ponytail: staleness is wall-clock — upgrade to a poll heartbeat (sequence/epoch from the adapter) if stalls prove real
private finalizeUnprovenProviderSubagents(
agentId: string,
scheduledRowIds?: ReadonlySet<string>,
): void {
const now = Date.now();
let recheckAt: number | undefined;
const recheckRowIds: string[] = [];
for (const subagent of this.providerSubagents.list(agentId)) {
if (scheduledRowIds && !scheduledRowIds.has(subagent.id)) continue;
if (subagent.status !== "running" || subagent.liveness !== "poll") continue;
const updatedAt = Date.parse(subagent.updatedAt);
if (updatedAt >= now - PI_SUBAGENT_LIVENESS_MS) {
// A live poll may still own this row, so it stays. Look again once its silence is old enough.
recheckAt = Math.min(
recheckAt ?? Number.POSITIVE_INFINITY,
updatedAt + PI_SUBAGENT_LIVENESS_MS,
);
recheckRowIds.push(subagent.id);
continue;
}
this.applyProviderSubagentStatus(agentId, subagent.provider, subagent.id, "canceled");
}
this.scheduleProviderSubagentSweep(agentId, recheckAt, recheckRowIds);
}
/**
* The one pending re-check per agent: a turn end while a poll-backed row was still fresh reschedules
* it for the rows that were fresh then, and a sweep that finds nothing left to prove clears it. The
* timer stops itself once the agent is gone.
*/
private scheduleProviderSubagentSweep(
agentId: string,
recheckAt: number | undefined,
recheckRowIds: readonly string[],
): void {
const pending = this.providerSubagentSweepTimers.get(agentId);
if (pending) clearTimeout(pending);
this.providerSubagentSweepTimers.delete(agentId);
if (recheckAt === undefined) return;
const scheduledRowIds = new Set(recheckRowIds);
const timer = setTimeout(
() => {
this.providerSubagentSweepTimers.delete(agentId);
if (!this.agents.has(agentId)) return;
this.finalizeUnprovenProviderSubagents(agentId, scheduledRowIds);
},
Math.max(0, recheckAt - Date.now()) + 1,
);
timer.unref();
this.providerSubagentSweepTimers.set(agentId, timer);
}
/**
* Replay re-creates a row for every spawn in the transcript, including runs that never finished.
* Replay re-runs the pi-subagents adapter mapping, so a replayed running pi-subagents row carries
* the poll marker. Replay builds its host without `follow()`
* (providers/pi/extensions/host.ts:61-69), and the live host's `callsByRun` is per-adapter-session,
* so no live poll is behind that row and nothing would ever update it again: left running, it is a
* spinner that returns on every reload.
*
* Gate: the poll marker plus the row was not already `running` in the store before the replay. A
* row that was running pre-replay belongs to the live session's poll (rewind force-hydrates without
* recreating the session) and is left with it, or the settle would cancel a live run and the next
* tick would flip it back to running. Every other replayed running row settles as canceled, never
* completed: a row that was already canceled or completed before the replay settles again as
* canceled, because replay re-reports every unfinished run as running and the transcript holds no
* proof that it survived. That is the verdict for unproven rows, and it is still better than the
* spinner that stays until the next reload.
*
* Known limitation: a row replayed for a run that no live poll owns may still describe work that
* is writing status.json, and this cancel is final — such a run keeps working but its card can
* never recover. Making replay resumable needs the live host to adopt replayed runs, which this
* change does not attempt.
*
* Rows without the marker belong to a reporter that settles its own children — Claude
* (providers/claude/agent.ts:4779,4805), OpenCode (providers/opencode/v2/children.ts:64-68), the
* event-driven Pi extensions — so replay leaves them alone. A live turn is the other case where a
* replayed tool call may still be in flight, so it is left alone too. Normal callers clear the
* turn before hydrating, but the interrupt-timeout rescue (cancelAgentRunNow) settles the
* foreground run without a terminal event, so `activeTurnId` can linger; those rows are left to
* the next turn-end sweep rather than to a poll.
*/
// ponytail: replay has no live poll; proper fix is letting the live host adopt replayed runs — revisit when host replay shares adapter state
private finalizeReplayedProviderSubagents(
agent: ActiveManagedAgent,
replayedRunningIds: ReadonlySet<string>,
preexistingRunningIds: ReadonlySet<string>,
): void {
if (agent.activeForegroundTurnId != null || agent.activeTurnId != null) return;
for (const id of replayedRunningIds) {
if (preexistingRunningIds.has(id)) continue;
const subagent = this.providerSubagents.get(agent.id, id);
if (!subagent || subagent.status !== "running" || subagent.liveness !== "poll") continue;
this.applyProviderSubagentStatus(agent.id, subagent.provider, id, "canceled");
}
}
/**
* Ids the store reports running now. Before a replay applies, a running row belongs to the live
* session's poll; a terminal row belongs to nobody and is replayed evidence like any other.
*/
private providerSubagentRunningIds(agentId: string): Set<string> {
return new Set(
this.providerSubagents
.list(agentId)
.filter((subagent) => subagent.status === "running")
.map((subagent) => subagent.id),
);
}
async archiveAgent(agentId: string): Promise<{ archivedAt: string }> {
return this.runLifecycleMutation(agentId, () => this.archiveAgentUnlocked(agentId));
}
@@ -4300,20 +4146,13 @@ export class AgentManager {
this.timelineStore.initialize(agent.id, { timestamp: new Date().toISOString() });
agent.historyPrimed = true;
// Rows the store already reports running are the live session's: a rewind replays history
// without recreating the session, so their poll keeps running.
const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
for (const event of this.providerSubagents.deleteParent(agent.id)) {
if (broadcast) {
this.dispatch({ type: "provider_subagent", event });
}
}
const replayedRunningIds = new Set<string>();
for (const event of providerSubagentEvents) {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
if (event.event.type === "upsert" && event.event.status === "running") {
replayedRunningIds.add(event.event.id);
}
if (broadcast) {
this.dispatch({ type: "provider_subagent", event: update });
}
@@ -4332,7 +4171,6 @@ export class AgentManager {
});
}
}
this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
this.touchUpdatedAt(agent);
this.emitState(agent);
}
@@ -4379,15 +4217,8 @@ export class AgentManager {
row: AgentTimelineRow;
}> = [];
const providerSubagentEvents: AgentManagerEvent[] = [];
const replayedRunningIds = new Set<string>();
// A row the store still reports running is not a fresh resume: it belongs to the live
// session's poll.
const preexistingRunningRowIds = this.providerSubagentRunningIds(agent.id);
for (const event of historySubagentEvents) {
const update = this.providerSubagents.apply(agent.id, event.provider, event.event);
if (event.event.type === "upsert" && event.event.status === "running") {
replayedRunningIds.add(event.event.id);
}
const managerEvent: AgentManagerEvent = { type: "provider_subagent", event: update };
if (deferredBroadcast) {
providerSubagentEvents.push(managerEvent);
@@ -4414,7 +4245,6 @@ export class AgentManager {
agent.historyPrimed = true;
if (typeof broadcast !== "function" || !broadcast()) {
this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
return;
}
for (const event of providerSubagentEvents) {
@@ -4427,7 +4257,6 @@ export class AgentManager {
timestamp: row.timestamp,
});
}
this.finalizeReplayedProviderSubagents(agent, replayedRunningIds, preexistingRunningRowIds);
}
private notifyForegroundTurnWaiters(agentId: string, event: AgentStreamEvent): void {
@@ -4513,7 +4342,6 @@ export class AgentManager {
if (isForegroundEvent) {
this.finalizeForegroundTurn(agent, eventTurnId);
}
this.finalizeUnprovenProviderSubagents(agent.id);
}
if (flags.shouldDispatchEvent) {
packages/server/src/server/agent/provider-subagents/store.test.ts
-17
View File
@@ -60,23 +60,6 @@ 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);
packages/server/src/server/agent/provider-subagents/store.ts
-9
View File
@@ -23,11 +23,6 @@ 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 =
@@ -45,8 +40,6 @@ 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;
}
| {
@@ -120,7 +113,6 @@ 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,
@@ -134,7 +126,6 @@ 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 };
packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts
+8 -17
View File
@@ -300,7 +300,6 @@ describe("pi-subagents adapter", () => {
]);
// 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();
});
@@ -391,13 +390,11 @@ describe("pi-subagents adapter", () => {
}),
);
await expect.poll(() => reportedUpsert(received, "call-wf:beta")).toBe(true);
expect(
subagentUpserts(received).map((event) => [event.id, event.status, event.liveness]),
).toEqual(
expect(subagentUpserts(received).map((event) => [event.id, event.status])).toEqual(
expect.arrayContaining([
["call-wf:alpha", "running", "poll"],
["call-wf:beta", "running", "poll"],
["call-wf", "running", "poll"],
["call-wf:alpha", "running"],
["call-wf:beta", "running"],
["call-wf", "running"],
]),
);
@@ -443,12 +440,9 @@ describe("pi-subagents adapter", () => {
},
});
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.
// A missing status file supplies no new outcome before the miss budget expires.
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);
expect(duringMisses).toEqual([]);
// 60 s of missing ticks, the budget before a status directory counts as gone.
await vi.advanceTimersByTimeAsync(250 * 236);
@@ -458,7 +452,6 @@ describe("pi-subagents adapter", () => {
type: "upsert",
id: "call-lost",
status: "canceled",
liveness: "poll",
});
expect(afterGiveUp.filter((event) => event.status === "canceled")).toHaveLength(1);
} finally {
@@ -987,7 +980,6 @@ describe("pi-subagents adapter", () => {
id: "call-wf:alpha",
title: "scout",
status: "completed",
liveness: "poll",
},
]);
@@ -1004,8 +996,7 @@ describe("pi-subagents adapter", () => {
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.
// The launch placeholder is replaced once the workflow supplies its child rows.
expect(
host.mapToolCall({
callId: "fanout",
@@ -1014,7 +1005,7 @@ describe("pi-subagents adapter", () => {
status: "running",
result: null,
})?.subagents,
).toEqual([expect.objectContaining({ id: "fanout", status: "running", liveness: "poll" })]);
).toEqual([expect.objectContaining({ id: "fanout", status: "running" })]);
expect(
host.mapToolCall({
callId: "fanout",
packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts
+5 -22
View File
@@ -205,7 +205,7 @@ export const piSubagents: PiExtension = {
const subagents: NonNullable<PiExtensionToolMapping["subagents"]> = [];
for (const id of new Set([run.owner, ...run.rows])) {
if (settledRows.has(id)) continue;
subagents.push({ type: "upsert", id, status: "canceled", liveness: "poll" });
subagents.push({ type: "upsert", id, status: "canceled" });
}
return subagents;
};
@@ -305,7 +305,6 @@ export const piSubagents: PiExtension = {
id,
title: step.agent,
status: childStatus,
liveness: "poll" as const,
});
} else if (!namedRows.has(id) || childSession) {
// Attach a late transcript with its known verdict, so the host reads it once instead of
@@ -316,7 +315,6 @@ export const piSubagents: PiExtension = {
id,
title: step.agent,
status: settledStatus,
liveness: "poll" as const,
});
}
if (childSession) {
@@ -326,7 +324,7 @@ export const piSubagents: PiExtension = {
}
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.
// 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)) {
@@ -336,7 +334,6 @@ export const piSubagents: PiExtension = {
type: "upsert",
id: run.owner,
status: ownerStatus,
liveness: "poll",
});
}
return { subagents, childSessions };
@@ -368,13 +365,13 @@ export const piSubagents: PiExtension = {
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.
// 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, liveness: "poll" as const }] };
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.
@@ -403,13 +400,11 @@ export const piSubagents: PiExtension = {
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: "running" as const, liveness: "poll" as const }],
subagents: [{ ...base, status: "running" as const }],
};
return {
detail,
@@ -486,18 +481,6 @@ export const piSubagents: PiExtension = {
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;