From e5f8c7ac256f4152613eb7d2cb0ff5e427f1ef7e Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 18:13:17 +0200 Subject: [PATCH] fix(pi): keep workflow lifecycle in the provider adapter 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. --- docs/agent-lifecycle.md | 2 +- .../agent-manager.provider-subagents.test.ts | 652 ++---------------- .../server/src/server/agent/agent-manager.ts | 184 +---- .../agent/provider-subagents/store.test.ts | 17 - .../server/agent/provider-subagents/store.ts | 9 - .../pi/extensions/pi-subagents/index.test.ts | 25 +- .../pi/extensions/pi-subagents/index.ts | 27 +- 7 files changed, 94 insertions(+), 822 deletions(-) diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index f5fbda4d48..a04116a93e 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -210,7 +210,7 @@ The rows combine two kinds of children: parentAgentId === thisAgent.id AND !archivedAt ``` -- **Provider subagents** are child executions owned by Claude, Codex, 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. diff --git a/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts b/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts index 36ac1b3c0d..929d5c1f6d 100644 --- a/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts +++ b/packages/server/src/server/agent/agent-manager.provider-subagents.test.ts @@ -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 { - return { sessionId: this.id, finalText: "", timeline: [] }; - } - - async startTurn(_prompt: AgentPromptInput): Promise<{ turnId: string }> { - return { turnId: "turn-1" }; - } - - subscribe(callback: (event: AgentStreamEvent) => void): () => void { - this.subscribers.add(callback); - return () => { - this.subscribers.delete(callback); - }; - } - - setHistory(events: AgentStreamEvent[]): void { - this.historyEvents = [...events]; - } - - pushEvent(event: AgentStreamEvent): void { - for (const callback of this.subscribers) callback(event); - } - - async *streamHistory(): AsyncGenerator { - for (const event of this.historyEvents) yield event; - } - - async getRuntimeInfo() { - return { provider: this.provider, sessionId: this.id, model: null, modeId: null }; - } - - async getAvailableModes() { - return []; - } - - async getCurrentMode() { - return null; - } - - async setMode(): Promise {} - - getPendingPermissions() { - return []; - } - - async respondToPermission(): Promise {} - - describePersistence(): AgentPersistenceHandle { - return { provider: this.provider, sessionId: this.id }; - } - - async interrupt(): Promise {} - - async close(): Promise {} -} - -class TestAgentClient implements AgentClient { - readonly capabilities = TEST_CAPABILITIES; - session?: TestAgentSession; - - constructor(readonly provider: AgentProvider) {} - - async createSession(config: AgentSessionConfig): Promise { - this.session = new TestAgentSession(config.provider); - return this.session; - } - - async resumeSession(): Promise { - this.session ??= new TestAgentSession(this.provider); - return this.session; - } - - async fetchCatalog(): Promise { - return { - models: [{ provider: this.provider, id: "test-model", label: "Test Model", isDefault: true }], - modes: [], - }; - } - - async isAvailable(): Promise { - return true; - } -} - -interface Harness { - manager: AgentManager; - session: TestAgentSession; - agentId: string; - cleanup: () => void; -} - -async function createHarness(provider: AgentProvider): Promise { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provider-subagents-")); - const client = new TestAgentClient(provider); - const manager = new AgentManager({ - clients: { [provider]: client }, - idFactory: () => AGENT_ID, - logger: createTestLogger(), - }); - await manager.createAgent({ provider, cwd: workdir }, AGENT_ID, { workspaceId: undefined }); - if (!client.session) throw new Error("Test client did not create a session"); - return { - manager, - session: client.session, - agentId: AGENT_ID, - cleanup: () => rmSync(workdir, { recursive: true, force: true }), - }; -} - -function subagentEvent( - provider: AgentProvider, - event: ProviderSubagentInputEvent, -): AgentStreamEvent { - return { type: "provider_subagent", provider, event }; -} - -function turnCompleted(provider: AgentProvider): AgentStreamEvent { - return { type: "turn_completed", provider, turnId: "turn-1" }; -} - async function drainSessionEvents(): Promise { for (let index = 0; index < 10_000; index++) await Promise.resolve(); } - -async function pushEvent(harness: Harness, event: AgentStreamEvent): Promise { - harness.session.pushEvent(event); - await drainSessionEvents(); -} - -function status(harness: Harness, subagentId: string) { - return harness.manager.getProviderSubagent(harness.agentId, subagentId)?.status; -} - -/** Moves the clock past the liveness window without touching timers the manager runs on. */ -function advanceClock(ms: number): void { - vi.useFakeTimers({ toFake: ["Date"] }); - vi.setSystemTime(new Date(Date.now() + ms)); -} - -/** Waits on real time: the Pi adapter polls on a real interval. */ async function waitFor(condition: () => boolean): Promise { for (let attempt = 0; attempt < 250; attempt++) { if (condition()) return; await new Promise((resolve) => setTimeout(resolve, 20)); } - throw new Error("Timed out waiting for the Pi adapter poll"); + 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 }); } }); }); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 05fe98cd02..ff91ef85ef 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -726,17 +726,6 @@ function detachedAgentLabelPatch(labels: Record): AgentLabelPatc const RETIRED_INTERNAL_AGENT_TTL_MS = 10 * 60 * 1000; const RETIRED_INTERNAL_AGENT_LIMIT = 500; -/** - * How recently a poll-backed Pi subagent row must have been updated to count as alive. Only the - * pi-subagents extension polls: it re-upserts every live row on each 250 ms follow tick and marks - * those upserts `liveness: "poll"`, so a marked row untouched for a whole window can never change - * again. The other Pi extensions (gotgenes/tintinweb pi-subagents and friends) are event-driven - * and may stay silent while a background child still runs, and Claude - * (providers/claude/agent.ts:4779,4805) and OpenCode (providers/opencode/v2/children.ts:64-68) - * settle their own rows at turn or child end. No row without the marker is ever swept. - */ -const PI_SUBAGENT_LIVENESS_MS = 5_000; - export interface RetiredInternalAgent { record: ArchivedStoredAgentRecord; lastMessage: string | null; @@ -754,7 +743,6 @@ export class AgentManager { private readonly agents = new Map(); private readonly timelineStore = new InMemoryAgentTimelineStore(); private readonly providerSubagents = new ProviderSubagentStore(); - private readonly providerSubagentSweepTimers = new Map>(); private readonly agentsAwaitingInitialSnapshotPersist = new Set(); private readonly sessionEventTails = new Map>(); private readonly steerEventBarriers = new Map(); @@ -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, - ): void { - const now = Date.now(); - let recheckAt: number | undefined; - const recheckRowIds: string[] = []; - for (const subagent of this.providerSubagents.list(agentId)) { - if (scheduledRowIds && !scheduledRowIds.has(subagent.id)) continue; - if (subagent.status !== "running" || subagent.liveness !== "poll") continue; - const updatedAt = Date.parse(subagent.updatedAt); - if (updatedAt >= now - PI_SUBAGENT_LIVENESS_MS) { - // A live poll may still own this row, so it stays. Look again once its silence is old enough. - recheckAt = Math.min( - recheckAt ?? Number.POSITIVE_INFINITY, - updatedAt + PI_SUBAGENT_LIVENESS_MS, - ); - recheckRowIds.push(subagent.id); - continue; - } - this.applyProviderSubagentStatus(agentId, subagent.provider, subagent.id, "canceled"); - } - this.scheduleProviderSubagentSweep(agentId, recheckAt, recheckRowIds); - } - - /** - * The one pending re-check per agent: a turn end while a poll-backed row was still fresh reschedules - * it for the rows that were fresh then, and a sweep that finds nothing left to prove clears it. The - * timer stops itself once the agent is gone. - */ - private scheduleProviderSubagentSweep( - agentId: string, - recheckAt: number | undefined, - recheckRowIds: readonly string[], - ): void { - const pending = this.providerSubagentSweepTimers.get(agentId); - if (pending) clearTimeout(pending); - this.providerSubagentSweepTimers.delete(agentId); - if (recheckAt === undefined) return; - const scheduledRowIds = new Set(recheckRowIds); - const timer = setTimeout( - () => { - this.providerSubagentSweepTimers.delete(agentId); - if (!this.agents.has(agentId)) return; - this.finalizeUnprovenProviderSubagents(agentId, scheduledRowIds); - }, - Math.max(0, recheckAt - Date.now()) + 1, - ); - timer.unref(); - this.providerSubagentSweepTimers.set(agentId, timer); - } - - /** - * Replay re-creates a row for every spawn in the transcript, including runs that never finished. - * Replay re-runs the pi-subagents adapter mapping, so a replayed running pi-subagents row carries - * the poll marker. Replay builds its host without `follow()` - * (providers/pi/extensions/host.ts:61-69), and the live host's `callsByRun` is per-adapter-session, - * so no live poll is behind that row and nothing would ever update it again: left running, it is a - * spinner that returns on every reload. - * - * Gate: the poll marker plus the row was not already `running` in the store before the replay. A - * row that was running pre-replay belongs to the live session's poll (rewind force-hydrates without - * recreating the session) and is left with it, or the settle would cancel a live run and the next - * tick would flip it back to running. Every other replayed running row settles as canceled, never - * completed: a row that was already canceled or completed before the replay settles again as - * canceled, because replay re-reports every unfinished run as running and the transcript holds no - * proof that it survived. That is the verdict for unproven rows, and it is still better than the - * spinner that stays until the next reload. - * - * Known limitation: a row replayed for a run that no live poll owns may still describe work that - * is writing status.json, and this cancel is final — such a run keeps working but its card can - * never recover. Making replay resumable needs the live host to adopt replayed runs, which this - * change does not attempt. - * - * Rows without the marker belong to a reporter that settles its own children — Claude - * (providers/claude/agent.ts:4779,4805), OpenCode (providers/opencode/v2/children.ts:64-68), the - * event-driven Pi extensions — so replay leaves them alone. A live turn is the other case where a - * replayed tool call may still be in flight, so it is left alone too. Normal callers clear the - * turn before hydrating, but the interrupt-timeout rescue (cancelAgentRunNow) settles the - * foreground run without a terminal event, so `activeTurnId` can linger; those rows are left to - * the next turn-end sweep rather than to a poll. - */ - // ponytail: replay has no live poll; proper fix is letting the live host adopt replayed runs — revisit when host replay shares adapter state - private finalizeReplayedProviderSubagents( - agent: ActiveManagedAgent, - replayedRunningIds: ReadonlySet, - preexistingRunningIds: ReadonlySet, - ): void { - if (agent.activeForegroundTurnId != null || agent.activeTurnId != null) return; - for (const id of replayedRunningIds) { - if (preexistingRunningIds.has(id)) continue; - const subagent = this.providerSubagents.get(agent.id, id); - if (!subagent || subagent.status !== "running" || subagent.liveness !== "poll") continue; - this.applyProviderSubagentStatus(agent.id, subagent.provider, id, "canceled"); - } - } - - /** - * Ids the store reports running now. Before a replay applies, a running row belongs to the live - * session's poll; a terminal row belongs to nobody and is replayed evidence like any other. - */ - private providerSubagentRunningIds(agentId: string): Set { - return new Set( - this.providerSubagents - .list(agentId) - .filter((subagent) => subagent.status === "running") - .map((subagent) => subagent.id), - ); - } - async archiveAgent(agentId: string): Promise<{ archivedAt: string }> { return this.runLifecycleMutation(agentId, () => this.archiveAgentUnlocked(agentId)); } @@ -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(); 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(); - // 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) { diff --git a/packages/server/src/server/agent/provider-subagents/store.test.ts b/packages/server/src/server/agent/provider-subagents/store.test.ts index 4496571908..dc71c0f5c6 100644 --- a/packages/server/src/server/agent/provider-subagents/store.test.ts +++ b/packages/server/src/server/agent/provider-subagents/store.test.ts @@ -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); diff --git a/packages/server/src/server/agent/provider-subagents/store.ts b/packages/server/src/server/agent/provider-subagents/store.ts index 98862bdcb0..b04246054a 100644 --- a/packages/server/src/server/agent/provider-subagents/store.ts +++ b/packages/server/src/server/agent/provider-subagents/store.ts @@ -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 }; diff --git a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts index f21e16ae96..2c3734c99b 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts +++ b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.test.ts @@ -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", diff --git a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts index 59cd96f8a9..854416429b 100644 --- a/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts +++ b/packages/server/src/server/agent/providers/pi/extensions/pi-subagents/index.ts @@ -205,7 +205,7 @@ export const piSubagents: PiExtension = { const subagents: NonNullable = []; for (const id of new Set([run.owner, ...run.rows])) { if (settledRows.has(id)) continue; - subagents.push({ type: "upsert", id, status: "canceled", liveness: "poll" }); + 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;