From 80879eaf7c391fc20ae4d2f272e18a8e71ac09b9 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 10 Nov 2026 13:09:38 +0100 Subject: [PATCH] fix(timeline): retain child state and recover rewinds on resume Provider children now live in the parent's timeline next to its history, so their descriptors, rows and retirements survive close, archive and a manager rebuilt over the same backend, and deleting the agent removes them all. Resuming settles a rewind that was still open when the runtime stopped instead of leaving its rows half-retired. A stored child's status is only the last report, possibly from provider history or an earlier runtime. Workspace activity and idle cleanup now read the children the current runtime reported running, and closing a runtime cancels only those, so a child stored as running from an earlier runtime no longer keeps the agent busy or blocks cleanup. --- CLAUDE.md | 1 + docs/agent-lifecycle.md | 9 +- docs/architecture.md | 1 + docs/data-model.md | 8 + docs/timeline-sync.md | 73 +- .../src/server/agent/agent-manager.test.ts | 217 +++- .../server/src/server/agent/agent-manager.ts | 115 ++- .../src/server/agent/plugin-provider.test.ts | 14 +- .../agent/projected-timeline-contract.test.ts | 4 +- .../agent/provider-subagents/store.test.ts | 51 +- .../server/agent/provider-subagents/store.ts | 149 ++- .../providers/codex-app-server-agent.test.ts | 2 +- .../src/server/agent/rewind/rewind.test.ts | 115 ++- .../agent/rewind/test-rewind-session.ts | 12 + .../agent/test-utils/plugin-stream-memory.ts | 23 +- .../agent/timeline/agent-timeline.test.ts | 947 ++++++++++-------- .../server/src/server/agent/timeline/index.ts | 1 + .../agent/timeline/internal/agent-timeline.ts | 92 +- .../server/agent/timeline/internal/backend.ts | 32 +- .../agent/timeline/internal/memory-backend.ts | 67 +- .../src/server/agent/timeline/testing.ts | 77 +- packages/server/src/server/session.ts | 7 +- 22 files changed, 1386 insertions(+), 631 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 71682af7e9..ea2ca57a4f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -53,6 +53,7 @@ At the start of non-trivial work, list `docs/` and skim anything relevant to the | [docs/permissions.md](docs/permissions.md) | Semantic daemon permissions, principals, credentials, pairing invitations, and Hub authority | | [docs/terminal-performance.md](docs/terminal-performance.md) | Terminal latency pipeline, coalescing/backpressure invariants, benchmark + perf spec usage | | [docs/terminal-rendering.md](docs/terminal-rendering.md) | Native terminal view: xterm owns the grid, cluster placement, emoji widths, mixed-direction limits, render regression | +| [docs/timeline-sync.md](docs/timeline-sync.md) | Owned agent history, rewind retirements, live stream vs authoritative fetch, paged catch-up, client replica | | [docs/agent-stream-performance.md](docs/agent-stream-performance.md) | Assistant text pipeline — coalescing window, paced reveal, why arrival lumps are smoothed at render | | [docs/markdown-renderer.md](docs/markdown-renderer.md) | Shared Markdown engine, native selection, links, images and library constraints | | [docs/native-text-selection.md](docs/native-text-selection.md) | Native diagram grouping, selection boundaries and leading inline image limitations | diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 2c2ce29403..253b89de90 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -38,8 +38,8 @@ An unarchived agent may be `closed` without being deleted or archived. Closing r processes and subscriptions while retaining its Paseo identity, persistence handle, timeline, workspace, labels, title, usage, attention, timestamps, and parent relationship. Opening or prompting the agent runs through `ensureAgentLoaded()`, which resumes the durable provider session under the -same Paseo agent ID. Provider history is not appended again when the canonical timeline is already -primed. +same Paseo agent ID. Resume does not read provider history when the agent already has a timeline; +see [owned history](timeline-sync.md#owned-history). Reload releases the old runtime before resuming its durable session: an idle provider process can still own an exclusive writer. A close failure retains that runtime for cleanup and blocks the @@ -213,7 +213,8 @@ lifecycle is separate from public workspace visibility and is not a public creat Archive is a **soft delete**: the agent record stays on disk with `archivedAt` set, the runtime is closed, and the agent disappears from active lists. Archive is **global** — it lives on the server and propagates to every connected client. Archive sets `archivedAt`, invokes the provider's native archive hook, and cascades to managed -children. +children. The timeline and the provider children stay, so an archived chat reads them without its +provider; deleting the agent removes them. `create_agent_request` can opt an agent into `autoArchive`. In that mode the daemon archives the agent after the first terminal turn event (`turn_completed`, `turn_failed`, or `turn_canceled`). When the agent owns an isolated workspace, auto-archive archives that workspace too; the managed worktree is removed when its final workspace reference is gone. @@ -285,6 +286,8 @@ Workspace status is an aggregate activity signal computed **per `workspaceId`**. Running provider-native subagents contribute `running` to the workspace owned by their parent agent. Their completed, failed, and canceled states stay in the parent's subagents track. +Only the parent's current runtime makes a provider child running. A child's stored status is the last report about it, and that report can come from provider history or from a runtime an earlier daemon ran, so a stored `running` child does not make the workspace running or keep the parent's runtime alive. It counts again once the live session reports it running. Closing a runtime cancels the children that runtime reported running; children it never reported keep their stored status. + A finished workspace can be marked unread after it has been reviewed. The daemon restores `finished` attention on its newest eligible workspace-root agent without sending a new completion notification. Opening the workspace clears that attention through the normal focus flow. diff --git a/docs/architecture.md b/docs/architecture.md index a7ad69c14d..6409161670 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -94,6 +94,7 @@ workspace for a directory clears it. The store lives in memory and resets on dae | `server/directory-sync/` | Daemon-global latest-state sequences for projects, workspaces, and agents | | `server/workspace-labels/` | Host-local label catalog, assignment mutations, and explicit subscriptions | | `server/agent/agent-manager.ts` | Agent lifecycle state machine, timeline tracking, subscriber management | +| `server/agent/timeline/` | Owned agent history, provider children, rewind retirements, behind a backend | | `server/agent/agent-storage.ts` | File-backed JSON persistence at `$PASEO_HOME/agents/` | | `server/agent/tools/` | Transport-neutral catalog for workspaces, agents, permissions, and automation | | `server/agent/mcp-server.ts` | Thin MCP adapter that registers the Paseo tool catalog with the MCP SDK | diff --git a/docs/data-model.md b/docs/data-model.md index dc65096b66..39d39e0ce7 100644 --- a/docs/data-model.md +++ b/docs/data-model.md @@ -175,6 +175,14 @@ Terminal activity contributes to the workspace status bucket **per `workspaceId` --- +## Runtime-only Agent Timelines + +Agent timelines are not written to `$PASEO_HOME`. `AgentTimeline` owns each agent's history behind +a backend, and the only backend is in memory, so a daemon restart loses it and the next resume +imports provider history. See [owned history](timeline-sync.md#owned-history). + +--- + ## 2. Daemon Configuration **Path:** `$PASEO_HOME/config.json` diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 3d6af249bf..de62d99491 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -5,15 +5,9 @@ Agent chat delivery has two paths: 1. **Live stream** — `agent_stream` WebSocket messages for immediacy. These may be delta-shaped lifecycle updates. 2. **Authoritative history** — `fetch_agent_timeline_request` for correctness. This always returns full projected timeline items, never lifecycle deltas. -The daemon retains projected items in memory. Each source event advances the stream sequence, -then replaces the previous tool state or merges into the current text item. Intermediate payloads -are never retained for history or catch-up. Provider history is the durable transcript authority -and rebuilds the projection when an agent resumes. - -Pagination bounds each read, not total resident history. Active and idle agents retain their -projected transcript; archive releases it and later history reads rebuild it from the provider. -Streaming overhead must remain proportional to the retained transcript. Cumulative text snapshots -must be released after their deltas have been consumed. +Pagination bounds each read, not total resident history. Streaming overhead must remain +proportional to the retained transcript. Cumulative text snapshots must be released after their +deltas have been consumed. The invariants are: @@ -32,6 +26,44 @@ bounded item is used for runtime timeline rows and live stream events. Provider applies the same rule so reopening an agent cannot restore an oversized payload. Fetches apply it again to assistant and reasoning rows the projection merges from many chunks. +## Owned history + +The daemon owns each agent's history in `AgentTimeline` +(`packages/server/src/server/agent/timeline/`): one ordered history per agent, with each provider +child's descriptor and history in a child scope of the same agent. Each source event advances the +stream sequence, then replaces the previous tool state or merges into the current text item. +Intermediate payloads are never retained for history or catch-up. + +History is append-only: rewind, import, reload and archive never delete or rewrite a row. Only +deleting the agent, or a provider removing one of its own children, deletes history. + +- **Rewind** retires a range of rows. They leave the conversation and stay stored, in the same + epoch. See [Rewind retires rows](#rewind-retires-rows). +- **Provider history** is imported only into an agent that has no history. After that, resume + does not read it. Reload agent and Unarchive append only the rows the provider has after the + owned conversation, as ordinary live rows. Any other difference, including an answer Paseo saw only part of or a prompt sent with + attachments, changes nothing and is reported as diverged. A provider child Paseo already holds, + listed or retired, is never imported again. +- **Archive and reload** keep the history and the provider children. An archived chat reads them + without its provider. Deleting the agent deletes its history and every child's, retired ones + included. + +`epoch` is the stored history's identity, minted once when the history is created. + +`AgentTimeline` serializes every operation of one agent, child scopes included, and keeps its +state in an `AgentTimelineBackend`: the sequence, the rows and retirements, prompt identities, the +open rewind, and the children. The backend contract suite is `agent-timeline.test.ts`, run over +`TIMELINE_CONTRACT_BACKENDS` in `timeline/testing.ts`. The only backend keeps everything in +memory, so a daemon restart loses the owned history: the next resume imports provider history into +a new epoch, and clients take the destructive reset below. Rewinds retired before the restart come +back if the provider history still holds them. Retained history costs memory for every loaded, +closed or archived agent until the daemon exits or the agent is deleted. + +An open rewind is stored with the history. Resuming an agent whose history holds one settles it +before the agent takes prompts: a recorded provider cut is committed, and an unknown outcome is +judged from the resumed provider history. When that history proves nothing, the rewind stays open +and the agent refuses prompts with the unknown-outcome error. + ## Presence is not delivery Client heartbeat reports presence: @@ -126,7 +158,7 @@ or retry backoff. This cannot keep a mobile socket alive after the operating sys bookkeeping without replacing timeline arrays, preserving an upward-scrolled viewport. - When the page overlaps or is adjacent to the local end cursor, only projected items newer than that cursor are applied. Already-covered rows are not replayed. -- A true middle gap, epoch change, or rewind atomically replaces stale canonical history with the +- A true middle gap or epoch change atomically replaces stale canonical history with the latest tail. The replacement reconciles positioned live rows beyond its coverage and unresolved local submissions; it never retains two discontiguous canonical ranges. @@ -139,6 +171,24 @@ history replay. The pending approval UI can hide that tool from presentation, bu must retain its position. Creating a new history card on rejection places it after the prompt that rejected it; changing steer-event ordering would also put new assistant output before that prompt. +## Rewind retires rows + +A conversation rewind commits a retirement `{ seq, retireFromSeq }`: active rows anchored in +`[retireFromSeq, seq)` leave the conversation, and `seq` is the retirement's own position in the +sequence. The provider children spawned by retired tool calls, and their descendants, are retired +in the same commit. They leave the listing, keep their history, and drop late provider events. + +Clients learn of a retirement live through `agent.timeline.rewind`, and on pages through +`retirements`, which lists the retirements whose `seq` lies inside the page's span. The client +drops the rows it holds in that range; rows without a timeline position are unechoed local sends +and stay. Retired ranges and the retirement's `seq` count as covered, so forward cursors advance +through them. Tail, before and prompt-jump pages are served from active rows, so one served before +a retirement the client already applied is stale and is refetched (`viewed-timeline-sync.ts`). + +Clients without `timeline_rewind_operations` cannot drop retired rows. For them the session derives +a wire epoch that changes with each retirement, so they take the destructive replacement they +always took on rewind. The shim is tagged `COMPAT(timelineRewindOperations)`. + ## Provider child history Child transcripts use the same projected-page reconciliation as the main conversation. The client @@ -303,7 +353,7 @@ the draft-create handoff is the one boundary that also permits the legacy canoni arrived first. Generic reducers and consumers do not reimplement message identity matching. Ordinary bootstrap, same-epoch reset, and catch-up replacement preserve unmatched locally submitted -rows because a provider may never echo them. A known epoch change or rewind replaces history and +rows because a provider may never echo them. A known epoch change replaces history and drops acknowledged local rows omitted by the new canonical epoch; every transaction not yet acknowledged by the provider, and no other local row, crosses that destructive boundary. A cold reset without an existing epoch is destructive because the client has no continuity anchor. @@ -319,6 +369,7 @@ canonical assistant prefix, it stays in the head lane. No row may be returned in ## Relevant code +- Server owned history: `packages/server/src/server/agent/timeline/` - Server live stream forwarding: `packages/server/src/server/session.ts` - App sync planning: `packages/app/src/timeline/timeline-sync-plan.ts` - App viewed-agent synchronization: `packages/app/src/timeline/viewed-timeline-sync.ts` diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index e5aa5c6a3e..07d6514b0c 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -811,6 +811,96 @@ test("Reload agent keeps the history and its identity, and adds only what the pr } }); +test("Reload agent adds the provider children of new turns and leaves the children Paseo shows alone", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-reload-children-")); + const providerHistory: AgentStreamEvent[] = [ + { + type: "timeline", + provider: "codex", + item: { type: "user_message", text: "explore", messageId: "provider-1" }, + }, + { + type: "provider_subagent", + provider: "codex", + event: { type: "upsert", id: "explorer", title: "Explorer", status: "completed" }, + }, + { + type: "provider_subagent", + provider: "codex", + event: { + type: "timeline", + id: "explorer", + item: { type: "assistant_message", text: "found it" }, + }, + }, + { type: "timeline", provider: "codex", item: { type: "assistant_message", text: "done" } }, + ]; + class HistorySession extends TestAgentSession { + override async *streamHistory(): AsyncGenerator { + yield* providerHistory; + } + } + class HistoryClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + return new HistorySession(config); + } + + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new HistorySession({ provider: "codex", cwd: config?.cwd ?? workdir }); + } + } + const manager = new AgentManager({ clients: { codex: new HistoryClient() }, logger }); + let agentId: string | null = null; + try { + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + agentId = agent.id; + expect(await reloadAgent(manager, agent.id)).toEqual({ kind: "appended" }); + + // The conversation was continued outside Paseo, with a second child in the new turn. + providerHistory.push( + { + type: "timeline", + provider: "codex", + item: { type: "user_message", text: "review", messageId: "provider-2" }, + }, + { + type: "provider_subagent", + provider: "codex", + event: { type: "upsert", id: "reviewer", title: "Reviewer", status: "completed" }, + }, + { + type: "provider_subagent", + provider: "codex", + event: { + type: "timeline", + id: "reviewer", + item: { type: "assistant_message", text: "looks good" }, + }, + }, + ); + expect(await reloadAgent(manager, agent.id)).toEqual({ kind: "appended" }); + + const childTexts = async (subagentId: string) => + (await manager.fetchProviderSubagentTimeline(agent.id, subagentId)).rows.map((row) => + row.item.type === "assistant_message" ? row.item.text : row.item.type, + ); + expect((await manager.listProviderSubagents(agent.id)).map((child) => child.id)).toEqual([ + "explorer", + "reviewer", + ]); + expect(await childTexts("explorer")).toEqual(["found it"]); + expect(await childTexts("reviewer")).toEqual(["looks good"]); + } finally { + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("Reload agent keeps the history when the provider history differs or cannot be read", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-reload-fail-")); let providerHistory: AgentTimelineItem[] = [ @@ -1883,6 +1973,14 @@ function fakeCodexEmitting(args: FakeCodexEmitterArgs): AgentClient { } /** Resumes a codex agent whose history Paseo does not hold, so it imports the provider's. */ +function runtimeIsCollectable(manager: AgentManager, agentId: string): boolean { + const candidate = manager + .getRuntimeCleanupCandidates() + .find((runtime) => runtime.agentId === agentId); + if (!candidate) throw new Error(`No runtime for agent '${agentId}'`); + return candidate.isEligible(); +} + function resumeWithProviderHistory(manager: AgentManager, cwd: string) { return manager.resumeAgentFromPersistence( { provider: "codex", sessionId: "provider-history" }, @@ -4662,7 +4760,7 @@ test("importProviderSession imports the selected session without listing and pub }, }), ]); - expect(manager.listProviderSubagents(imported.id)).toEqual([ + expect(await manager.listProviderSubagents(imported.id)).toEqual([ expect.objectContaining({ id: "thread-child", title: "Imported child", status: "completed" }), ]); expect((await manager.fetchProviderSubagentTimeline(imported.id, "thread-child")).rows).toEqual([ @@ -4881,11 +4979,13 @@ test("Reload agent keeps provider children and ends the running ones", async () provider: "codex", event: { type: "upsert", id: "stale-child", title: "Stale child", status: "running" }, }); - await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1)); + await vi.waitFor(async () => + expect(await manager.listProviderSubagents(snapshot.id)).toHaveLength(1), + ); await reloadAgent(manager, snapshot.id); - expect(manager.listProviderSubagents(snapshot.id)).toEqual([ + expect(await manager.listProviderSubagents(snapshot.id)).toEqual([ expect.objectContaining({ id: "stale-child", status: "canceled" }), ]); }); @@ -4924,11 +5024,13 @@ test("reloadAgentSession terminalizes running provider children when preserving provider: "codex", event: { type: "upsert", id: "running-child", title: "Running child", status: "running" }, }); - await vi.waitFor(() => expect(manager.listProviderSubagents(snapshot.id)).toHaveLength(1)); + await vi.waitFor(async () => + expect(await manager.listProviderSubagents(snapshot.id)).toHaveLength(1), + ); await manager.reloadAgentSession(snapshot.id); - expect(manager.listProviderSubagents(snapshot.id)).toEqual([ + expect(await manager.listProviderSubagents(snapshot.id)).toEqual([ expect.objectContaining({ id: "running-child", status: "canceled" }), ]); }); @@ -4938,6 +5040,7 @@ test("resuming without stored history restores provider children from the provid const storage = new AgentStorage(join(workdir, "agents"), logger); class ProviderChildHistorySession extends TestAgentSession { override async *streamHistory(): AsyncGenerator { + // The history's last report on this child is that it started. yield { type: "provider_subagent", provider: "codex", @@ -4945,7 +5048,7 @@ test("resuming without stored history restores provider children from the provid type: "upsert", id: "restored-child", title: "Restored child", - status: "completed", + status: "running", }, }; } @@ -4967,20 +5070,24 @@ test("resuming without stored history restores provider children from the provid const snapshot = await resumeWithProviderHistory(manager, workdir); - expect(manager.listProviderSubagents(snapshot.id)).toEqual([ + expect(await manager.listProviderSubagents(snapshot.id)).toEqual([ expect.objectContaining({ id: "restored-child", parentAgentId: snapshot.id, title: "Restored child", - status: "completed", - }), - ]); - expect(manager.listProviderSubagentActivity()).toEqual([ - expect.objectContaining({ - id: "restored-child", - parentAgentId: snapshot.id, + status: "running", }), ]); + // A history report is not the runtime's report: it does not keep the workspace or runtime busy. + expect(await manager.listProviderSubagentActivity()).toEqual([]); + expect(runtimeIsCollectable(manager, snapshot.id)).toBe(true); + + // Closing the runtime does not invent an outcome for a child that runtime never reported. + await manager.closeAgent(snapshot.id); + expect((await manager.getProviderSubagent(snapshot.id, "restored-child"))?.status).toBe( + "running", + ); + await storage.flush(); }); test("reloadAgentSession preserves current title when config title is unset", async () => { @@ -8352,16 +8459,16 @@ test("subscribe hides provider subagents of internal parents from global subscri }), }), ); - expect(() => manager.listProviderSubagents(internalAgentId)).toThrow( + await expect(manager.listProviderSubagents(internalAgentId)).rejects.toThrow( `Unknown agent '${internalAgentId}'`, ); - expect(() => manager.getProviderSubagent(internalAgentId, "hidden-child")).toThrow( + await expect(manager.getProviderSubagent(internalAgentId, "hidden-child")).rejects.toThrow( `Unknown agent '${internalAgentId}'`, ); await expect( manager.fetchProviderSubagentTimeline(internalAgentId, "hidden-child"), ).rejects.toThrow(`Unknown agent '${internalAgentId}'`); - expect(manager.listProviderSubagentActivity()).toEqual([]); + expect(await manager.listProviderSubagentActivity()).toEqual([]); }); test("subscribe emits state events for internal agents when subscribed by agentId", async () => { @@ -10680,10 +10787,10 @@ test("explicit close cancels running provider subagents before resume", async () logger, }); - expect(manager.getProviderSubagent(parent.id, "provider-child-running")?.status).toBe( + expect((await manager.getProviderSubagent(parent.id, "provider-child-running"))?.status).toBe( "canceled", ); - expect(manager.getProviderSubagent(parent.id, "provider-child-finishing")?.status).toBe( + expect((await manager.getProviderSubagent(parent.id, "provider-child-finishing"))?.status).toBe( "completed", ); } finally { @@ -10695,6 +10802,78 @@ test("explicit close cancels running provider subagents before resume", async () } }); +test("a child an earlier runtime reported running is live activity only once the provider reports it again", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-restarted-provider-child-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const backend = new DeferredTimelineBackend(); + class ResumeRecordingClient extends SessionRecordingAgentClient { + override async resumeSession( + handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + const session = new TestAgentSession({ provider: "codex", cwd: config?.cwd ?? workdir }); + session.describePersistence = () => handle; + this.sessions.push(session); + return session; + } + } + const firstClient = new ResumeRecordingClient(); + const first = new AgentManager({ + clients: { codex: firstClient }, + registry: storage, + logger, + timeline: new AgentTimeline(backend), + }); + const nextClient = new ResumeRecordingClient(); + const next = new AgentManager({ + clients: { codex: nextClient }, + registry: storage, + logger, + timeline: new AgentTimeline(backend), + }); + const childReport = (status: "running" | "completed"): AgentStreamEvent => ({ + type: "provider_subagent", + provider: "codex", + event: { type: "upsert", id: "provider-child", title: "Provider child", status }, + }); + + try { + const parent = await first.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + firstClient.sessions[0]!.pushEvent(childReport("running")); + await first.flush(); + expect(await first.listProviderSubagentActivity()).toEqual([ + expect.objectContaining({ id: "provider-child", status: "running" }), + ]); + await storage.flush(); + + // The first daemon stops without closing its agent, as a crash does; the next one resumes it. + await ensureAgentLoaded(parent.id, { agentManager: next, agentStorage: storage, logger }); + expect((await next.getProviderSubagent(parent.id, "provider-child"))?.status).toBe("running"); + expect(await next.listProviderSubagentActivity()).toEqual([]); + expect(runtimeIsCollectable(next, parent.id)).toBe(true); + + nextClient.sessions[0]!.pushEvent(childReport("running")); + await next.flush(); + expect(await next.listProviderSubagentActivity()).toEqual([ + expect.objectContaining({ id: "provider-child", parentAgentId: parent.id }), + ]); + expect(runtimeIsCollectable(next, parent.id)).toBe(false); + + nextClient.sessions[0]!.pushEvent(childReport("completed")); + await next.flush(); + expect(await next.listProviderSubagentActivity()).toEqual([]); + expect(runtimeIsCollectable(next, parent.id)).toBe(true); + } finally { + await Promise.all(next.listAgents().map((agent) => next.closeAgent(agent.id))).catch( + () => undefined, + ); + await storage.flush().catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("load waits for an in-flight explicit close and creates one resumed runtime", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-explicit-close-race-")); const storage = new AgentStorage(join(workdir, "agents"), logger); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 9b89c32908..bd1817733a 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -804,6 +804,12 @@ export class AgentManager { private readonly agents = new Map(); private readonly timeline: AgentTimeline; private readonly providerSubagents: ProviderSubagentStore; + /** + * Provider children each parent's current runtime reported running. A stored descriptor's status + * is only the last report, possibly from provider history or an earlier runtime, so workspace + * activity and idle cleanup read this instead. + */ + private readonly runningProviderSubagents = new Map>(); private readonly agentsAwaitingInitialSnapshotPersist = new Set(); private readonly sessionEventTails = new Map>(); private readonly steerEventBarriers = new Map(); @@ -1359,28 +1365,30 @@ export class AgentManager { return await this.timeline.fetch({ agentId: id }, options); } - listProviderSubagents(parentAgentId: string): ProviderSubagentDescriptor[] { + async listProviderSubagents(parentAgentId: string): Promise { this.requirePublicParent(parentAgentId); - return this.providerSubagents.list(parentAgentId); + return await this.providerSubagents.list(parentAgentId); } - listProviderSubagentActivity(): ProviderSubagentDescriptor[] { - const publicParentIds = new Set( - Array.from(this.agents.values()) - .filter((agent) => !agent.internal) - .map((agent) => agent.id), + /** The children each loaded public agent's current runtime reports running. */ + async listProviderSubagentActivity(): Promise { + const publicParents = Array.from(this.agents.values()).filter((agent) => !agent.internal); + const children = await Promise.all( + publicParents.map(async (agent) => { + const listed = await this.providerSubagents.list(agent.id); + const running = this.runningProviderSubagents.get(agent.id); + return listed.filter((child) => running?.has(child.id)); + }), ); - return this.providerSubagents - .listAll() - .filter((subagent) => publicParentIds.has(subagent.parentAgentId)); + return children.flat(); } - getProviderSubagent( + async getProviderSubagent( parentAgentId: string, subagentId: string, - ): ProviderSubagentDescriptor | null { + ): Promise { this.requirePublicParent(parentAgentId); - return this.providerSubagents.get(parentAgentId, subagentId); + return await this.providerSubagents.get(parentAgentId, subagentId); } async fetchProviderSubagentTimeline( @@ -1574,13 +1582,14 @@ export class AgentManager { currentResumeOptions, ); await this.requireExternalMcpSupport(session, providerLaunchConfig, launchContext); - return this.registerSession(session, storedConfig, resolvedAgentId, { + const registered = await this.registerSession(session, storedConfig, resolvedAgentId, { ...options, launchEnv, persistence: handle, restoring: true, importMissingHistory: true, }); + return await this.settleStoredRewind(registered); } importProviderSession(input: { @@ -1659,7 +1668,7 @@ export class AgentManager { }); for (const event of imported.providerSubagentEvents ?? []) { const update = await this.providerSubagents.apply(agent.id, event.provider, event.event); - if (update) this.dispatch({ type: "provider_subagent", event: update }); + if (update) this.publishProviderSubagent(update); } return agent; } finally { @@ -1772,7 +1781,7 @@ export class AgentManager { restoring: true, }); return settleRewind - ? registered.then((snapshot) => this.settleRewindAfterReload(snapshot)) + ? registered.then((snapshot) => this.settleRewindAfterResume(snapshot)) : registered.then((agent) => ({ agent, rewind: "none" as const })); } catch (error) { if (closedExisting) { @@ -1981,7 +1990,7 @@ export class AgentManager { (this.collectingRuntimes.has(agent.id) || !this.lifecycleMutationTails.has(agent.id)) && !this.foregroundMutationTails.has(agent.id) && !this.sessionEventTails.has(agent.id) && - !this.providerSubagents.list(agent.id).some((child) => child.status === "running") + !this.runningProviderSubagents.has(agent.id) ); } @@ -2050,17 +2059,21 @@ export class AgentManager { return true; } + /** + * Closing a runtime stops the children it reported running. A child it never reported keeps its + * last reported status: nothing this daemon observed says how it ended. + */ private async cancelRunningProviderSubagents(parentAgentId: string): Promise { - for (const subagent of this.providerSubagents.list(parentAgentId)) { - if (subagent.status !== "running") { - continue; - } + const running = this.runningProviderSubagents.get(parentAgentId); + this.runningProviderSubagents.delete(parentAgentId); + for (const subagent of await this.providerSubagents.list(parentAgentId)) { + if (!running?.has(subagent.id)) continue; const event = await this.providerSubagents.apply(parentAgentId, subagent.provider, { type: "upsert", id: subagent.id, status: "canceled", }); - if (event) this.dispatch({ type: "provider_subagent", event }); + if (event) this.publishProviderSubagent(event); } } @@ -3708,8 +3721,17 @@ export class AgentManager { return "open"; } - /** The reload itself succeeded; a rewind that still cannot settle stays open and is reported. */ - private async settleRewindAfterReload( + /** Stored history can hold a rewind an earlier runtime, or an earlier daemon, left open. */ + private async settleStoredRewind(agent: ManagedAgent): Promise { + if (!(await this.timeline.describe({ agentId: agent.id }))?.rewind) return agent; + return (await this.settleRewindAfterResume(agent)).agent; + } + + /** + * Judges an open rewind from the history of a session that just resumed. The resume itself + * succeeded; a rewind that still cannot settle stays open and is reported. + */ + private async settleRewindAfterResume( snapshot: ManagedAgent, ): Promise<{ agent: ManagedAgent; rewind: RewindSettlement }> { let rewind: RewindSettlement = "open"; @@ -4174,9 +4196,8 @@ export class AgentManager { events: readonly ProviderSubagentStreamEvent[], options: { dispatch: boolean }, ): Promise { - for (const event of events) { - const update = await this.providerSubagents.apply(agentId, event.provider, event.event); - if (update && options.dispatch) this.dispatch({ type: "provider_subagent", event: update }); + for (const update of await this.providerSubagents.applyHistory(agentId, events)) { + if (options.dispatch) this.dispatch({ type: "provider_subagent", event: update }); } } @@ -4279,12 +4300,10 @@ export class AgentManager { } private async discardRetainedAgentState(agentId: string): Promise { - await this.timeline.delete({ agentId }); + const removed = await this.providerSubagents.deleteParent(agentId); this.paseoToolPolicies.delete(agentId); this.detachPluginTools(agentId); - for (const event of await this.providerSubagents.deleteParent(agentId)) { - this.dispatch({ type: "provider_subagent", event }); - } + for (const event of removed) this.publishProviderSubagent(event); } private emitClosedAgent(agent: ManagedAgentClosed, options?: { persist?: boolean }): void { @@ -4391,7 +4410,11 @@ export class AgentManager { ): Promise { if (event.type === "provider_subagent") { const update = await this.providerSubagents.apply(agent.id, event.provider, event.event); - if (update) this.dispatch({ type: "provider_subagent", event: update }); + if (!update) return; + if (event.event.type === "upsert" && event.event.status === "running") { + this.markProviderSubagentRunning(agent.id, event.event.id); + } + this.publishProviderSubagent(update); return; } const turnId = getAgentStreamEventTurnId(event); @@ -5092,13 +5115,31 @@ export class AgentManager { }); } + /** Only the live session marks a child running; any report that ends or removes it unmarks it. */ + private publishProviderSubagent(event: ProviderSubagentStoreEvent): void { + if (event.type === "remove") { + this.unmarkProviderSubagentRunning(event.parentAgentId, event.subagentId); + } else if (event.type === "upsert" && event.subagent.status !== "running") { + this.unmarkProviderSubagentRunning(event.subagent.parentAgentId, event.subagent.id); + } + this.dispatch({ type: "provider_subagent", event }); + } + + private markProviderSubagentRunning(parentAgentId: string, subagentId: string): void { + const running = this.runningProviderSubagents.get(parentAgentId) ?? new Set(); + this.runningProviderSubagents.set(parentAgentId, running.add(subagentId)); + } + + private unmarkProviderSubagentRunning(parentAgentId: string, subagentId: string): void { + const running = this.runningProviderSubagents.get(parentAgentId); + if (!running?.delete(subagentId) || running.size > 0) return; + this.runningProviderSubagents.delete(parentAgentId); + } + /** Children spawned by retired tool calls leave with them, before clients learn of the retirement. */ private publishRetirement(agentId: string, commit: TimelineRetirementCommit): void { - const toolCallIds = new Set( - commit.retiredRows.flatMap((row) => (row.item.type === "tool_call" ? [row.item.callId] : [])), - ); - for (const event of this.providerSubagents.retireSpawnedBy(agentId, toolCallIds)) { - this.dispatch({ type: "provider_subagent", event }); + for (const subagentId of commit.retiredChildIds) { + this.publishProviderSubagent({ type: "remove", parentAgentId: agentId, subagentId }); } this.dispatch({ type: "timeline_retirement", diff --git a/packages/server/src/server/agent/plugin-provider.test.ts b/packages/server/src/server/agent/plugin-provider.test.ts index 68f96a69ee..9bb0a7f8a1 100644 --- a/packages/server/src/server/agent/plugin-provider.test.ts +++ b/packages/server/src/server/agent/plugin-provider.test.ts @@ -31,7 +31,7 @@ import { StaleProviderSessionError, } from "./stale-provider-session-error.js"; -test("long concurrent plugin and child streams retain text-sized history and release it on archive", () => { +test("long concurrent plugin and child streams retain text-sized history through archive and release it on delete", () => { const fixture = fileURLToPath(new URL("./test-utils/plugin-stream-memory.ts", import.meta.url)); const output = execFileSync( process.execPath, @@ -41,8 +41,12 @@ test("long concurrent plugin and child streams retain text-sized history and rel timeout: 20_000, }, ); - const memory = JSON.parse(output) as { retained: number; closed: number }; + const memory = JSON.parse(output) as { retained: number; archived: number; closed: number }; expect(memory.retained).toBeLessThan(24 * 1024 * 1024); + // Archiving closes the runtimes and keeps the stored history; it adds nothing. + expect(memory.archived).toBeLessThanOrEqual(memory.retained); + // Deleting the agents releases their histories and their children's. + expect(memory.closed).toBeLessThan(memory.archived); expect(memory.closed).toBeLessThan(6 * 1024 * 1024); }); @@ -642,7 +646,7 @@ describe("PluginAgentClientRegistry", () => { state: "completed", }); await nextTurn(); - const [child] = manager.listProviderSubagents(agent.id); + const [child] = await manager.listProviderSubagents(agent.id); expect(child).toMatchObject({ id: "reusable-child", title: "Plugin child", @@ -656,7 +660,7 @@ describe("PluginAgentClientRegistry", () => { state: "started", }); await nextTurn(); - expect(manager.listProviderSubagents(agent.id)).toMatchObject([ + expect(await manager.listProviderSubagents(agent.id)).toMatchObject([ { id: child.id, title: child.title, status: "running" }, ]); @@ -667,7 +671,7 @@ describe("PluginAgentClientRegistry", () => { state: outcome, }); await nextTurn(); - expect(manager.listProviderSubagents(agent.id)).toMatchObject([ + expect(await manager.listProviderSubagents(agent.id)).toMatchObject([ { id: child.id, title: child.title, status: outcome }, ]); } finally { diff --git a/packages/server/src/server/agent/projected-timeline-contract.test.ts b/packages/server/src/server/agent/projected-timeline-contract.test.ts index e300255f8a..b4dfa6cd11 100644 --- a/packages/server/src/server/agent/projected-timeline-contract.test.ts +++ b/packages/server/src/server/agent/projected-timeline-contract.test.ts @@ -60,7 +60,9 @@ test("projects Codex child history and confines old-client degradation to the ch ); } await expect - .poll(() => ctx.daemon.daemon.agentManager.listProviderSubagents(agent.id).length) + .poll( + async () => (await ctx.daemon.daemon.agentManager.listProviderSubagents(agent.id)).length, + ) .toBe(1); await expect .poll( 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 df47f2a37a..b5f25c28d8 100644 --- a/packages/server/src/server/agent/provider-subagents/store.test.ts +++ b/packages/server/src/server/agent/provider-subagents/store.test.ts @@ -1,7 +1,19 @@ import { describe, expect, test } from "vitest"; +import type { AgentTimelineItem } from "../agent-sdk-types.js"; import { AgentTimeline } from "../timeline/index.js"; import { ProviderSubagentStore } from "./store.js"; +function spawningCall(callId: string): AgentTimelineItem { + return { + type: "tool_call", + callId, + name: "Task", + status: "completed", + error: null, + detail: { type: "unknown", input: null, output: null }, + }; +} + describe("ProviderSubagentStore", () => { test("keeps provider children and their timelines scoped to the parent agent", async () => { const subagents = new ProviderSubagentStore(new AgentTimeline()); @@ -34,7 +46,7 @@ describe("ProviderSubagentStore", () => { timestamp: "2026-07-12T10:00:03.000Z", }); - expect(subagents.list("parent-a")).toEqual([ + expect(await subagents.list("parent-a")).toEqual([ expect.objectContaining({ id: "child-1", parentAgentId: "parent-a", @@ -53,12 +65,15 @@ describe("ProviderSubagentStore", () => { item: { type: "assistant_message", text: "Found it." }, }, ]); - expect(subagents.list("parent-b")[0]).toMatchObject({ provider: "claude", title: "Review" }); + expect((await subagents.list("parent-b"))[0]).toMatchObject({ + provider: "claude", + title: "Review", + }); expect(await subagents.deleteParent("parent-a")).toEqual([ { type: "remove", parentAgentId: "parent-a", subagentId: "child-1" }, ]); - expect(subagents.list("parent-a")).toEqual([]); - expect(subagents.list("parent-b")).toHaveLength(1); + expect(await subagents.list("parent-a")).toEqual([]); + expect(await subagents.list("parent-b")).toHaveLength(1); }); test("limits oversized provider child tool output before storage", async () => { @@ -86,7 +101,16 @@ describe("ProviderSubagentStore", () => { }); test("retires the children of rewound tool calls and drops their late events", async () => { - const subagents = new ProviderSubagentStore(new AgentTimeline()); + const timeline = new AgentTimeline(); + const subagents = new ProviderSubagentStore(timeline); + const parent = { agentId: "parent-a" }; + await timeline.create(parent, {}); + await timeline.append(parent, [ + { item: { type: "user_message", text: "A", messageId: "a" }, turnId: "t1" }, + { item: spawningCall("call-a"), turnId: "t1" }, + { item: { type: "user_message", text: "B", messageId: "b" }, turnId: "t2" }, + { item: spawningCall("call-b"), turnId: "t2" }, + ]); const upsert = (id: string, fields: { toolCallId?: string; parentSubagentId?: string }) => subagents.apply("parent-a", "claude", { type: "upsert", @@ -104,11 +128,13 @@ describe("ProviderSubagentStore", () => { item: { type: "assistant_message", text: "Working on B." }, }); - expect(subagents.retireSpawnedBy("parent-a", new Set(["call-b"]))).toEqual([ - { type: "remove", parentAgentId: "parent-a", subagentId: "rewound" }, - { type: "remove", parentAgentId: "parent-a", subagentId: "grandchild" }, - ]); - expect(subagents.list("parent-a").map((subagent) => subagent.id)).toEqual(["kept"]); + const intent = await timeline.beginRewind(parent, "b"); + await timeline.recordRewindCut(parent, intent.id, { kind: "at_target" }); + expect(await timeline.completeRewind(parent, intent.id)).toMatchObject({ + kind: "retired", + retiredChildIds: ["rewound", "grandchild"], + }); + expect((await subagents.list("parent-a")).map((subagent) => subagent.id)).toEqual(["kept"]); // The retired child's history stays stored; a late event neither revives nor extends it. expect( @@ -119,7 +145,10 @@ describe("ProviderSubagentStore", () => { }), ).toBeNull(); expect(await upsert("rewound", { toolCallId: "call-b" })).toBeNull(); - expect(subagents.list("parent-a").map((subagent) => subagent.id)).toEqual(["kept"]); + expect( + await subagents.apply("parent-a", "claude", { type: "remove", id: "rewound" }), + ).toBeNull(); + expect((await subagents.list("parent-a")).map((subagent) => subagent.id)).toEqual(["kept"]); expect( (await subagents.fetchTimeline("parent-a", "rewound")).rows.map((row) => row.item), ).toEqual([{ type: "assistant_message", text: "Working on B." }]); diff --git a/packages/server/src/server/agent/provider-subagents/store.ts b/packages/server/src/server/agent/provider-subagents/store.ts index 43393c3606..7cdaf116fe 100644 --- a/packages/server/src/server/agent/provider-subagents/store.ts +++ b/packages/server/src/server/agent/provider-subagents/store.ts @@ -61,10 +61,6 @@ export type ProviderSubagentStoreEvent = } | { type: "remove"; parentAgentId: string; subagentId: string }; -function storeKey(parentAgentId: string, subagentId: string): string { - return `${parentAgentId}\0${subagentId}`; -} - function childScope(parentAgentId: string, subagentId: string): TimelineScope { return { agentId: parentAgentId, subagentId }; } @@ -78,11 +74,11 @@ function stickyField(next: T | undefined, previous: T | null | undefined): T return next === undefined ? (previous ?? null) : next; } +/** + * Applies provider child events to the parent's timeline, which stores each child's descriptor, + * history and retirement next to the parent's own history. + */ export class ProviderSubagentStore { - private readonly descriptors = new Map(); - /** Children of rewound turns. Their history stays stored; late provider events are dropped. */ - private readonly retired = new Set(); - constructor(private readonly timeline: AgentTimeline) {} /** Resolves null when the event belongs to a retired child; nothing is recorded. */ @@ -91,20 +87,18 @@ export class ProviderSubagentStore { provider: AgentProvider, event: ProviderSubagentInputEvent, ): Promise { - const key = storeKey(parentAgentId, event.id); const scope = childScope(parentAgentId, event.id); - if (this.retired.has(key)) return null; if (event.type === "remove") { - this.descriptors.delete(key); - await this.timeline.delete(scope); + if (!(await this.timeline.removeChild(scope))) return null; return { type: "remove", parentAgentId, subagentId: event.id }; } if (event.type === "timeline") { - await this.timeline.open(scope); - const { - commits: [{ row, epoch }], - } = await this.timeline.append(scope, [{ item: event.item, timestamp: event.timestamp }]); + const commits = await this.timeline.recordChildEntries(scope, [ + { item: event.item, timestamp: event.timestamp }, + ]); + if (!commits) return null; + const [{ row, epoch }] = commits; return { type: "timeline", parentAgentId, @@ -115,39 +109,55 @@ export class ProviderSubagentStore { }; } - const previous = this.descriptors.get(key); - const timestamp = event.timestamp ?? new Date().toISOString(); - const subagent: ProviderSubagentDescriptor = { - id: event.id, - parentAgentId, - provider, - title: stickyField(event.title, previous?.title), - description: stickyField(event.description, previous?.description), - status: event.status ?? previous?.status ?? "running", - createdAt: previous?.createdAt ?? timestamp, - updatedAt: timestamp, - toolCallId: stickyField(event.toolCallId, previous?.toolCallId), - cwd: stickyField(event.cwd, previous?.cwd), - subtitle: stickyField(event.subtitle, previous?.subtitle), - parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId), - }; - this.descriptors.set(key, subagent); - await this.timeline.open(scope); - return { type: "upsert", subagent }; + const subagent = await this.timeline.recordChild(scope, (previous) => { + const timestamp = event.timestamp ?? new Date().toISOString(); + return { + id: event.id, + parentAgentId, + provider, + title: stickyField(event.title, previous?.title), + description: stickyField(event.description, previous?.description), + status: event.status ?? previous?.status ?? "running", + createdAt: previous?.createdAt ?? timestamp, + updatedAt: timestamp, + toolCallId: stickyField(event.toolCallId, previous?.toolCallId), + cwd: stickyField(event.cwd, previous?.cwd), + subtitle: stickyField(event.subtitle, previous?.subtitle), + parentSubagentId: stickyField(event.parentSubagentId, previous?.parentSubagentId), + }; + }); + return subagent ? { type: "upsert", subagent } : null; } - list(parentAgentId: string): ProviderSubagentDescriptor[] { - return [...this.descriptors.values()] - .filter((subagent) => subagent.parentAgentId === parentAgentId) + /** + * Applies the child events of a provider history read. Children the parent already holds, + * listed or retired, are skipped: like the parent's, their owned history is never changed. + */ + async applyHistory( + parentAgentId: string, + events: readonly { provider: AgentProvider; event: ProviderSubagentInputEvent }[], + ): Promise { + const held = new Set( + (await this.timeline.listChildren(parentAgentId)).map((child) => child.subagentId), + ); + const updates: ProviderSubagentStoreEvent[] = []; + for (const { provider, event } of events) { + if (held.has(event.id)) continue; + const update = await this.apply(parentAgentId, provider, event); + if (update) updates.push(update); + } + return updates; + } + + /** The parent's listed children, oldest first. Retired children are not listed. */ + async list(parentAgentId: string): Promise { + return (await this.timeline.listChildren(parentAgentId)) + .flatMap((child) => (!child.retired && child.descriptor ? [child.descriptor] : [])) .sort((left, right) => left.createdAt.localeCompare(right.createdAt)); } - listAll(): ProviderSubagentDescriptor[] { - return [...this.descriptors.values()]; - } - - get(parentAgentId: string, subagentId: string): ProviderSubagentDescriptor | null { - return this.descriptors.get(storeKey(parentAgentId, subagentId)) ?? null; + async get(parentAgentId: string, subagentId: string): Promise { + return (await this.list(parentAgentId)).find((child) => child.id === subagentId) ?? null; } fetchTimeline( @@ -158,53 +168,10 @@ export class ProviderSubagentStore { return this.timeline.fetch(childScope(parentAgentId, subagentId), options); } - /** - * Retires the children spawned by retired tool calls of the parent, and their descendants. They - * leave the listing; their history stays stored. - */ - retireSpawnedBy( - parentAgentId: string, - toolCallIds: ReadonlySet, - ): Extract[] { - const children = this.list(parentAgentId); - const retiredIds = new Set( - children - .filter((subagent) => subagent.toolCallId !== null && toolCallIds.has(subagent.toolCallId)) - .map((subagent) => subagent.id), - ); - let grew = retiredIds.size > 0; - while (grew) { - grew = false; - for (const subagent of children) { - if (retiredIds.has(subagent.id) || subagent.parentSubagentId === null) continue; - if (!retiredIds.has(subagent.parentSubagentId)) continue; - retiredIds.add(subagent.id); - grew = true; - } - } - return [...retiredIds].map((subagentId) => { - const key = storeKey(parentAgentId, subagentId); - this.descriptors.delete(key); - this.retired.add(key); - return { type: "remove", parentAgentId, subagentId }; - }); - } - + /** Deletes the parent's history and every child's, retired children included. */ async deleteParent(parentAgentId: string): Promise { - const prefix = storeKey(parentAgentId, ""); - const retired = [...this.retired] - .filter((key) => key.startsWith(prefix)) - .map((key) => key.slice(prefix.length)); - for (const subagentId of retired) this.retired.delete(storeKey(parentAgentId, subagentId)); - const removed = this.list(parentAgentId).map((subagent) => subagent.id); - for (const subagentId of removed) { - this.descriptors.delete(storeKey(parentAgentId, subagentId)); - } - await Promise.all( - [...removed, ...retired].map((subagentId) => - this.timeline.delete(childScope(parentAgentId, subagentId)), - ), - ); - return removed.map((subagentId) => ({ type: "remove", parentAgentId, subagentId })); + const listed = await this.list(parentAgentId); + await this.timeline.delete({ agentId: parentAgentId }); + return listed.map((child) => ({ type: "remove", parentAgentId, subagentId: child.id })); } } diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index c7f45cda7f..22941ed2fc 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -2155,7 +2155,7 @@ describe("Codex app-server provider", () => { ); expect(new Set(toolCalls.map((item) => item.callId))).toEqual(new Set(["spawn-child"])); expect(toolCalls.at(-1)).toMatchObject({ callId: "spawn-child", status: "completed" }); - expect(codex.manager.getProviderSubagent(codex.agentId, "child-thread")).toMatchObject({ + expect(await codex.manager.getProviderSubagent(codex.agentId, "child-thread")).toMatchObject({ toolCallId: "spawn-child", status: "completed", }); diff --git a/packages/server/src/server/agent/rewind/rewind.test.ts b/packages/server/src/server/agent/rewind/rewind.test.ts index f66bd41c53..8e31a2c40b 100644 --- a/packages/server/src/server/agent/rewind/rewind.test.ts +++ b/packages/server/src/server/agent/rewind/rewind.test.ts @@ -344,6 +344,115 @@ describe("AgentManager conversation rewind", () => { expect(await harness.activeTexts()).toEqual(["A", "reply to A", "C", "reply to C"]); }); + test("a restarted daemon whose stored history holds an unknown outcome settles it on resume", async () => { + const registry = await createFlakyStorage(); + const harness = await createConversationHarness({ registry }); + const { manager, agentId, session, client, backend } = harness; + await harness.prompt("A"); + await harness.prompt("B"); + session.nextRewind = { kind: "unknown", applied: true }; + await expect(manager.rewind(agentId, "B", "conversation")).rejects.toThrow(); + await manager.closeAgent(agentId); + await manager.flush(); + + // The next daemon reads the same stored history; only construction differs. + const restarted = new AgentManager({ + clients: { claude: client }, + logger: createTestLogger(), + timeline: new AgentTimeline(backend), + registry, + }); + await ensureAgentLoaded(agentId, { + agentManager: restarted, + agentStorage: registry, + logger: createTestLogger(), + }); + for await (const _event of restarted.streamAgent(agentId, "C", { clientMessageId: "C" })) { + // Drain the turn. + } + await restarted.flush(); + + expect(session.recordedRewinds).toHaveLength(1); + const page = await restarted.fetchTimeline(agentId, { limit: 0 }); + expect(page.rows.map((row) => ("text" in row.item ? row.item.text : row.item.type))).toEqual([ + "A", + "reply to A", + "C", + "reply to C", + ]); + expect(page.retirementCount).toBe(1); + await restarted.closeAgent(agentId); + await restarted.flush(); + await registry.flush(); + }); + + test("a restarted daemon keeps the provider children it listed and the ones a rewind retired", async () => { + const registry = await createFlakyStorage(); + const harness = await createConversationHarness({ registry }); + const { manager, agentId, session, client, backend } = harness; + session.subagentPrompts.add("A").add("B"); + await harness.prompt("A"); + await harness.prompt("B"); + await manager.rewind(agentId, "B", "conversation"); + await manager.closeAgent(agentId); + await manager.flush(); + + // The next daemon reads the same stored history; only construction differs. + const restarted = new AgentManager({ + clients: { claude: client }, + logger: createTestLogger(), + timeline: new AgentTimeline(backend), + registry, + }); + await ensureAgentLoaded(agentId, { + agentManager: restarted, + agentStorage: registry, + logger: createTestLogger(), + }); + const listed = async () => + (await restarted.listProviderSubagents(agentId)).map((child) => child.id); + const childTexts = async (subagentId: string) => + (await restarted.fetchProviderSubagentTimeline(agentId, subagentId)).rows.map((row) => + "text" in row.item ? row.item.text : row.item.type, + ); + + expect(await listed()).toEqual(["child-of-A"]); + expect(await childTexts("child-of-A")).toEqual(["working on A"]); + expect(await childTexts("child-of-B")).toEqual(["working on B"]); + + // The provider's history still holds the rewound child; importing it must not bring it back. + session.history.push({ type: "assistant_message", text: "continued outside Paseo" }); + session.historySubagentEvents = [ + { + type: "provider_subagent", + provider: "claude", + event: { type: "upsert", id: "child-of-B", toolCallId: "call-2", status: "completed" }, + }, + { + type: "provider_subagent", + provider: "claude", + event: { + type: "timeline", + id: "child-of-B", + item: { type: "assistant_message", text: "late output" }, + }, + }, + ]; + expect(await reloadAgent(restarted, agentId)).toEqual({ kind: "appended" }); + + expect(await listed()).toEqual(["child-of-A"]); + expect(await childTexts("child-of-B")).toEqual(["working on B"]); + + await restarted.closeAgent(agentId); + await restarted.deleteAgentState(agentId); + await restarted.flush(); + const stored = new AgentTimeline(backend); + expect(await stored.describe({ agentId })).toBeNull(); + expect(await stored.describe({ agentId, subagentId: "child-of-A" })).toBeNull(); + expect(await stored.describe({ agentId, subagentId: "child-of-B" })).toBeNull(); + await registry.flush(); + }); + test("reloading settles an unknown outcome the provider history proves unchanged", async () => { const harness = await createConversationHarness(); const { manager, agentId, session } = harness; @@ -544,14 +653,16 @@ describe("AgentManager conversation rewind", () => { await harness.prompt("A"); await harness.prompt("B"); await manager.flush(); - expect(manager.listProviderSubagents(agentId).map((child) => child.id)).toEqual([ + expect((await manager.listProviderSubagents(agentId)).map((child) => child.id)).toEqual([ "child-of-A", "child-of-B", ]); await manager.rewind(agentId, "B", "conversation"); - expect(manager.listProviderSubagents(agentId).map((child) => child.id)).toEqual(["child-of-A"]); + expect((await manager.listProviderSubagents(agentId)).map((child) => child.id)).toEqual([ + "child-of-A", + ]); expect( harness.events.flatMap((event) => event.type === "provider_subagent" && event.event.type === "remove" ? [event.event] : [], diff --git a/packages/server/src/server/agent/rewind/test-rewind-session.ts b/packages/server/src/server/agent/rewind/test-rewind-session.ts index fb06fbecf8..a40dfa5960 100644 --- a/packages/server/src/server/agent/rewind/test-rewind-session.ts +++ b/packages/server/src/server/agent/rewind/test-rewind-session.ts @@ -58,6 +58,8 @@ export class FakeRewindSession implements AgentSession { nextRewind: FakeRewindOutcome = { kind: "applied" }; /** Prompts whose turn spawns a provider subagent through a tool call. */ readonly subagentPrompts = new Set(); + /** Provider child events the provider's history replays after its conversation. */ + historySubagentEvents: AgentStreamEvent[] = []; private promptCount = 0; private subscribers = new Set<(event: AgentStreamEvent) => void>(); @@ -118,6 +120,15 @@ export class FakeRewindSession implements AgentSession { provider: this.provider, event: { type: "upsert", id: `child-of-${text}`, toolCallId: callId, status: "running" }, }); + this.emit({ + type: "provider_subagent", + provider: this.provider, + event: { + type: "timeline", + id: `child-of-${text}`, + item: { type: "assistant_message", text: `working on ${text}` }, + }, + }); } this.activeTurnId = null; this.emit({ type: "turn_completed", provider: this.provider, turnId }); @@ -136,6 +147,7 @@ export class FakeRewindSession implements AgentSession { for (const item of this.history) { yield { type: "timeline", provider: this.provider, item }; } + yield* this.historySubagentEvents; } async getRuntimeInfo() { diff --git a/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts b/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts index 0807b1ec45..93950bb075 100644 --- a/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts +++ b/packages/server/src/server/agent/test-utils/plugin-stream-memory.ts @@ -149,7 +149,7 @@ for (let index = 0; index < sessions.length; index++) { const item = items[step * 2]!; assert.equal(item.type === "reasoning" && item.text, expected.get(`${index}:${step}`)); } - const children = manager.listProviderSubagents(agents[index]!.id); + const children = await manager.listProviderSubagents(agents[index]!.id); assert.equal(children.length, 1); let rootItems = 0; let childItems = 0; @@ -160,17 +160,32 @@ for (let index = 0; index < sessions.length; index++) { assert.equal(rootItems, steps * 2); assert.equal(childItems, steps * 2); await manager.archiveAgent(agents[index]!.id); - assert.equal(await timeline.describe({ agentId: agents[index]!.id }), null); assert.equal( manager.listAgents().some((agent) => agent.id === agents[index]!.id), false, ); + // An archived chat reads its history and provider children without its provider. + assert.equal((await timeline.readRows({ agentId: agents[index]!.id })).length, steps * 2); + assert.equal((await timeline.listChildren(agents[index]!.id)).length, 1); +} +global.gc(); +const archived = process.memoryUsage().heapUsed - before; +for (const agent of agents) { + const stored = await timeline.listChildren(agent.id); + await manager.deleteAgentState(agent.id); + assert.equal(await timeline.describe({ agentId: agent.id }), null); + for (const child of stored) { + assert.equal( + await timeline.describe({ agentId: agent.id, subagentId: child.subagentId }), + null, + ); + } } await plugins.shutdown(); await manager.flushForShutdown(); await rm(root, { recursive: true, force: true }); global.gc(); const closed = process.memoryUsage().heapUsed - before; -// Keep closed session references alive, as callers can retain them after archive. +// Keep closed session references alive, as callers can retain them after deletion. assert.equal(sessions.length, 3); -process.stdout.write(JSON.stringify({ retained, closed })); +process.stdout.write(JSON.stringify({ retained, archived, closed })); diff --git a/packages/server/src/server/agent/timeline/agent-timeline.test.ts b/packages/server/src/server/agent/timeline/agent-timeline.test.ts index 777528761b..a846e22731 100644 --- a/packages/server/src/server/agent/timeline/agent-timeline.test.ts +++ b/packages/server/src/server/agent/timeline/agent-timeline.test.ts @@ -3,9 +3,10 @@ import { promisify } from "node:util"; import { fileURLToPath } from "node:url"; import { describe, expect, it } from "vitest"; import type { AgentTimelineItem, ConversationRewindCut } from "../agent-sdk-types.js"; +import type { ProviderSubagentDescriptor } from "../provider-subagents/store.js"; import { limitAgentTimelineItemContent } from "../agent-timeline-content.js"; import type { AgentTimelineFetchResult, AgentTimelineRow } from "../agent-timeline-store-types.js"; -import { DeferredTimelineBackend } from "./testing.js"; +import { DeferredTimelineBackend, TIMELINE_CONTRACT_BACKENDS } from "./testing.js"; import { selectReferenceProjectedTimelinePage } from "../test-utils/timeline-page-reference.js"; import { AgentTimeline, @@ -33,10 +34,16 @@ async function append( return commit.row; } -describe.each([ - ["memory", () => new AgentTimeline()], - ["deferred", () => new AgentTimeline(new DeferredTimelineBackend())], -])("AgentTimeline over a %s backend", (_name, createTimeline) => { +describe.each( + TIMELINE_CONTRACT_BACKENDS.flatMap(({ name, create }) => [ + { name, create, createTimeline: () => new AgentTimeline(create()) }, + { + name: `deferred ${name}`, + create, + createTimeline: () => new AgentTimeline(new DeferredTimelineBackend(create())), + }, + ]), +)("AgentTimeline over a $name backend", ({ create, createTimeline }) => { it("lets a read observe writes called before it without awaiting them", async () => { const timeline = createTimeline(); void timeline.create(root("a"), { epoch: "e" }); @@ -77,7 +84,7 @@ describe.each([ }); it("rejects a failed write without publishing it or consuming a sequence number", async () => { - const backend = new DeferredTimelineBackend(); + const backend = new DeferredTimelineBackend(create()); const timeline = new AgentTimeline(backend); await timeline.create(root("a"), { epoch: "e" }); const published: TimelineCommit[] = []; @@ -230,9 +237,9 @@ describe.each([ }); }); -describe("history paging", () => { +describe.each(TIMELINE_CONTRACT_BACKENDS)("history paging over a $name backend", ({ create }) => { it("resets stale forward cursors to complete available turns", async () => { - const timeline = new AgentTimeline(); + const timeline = new AgentTimeline(create()); await timeline.create(root("agent"), { epoch: "current" }); for (let index = 0; index < 60; index++) await append( @@ -284,7 +291,7 @@ describe("history paging", () => { ]; it("clamps an overshooting before cursor into the bounded tail window", async () => { - const timeline = new AgentTimeline(); + const timeline = new AgentTimeline(create()); await timeline.create(root("agent-1"), { epoch: "epoch-1", nextSeq: 8, rows: seededRows }); const result = await timeline.fetch(root("agent-1"), { @@ -318,7 +325,7 @@ describe("history paging", () => { }); it("returns a bounded reset window when an after cursor is behind retained history", async () => { - const timeline = new AgentTimeline(); + const timeline = new AgentTimeline(create()); await timeline.create(root("agent-1"), { epoch: "epoch-1", nextSeq: 8, rows: seededRows }); const result = await timeline.fetch(root("agent-1"), { @@ -408,64 +415,67 @@ describe("projected timeline retention", () => { }); }); -describe("projected sequence ownership", () => { - it("preserves fetched coverage when another source chunk arrives", async () => { - const timeline = new AgentTimeline(); - await timeline.create(root("a")); - await append(timeline, "a", { type: "assistant_message", text: "A" }); - const before = await timeline.fetch(root("a")); - await append(timeline, "a", { type: "assistant_message", text: "B" }); - expect(before.rows[0]).toMatchObject({ - item: { text: "A" }, - seqEnd: 1, - sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], +describe.each(TIMELINE_CONTRACT_BACKENDS)( + "projected sequence ownership over a $name backend", + ({ create }) => { + it("preserves fetched coverage when another source chunk arrives", async () => { + const timeline = new AgentTimeline(create()); + await timeline.create(root("a")); + await append(timeline, "a", { type: "assistant_message", text: "A" }); + const before = await timeline.fetch(root("a")); + await append(timeline, "a", { type: "assistant_message", text: "B" }); + expect(before.rows[0]).toMatchObject({ + item: { text: "A" }, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + }); + expect((await timeline.fetch(root("a"))).rows[0]).toMatchObject({ + item: { text: "AB" }, + seqEnd: 2, + sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }], + }); }); - expect((await timeline.fetch(root("a"))).rows[0]).toMatchObject({ - item: { text: "AB" }, - seqEnd: 2, - sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }], - }); - }); - it("preserves source positions when seeding a projected history whose last update is an earlier tool", async () => { - const timeline = new AgentTimeline(); - await timeline.create(root("a")); - const tool = { - type: "tool_call" as const, - callId: "t", - name: "shell", - error: null, - detail: { type: "plain_text" as const, label: "work" }, - }; - await append(timeline, "a", { ...tool, status: "running" }); - await append(timeline, "a", { type: "assistant_message", text: "Answer" }); - await append(timeline, "a", { ...tool, status: "completed" }); - await timeline.create(root("b"), { rows: await timeline.readRows(root("a")) }); - expect(await timeline.lastItem(root("b"))).toMatchObject({ - type: "tool_call", - status: "completed", + it("preserves source positions when seeding a projected history whose last update is an earlier tool", async () => { + const timeline = new AgentTimeline(create()); + await timeline.create(root("a")); + const tool = { + type: "tool_call" as const, + callId: "t", + name: "shell", + error: null, + detail: { type: "plain_text" as const, label: "work" }, + }; + await append(timeline, "a", { ...tool, status: "running" }); + await append(timeline, "a", { type: "assistant_message", text: "Answer" }); + await append(timeline, "a", { ...tool, status: "completed" }); + await timeline.create(root("b"), { rows: await timeline.readRows(root("a")) }); + expect(await timeline.lastItem(root("b"))).toMatchObject({ + type: "tool_call", + status: "completed", + }); + expect((await timeline.readRows(root("b")))[1]?.item).toEqual({ + type: "assistant_message", + text: "Answer", + }); + expect((await append(timeline, "b", { type: "user_message", text: "next" })).seq).toBe(4); + expect(await timeline.lastItem(root("b"))).toEqual({ type: "user_message", text: "next" }); + expect((await timeline.fetch(root("b"))).rows.map((row) => row.seqStart)).toEqual([1, 2, 4]); }); - expect((await timeline.readRows(root("b")))[1]?.item).toEqual({ - type: "assistant_message", - text: "Answer", - }); - expect((await append(timeline, "b", { type: "user_message", text: "next" })).seq).toBe(4); - expect(await timeline.lastItem(root("b"))).toEqual({ type: "user_message", text: "next" }); - expect((await timeline.fetch(root("b"))).rows.map((row) => row.seqStart)).toEqual([1, 2, 4]); - }); - it("keeps differently identified assistant messages as separate rows", async () => { - const timeline = new AgentTimeline(); - await timeline.create(root("a")); - await append(timeline, "a", { type: "assistant_message", messageId: "first", text: "First" }); - await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "Sec" }); - await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "ond" }); - expect((await timeline.readRows(root("a"))).map((row) => row.item)).toEqual([ - { type: "assistant_message", messageId: "first", text: "First" }, - { type: "assistant_message", messageId: "second", text: "Second" }, - ]); - }); -}); + it("keeps differently identified assistant messages as separate rows", async () => { + const timeline = new AgentTimeline(create()); + await timeline.create(root("a")); + await append(timeline, "a", { type: "assistant_message", messageId: "first", text: "First" }); + await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "Sec" }); + await append(timeline, "a", { type: "assistant_message", messageId: "second", text: "ond" }); + expect((await timeline.readRows(root("a"))).map((row) => row.item)).toEqual([ + { type: "assistant_message", messageId: "first", text: "First" }, + { type: "assistant_message", messageId: "second", text: "Second" }, + ]); + }); + }, +); it("includes transitive tool updates in a contiguous projected tail", async () => { const timeline = new AgentTimeline(); @@ -673,369 +683,478 @@ it("continues a scroll across appends and lifecycle collapse without losing sour expect(last.hasNewer).toBe(false); }); -describe("conversation rewind", () => { - const scope = root("agent"); - const createBackend = () => new DeferredTimelineBackend(); +describe.each(TIMELINE_CONTRACT_BACKENDS)( + "conversation rewind over a $name backend", + ({ create }) => { + const scope = root("agent"); + const createBackend = () => new DeferredTimelineBackend(create()); - function prompt(messageId: string, turnId: string): TimelineEntry { - return { item: { type: "user_message", text: messageId, messageId }, turnId }; - } + function prompt(messageId: string, turnId: string): TimelineEntry { + return { item: { type: "user_message", text: messageId, messageId }, turnId }; + } - function reply(text: string, turnId: string): TimelineEntry { - return { item: { type: "assistant_message", text, messageId: `reply-${text}` }, turnId }; - } + function reply(text: string, turnId: string): TimelineEntry { + return { item: { type: "assistant_message", text, messageId: `reply-${text}` }, turnId }; + } - function tool(callId: string, turnId: string, status: "running" | "completed"): TimelineEntry { - return { - item: { - type: "tool_call", - callId, - name: "shell", - status, - error: null, - detail: { type: "unknown", input: null, output: null }, - }, - turnId, - }; - } + function tool(callId: string, turnId: string, status: "running" | "completed"): TimelineEntry { + return { + item: { + type: "tool_call", + callId, + name: "shell", + status, + error: null, + detail: { type: "unknown", input: null, output: null }, + }, + turnId, + }; + } - async function conversation(): Promise<{ - backend: DeferredTimelineBackend; - timeline: AgentTimeline; - }> { - const backend = createBackend(); - const timeline = new AgentTimeline(backend); - await timeline.create(scope, { epoch: "e" }); - await timeline.append(scope, [ - prompt("a", "t1"), - reply("A", "t1"), - prompt("b", "t2"), - tool("call-b", "t2", "running"), - reply("B", "t2"), - ]); - return { backend, timeline }; - } + async function conversation(): Promise<{ + backend: DeferredTimelineBackend; + timeline: AgentTimeline; + }> { + const backend = createBackend(); + const timeline = new AgentTimeline(backend); + await timeline.create(scope, { epoch: "e" }); + await timeline.append(scope, [ + prompt("a", "t1"), + reply("A", "t1"), + prompt("b", "t2"), + tool("call-b", "t2", "running"), + reply("B", "t2"), + ]); + return { backend, timeline }; + } - async function rewind( - timeline: AgentTimeline, - messageId: string, - cut: ConversationRewindCut = { kind: "at_target" }, - ): Promise { - const intent = await timeline.beginRewind(scope, messageId); - await timeline.recordRewindCut(scope, intent.id, cut); - return timeline.completeRewind(scope, intent.id); - } + async function rewind( + timeline: AgentTimeline, + messageId: string, + cut: ConversationRewindCut = { kind: "at_target" }, + ): Promise { + const intent = await timeline.beginRewind(scope, messageId); + await timeline.recordRewindCut(scope, intent.id, cut); + return timeline.completeRewind(scope, intent.id); + } - function texts(rows: readonly { item: AgentTimelineItem }[]): string[] { - return rows.map((row) => ("text" in row.item ? row.item.text : row.item.type)); - } + function texts(rows: readonly { item: AgentTimelineItem }[]): string[] { + return rows.map((row) => ("text" in row.item ? row.item.text : row.item.type)); + } - it("retires the rewound turn from active history and keeps it stored", async () => { - const { backend, timeline } = await conversation(); - const published: TimelineRetirementCommit[] = []; - const intent = await timeline.beginRewind(scope, "b"); - await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); - const completion = await timeline.completeRewind(scope, intent.id, { - publish: (commit) => published.push(commit), - }); - await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - - expect(completion).toMatchObject({ - kind: "retired", - retirement: { seq: 6, retireFromSeq: 3 }, - epoch: "e", - retirementCount: 1, - }); - expect(published.map((commit) => texts(commit.retiredRows))).toEqual([["b", "tool_call", "B"]]); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "c", "C"]); - expect(texts(backend.retiredRows(scope))).toEqual(["b", "tool_call", "B"]); - expect(await timeline.describe(scope)).toEqual({ - epoch: "e", - size: 4, - retirementCount: 1, - rewind: null, - }); - const tail = await timeline.fetch(scope); - expect(texts(tail.rows)).toEqual(["a", "A", "c", "C"]); - expect(tail).toMatchObject({ - epoch: "e", - startSeq: 1, - endSeq: 8, - retirements: [{ seq: 6, retireFromSeq: 3 }], - }); - }); - - it("advances a caught-up cursor through the retirement to the next turn", async () => { - const { timeline } = await conversation(); - await rewind(timeline, "b"); - await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - - const catchUp = await timeline.fetch(scope, { - direction: "after", - cursor: { epoch: "e", seq: 5 }, - }); - expect(texts(catchUp.rows)).toEqual(["c", "C"]); - expect(catchUp).toMatchObject({ - reset: false, - startSeq: 6, - endSeq: 8, - hasNewer: false, - retirements: [{ seq: 6, retireFromSeq: 3 }], - }); - - const retirementOnly = await timeline.fetch(scope, { - direction: "after", - cursor: { epoch: "e", seq: 5 }, - limit: 1, - }); - expect(retirementOnly).toMatchObject({ - rows: [], - startSeq: 6, - endSeq: 6, - hasNewer: true, - retirements: [{ seq: 6, retireFromSeq: 3 }], - }); - const next = await timeline.fetch(scope, { - direction: "after", - cursor: { epoch: "e", seq: 6 }, - }); - expect(texts(next.rows)).toEqual(["c", "C"]); - expect(next.retirements).toEqual([]); - }); - - it("pages a tail whose last turn was rewound with nothing after it", async () => { - const { timeline } = await conversation(); - await rewind(timeline, "b"); - - const tail = await timeline.fetch(scope); - expect(texts(tail.rows)).toEqual(["a", "A"]); - expect(tail).toMatchObject({ endSeq: 6, retirements: [{ seq: 6, retireFromSeq: 3 }] }); - const caughtUp = await timeline.fetch(scope, { - direction: "after", - cursor: { epoch: "e", seq: 2 }, - }); - expect(caughtUp).toMatchObject({ - rows: [], - endSeq: 6, - hasNewer: false, - retirements: [{ seq: 6, retireFromSeq: 3 }], - }); - }); - - it("cannot retire a later turn by completing the same rewind again", async () => { - const { timeline } = await conversation(); - const intent = await timeline.beginRewind(scope, "b"); - await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); - await timeline.completeRewind(scope, intent.id); - await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - - expect(await timeline.completeRewind(scope, intent.id)).toEqual({ kind: "absent" }); - expect(await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" })).toBeNull(); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "c", "C"]); - expect((await timeline.describe(scope))?.retirementCount).toBe(1); - }); - - it("drops late output of a retired turn and still accepts updates to active items", async () => { - const timeline = new AgentTimeline(createBackend()); - await timeline.create(scope, { epoch: "e" }); - await timeline.append(scope, [ - prompt("a", "t1"), - tool("call-a", "t1", "running"), - prompt("b", "t2"), - tool("call-b", "t2", "running"), - ]); - await rewind(timeline, "b"); - - const late = await timeline.ingest(scope, tool("call-b", "t2", "completed")); - const lateText = await timeline.ingest(scope, reply("B again", "t2")); - const reusedCallId = await timeline.ingest(scope, tool("call-a", "t2", "completed")); - const active = await timeline.ingest(scope, tool("call-a", "t1", "completed")); - - expect([late, lateText, reusedCallId].map((result) => result.kind)).toEqual([ - "dropped", - "dropped", - "dropped", - ]); - expect(active.kind).toBe("recorded"); - const rows = await timeline.readRows(scope); - expect(rows.map((row) => row.item)).toEqual([ - { type: "user_message", text: "a", messageId: "a" }, - expect.objectContaining({ type: "tool_call", callId: "call-a", status: "completed" }), - ]); - }); - - it("allows one open rewind at a time and leaves history alone when abandoned", async () => { - const { timeline } = await conversation(); - const intent = await timeline.beginRewind(scope, "b"); - - await expect(timeline.beginRewind(scope, "a")).rejects.toBeInstanceOf( - TimelineRewindPendingError, - ); - await expect(timeline.completeRewind(scope, intent.id)).rejects.toThrow( - "Cannot complete a rewind before its provider outcome", - ); - await timeline.abandonRewind(scope, intent.id); - expect((await timeline.describe(scope))?.rewind).toBeNull(); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); - await expect(timeline.beginRewind(scope, "missing")).rejects.toBeInstanceOf( - TimelineRewindTargetError, - ); - expect((await timeline.beginRewind(scope, "a")).targetSeq).toBe(1); - }); - - it("retires from the boundary the provider actually kept", async () => { - const { timeline } = await conversation(); - await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - - const completion = await rewind(timeline, "c", { kind: "after", retainedMessageId: "a" }); - - expect(completion).toMatchObject({ kind: "retired", retirement: { retireFromSeq: 3 } }); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A"]); - }); - - it("matches the retained boundary by the provider's message id", async () => { - const backend = createBackend(); - const timeline = new AgentTimeline(backend); - await timeline.create(scope, { epoch: "e" }); - await timeline.ingest(scope, { - item: { type: "user_message", text: "a", messageId: "a", clientMessageId: "client-a" }, - }); - await timeline.attachProviderMessageId(scope, "client-a", "provider-a"); - await timeline.append(scope, [reply("A", "t1"), prompt("b", "t2"), prompt("c", "t3")]); - - await rewind(timeline, "c", { kind: "after", retainedMessageId: "provider-a" }); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A"]); - - await rewind(timeline, "a", { kind: "after", retainedMessageId: null }); - expect(await timeline.readRows(scope)).toEqual([]); - }); - - it("keeps the rewind open when the provider's boundary is after the target", async () => { - const { timeline } = await conversation(); - const intent = await timeline.beginRewind(scope, "a"); - await timeline.recordRewindCut(scope, intent.id, { kind: "after", retainedMessageId: "a" }); - - await expect(timeline.completeRewind(scope, intent.id)).rejects.toBeInstanceOf( - TimelineRewindBoundaryError, - ); - expect((await timeline.describe(scope))?.rewind).toMatchObject({ id: intent.id }); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); - }); - - it("keeps active history and the open rewind when the retirement fails to commit", async () => { - const { backend, timeline } = await conversation(); - const intent = await timeline.beginRewind(scope, "b"); - await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); - backend.failNextRetire = true; - - await expect(timeline.completeRewind(scope, intent.id)).rejects.toThrow("disk full"); - expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); - expect(await timeline.describe(scope)).toMatchObject({ retirementCount: 0 }); - - expect(await timeline.completeRewind(scope, intent.id)).toMatchObject({ - kind: "retired", - retirement: { seq: 6, retireFromSeq: 3 }, - }); - }); - - it("judges an unsettled rewind from the provider's own history", async () => { - const { timeline } = await conversation(); - await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - const intent = await timeline.beginRewind(scope, "b"); - - expect(await timeline.judgeRewind(scope, intent.id, ["a", "b", "c"])).toEqual({ - kind: "unchanged", - }); - expect(await timeline.judgeRewind(scope, intent.id, ["a"])).toEqual({ - kind: "applied", - cut: { kind: "after", retainedMessageId: "a" }, - }); - expect(await timeline.judgeRewind(scope, intent.id, ["a", "b"])).toMatchObject({ - kind: "unknown", - reason: "The provider history ends after the rewind target", - }); - expect(await timeline.judgeRewind(scope, intent.id, ["a", "x"])).toMatchObject({ - kind: "unknown", - reason: "The provider history differs from Paseo's", - }); - expect(await timeline.judgeRewind(scope, intent.id, [])).toMatchObject({ - kind: "unknown", - reason: "The provider history has no prompts", - }); - expect(await timeline.judgeRewind(scope, "other", ["a"])).toEqual({ kind: "absent" }); - expect(texts(await timeline.readRows(scope))).toEqual([ - "a", - "A", - "b", - "tool_call", - "B", - "c", - "C", - ]); - expect((await timeline.describe(scope))?.rewind).toMatchObject({ id: intent.id }); - }); - - it("judges against provider ids and skips prompts the provider never acknowledged", async () => { - const timeline = new AgentTimeline(createBackend()); - await timeline.create(scope, { epoch: "e" }); - for (const id of ["a", "b", "c"]) { - await timeline.ingest(scope, { - item: { type: "user_message", text: id, messageId: id, clientMessageId: id }, + it("retires the rewound turn from active history and keeps it stored", async () => { + const { backend, timeline } = await conversation(); + const published: TimelineRetirementCommit[] = []; + const intent = await timeline.beginRewind(scope, "b"); + await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); + const completion = await timeline.completeRewind(scope, intent.id, { + publish: (commit) => published.push(commit), }); - if (id !== "b") await timeline.attachProviderMessageId(scope, id, `provider-${id}`); - } - const intent = await timeline.beginRewind(scope, "c"); + await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - expect(await timeline.judgeRewind(scope, intent.id, ["provider-a", "provider-c"])).toEqual({ - kind: "unchanged", + expect(completion).toMatchObject({ + kind: "retired", + retirement: { seq: 6, retireFromSeq: 3 }, + epoch: "e", + retirementCount: 1, + }); + expect(published.map((commit) => texts(commit.retiredRows))).toEqual([ + ["b", "tool_call", "B"], + ]); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "c", "C"]); + expect(texts(backend.retiredRows(scope))).toEqual(["b", "tool_call", "B"]); + expect(await timeline.describe(scope)).toEqual({ + epoch: "e", + size: 4, + retirementCount: 1, + rewind: null, + }); + const tail = await timeline.fetch(scope); + expect(texts(tail.rows)).toEqual(["a", "A", "c", "C"]); + expect(tail).toMatchObject({ + epoch: "e", + startSeq: 1, + endSeq: 8, + retirements: [{ seq: 6, retireFromSeq: 3 }], + }); }); - const verdict = await timeline.judgeRewind(scope, intent.id, ["provider-a"]); - expect(verdict).toEqual({ - kind: "applied", - cut: { kind: "after", retainedMessageId: "provider-a" }, - }); - if (verdict.kind !== "applied") throw new Error("expected an applied verdict"); - await timeline.recordRewindCut(scope, intent.id, verdict.cut); - await timeline.completeRewind(scope, intent.id); - expect(texts(await timeline.readRows(scope))).toEqual(["a"]); - }); - it("bounds pages across a thousand rewinds", async () => { - const backend = createBackend(); - const timeline = new AgentTimeline(backend); - await timeline.create(scope, { epoch: "e" }); - await timeline.append(scope, [prompt("kept", "t0"), reply("kept", "t0")]); - for (let index = 0; index < 1000; index++) { - await timeline.append(scope, [prompt(`p${index}`, `t${index + 1}`)]); - await rewind(timeline, `p${index}`); - } + it("advances a caught-up cursor through the retirement to the next turn", async () => { + const { timeline } = await conversation(); + await rewind(timeline, "b"); + await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); - const tail = await timeline.fetch(scope, { limit: 50 }); - expect(texts(tail.rows)).toEqual(["kept", "kept"]); - expect(tail.retirements).toEqual([{ seq: 2002, retireFromSeq: 3 }]); - - let cursor = 2; - let pages = 0; - for (;;) { - const page = await timeline.fetch(scope, { + const catchUp = await timeline.fetch(scope, { direction: "after", - cursor: { epoch: "e", seq: cursor }, - limit: 50, + cursor: { epoch: "e", seq: 5 }, + }); + expect(texts(catchUp.rows)).toEqual(["c", "C"]); + expect(catchUp).toMatchObject({ + reset: false, + startSeq: 6, + endSeq: 8, + hasNewer: false, + retirements: [{ seq: 6, retireFromSeq: 3 }], }); - pages += 1; - expect(page.rows).toEqual([]); - expect(page.retirements.length).toBe(1); - if (page.endSeq === null) throw new Error("after-page lost its cursor"); - cursor = page.endSeq; - if (!page.hasNewer) break; - } - expect(cursor).toBe(2002); - // Each page covers 50 prompts and their 50 retirements. - expect(pages).toBe(20); - }); -}); -describe("provider history", () => { + const retirementOnly = await timeline.fetch(scope, { + direction: "after", + cursor: { epoch: "e", seq: 5 }, + limit: 1, + }); + expect(retirementOnly).toMatchObject({ + rows: [], + startSeq: 6, + endSeq: 6, + hasNewer: true, + retirements: [{ seq: 6, retireFromSeq: 3 }], + }); + const next = await timeline.fetch(scope, { + direction: "after", + cursor: { epoch: "e", seq: 6 }, + }); + expect(texts(next.rows)).toEqual(["c", "C"]); + expect(next.retirements).toEqual([]); + }); + + it("pages a tail whose last turn was rewound with nothing after it", async () => { + const { timeline } = await conversation(); + await rewind(timeline, "b"); + + const tail = await timeline.fetch(scope); + expect(texts(tail.rows)).toEqual(["a", "A"]); + expect(tail).toMatchObject({ endSeq: 6, retirements: [{ seq: 6, retireFromSeq: 3 }] }); + const caughtUp = await timeline.fetch(scope, { + direction: "after", + cursor: { epoch: "e", seq: 2 }, + }); + expect(caughtUp).toMatchObject({ + rows: [], + endSeq: 6, + hasNewer: false, + retirements: [{ seq: 6, retireFromSeq: 3 }], + }); + }); + + it("cannot retire a later turn by completing the same rewind again", async () => { + const { timeline } = await conversation(); + const intent = await timeline.beginRewind(scope, "b"); + await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); + await timeline.completeRewind(scope, intent.id); + await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); + + expect(await timeline.completeRewind(scope, intent.id)).toEqual({ kind: "absent" }); + expect(await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" })).toBeNull(); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "c", "C"]); + expect((await timeline.describe(scope))?.retirementCount).toBe(1); + }); + + it("drops late output of a retired turn and still accepts updates to active items", async () => { + const timeline = new AgentTimeline(createBackend()); + await timeline.create(scope, { epoch: "e" }); + await timeline.append(scope, [ + prompt("a", "t1"), + tool("call-a", "t1", "running"), + prompt("b", "t2"), + tool("call-b", "t2", "running"), + ]); + await rewind(timeline, "b"); + + const late = await timeline.ingest(scope, tool("call-b", "t2", "completed")); + const lateText = await timeline.ingest(scope, reply("B again", "t2")); + const reusedCallId = await timeline.ingest(scope, tool("call-a", "t2", "completed")); + const active = await timeline.ingest(scope, tool("call-a", "t1", "completed")); + + expect([late, lateText, reusedCallId].map((result) => result.kind)).toEqual([ + "dropped", + "dropped", + "dropped", + ]); + expect(active.kind).toBe("recorded"); + const rows = await timeline.readRows(scope); + expect(rows.map((row) => row.item)).toEqual([ + { type: "user_message", text: "a", messageId: "a" }, + expect.objectContaining({ type: "tool_call", callId: "call-a", status: "completed" }), + ]); + }); + + it("allows one open rewind at a time and leaves history alone when abandoned", async () => { + const { timeline } = await conversation(); + const intent = await timeline.beginRewind(scope, "b"); + + await expect(timeline.beginRewind(scope, "a")).rejects.toBeInstanceOf( + TimelineRewindPendingError, + ); + await expect(timeline.completeRewind(scope, intent.id)).rejects.toThrow( + "Cannot complete a rewind before its provider outcome", + ); + await timeline.abandonRewind(scope, intent.id); + expect((await timeline.describe(scope))?.rewind).toBeNull(); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); + await expect(timeline.beginRewind(scope, "missing")).rejects.toBeInstanceOf( + TimelineRewindTargetError, + ); + expect((await timeline.beginRewind(scope, "a")).targetSeq).toBe(1); + }); + + it("retires from the boundary the provider actually kept", async () => { + const { timeline } = await conversation(); + await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); + + const completion = await rewind(timeline, "c", { kind: "after", retainedMessageId: "a" }); + + expect(completion).toMatchObject({ kind: "retired", retirement: { retireFromSeq: 3 } }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A"]); + }); + + it("matches the retained boundary by the provider's message id", async () => { + const backend = createBackend(); + const timeline = new AgentTimeline(backend); + await timeline.create(scope, { epoch: "e" }); + await timeline.ingest(scope, { + item: { type: "user_message", text: "a", messageId: "a", clientMessageId: "client-a" }, + }); + await timeline.attachProviderMessageId(scope, "client-a", "provider-a"); + await timeline.append(scope, [reply("A", "t1"), prompt("b", "t2"), prompt("c", "t3")]); + + await rewind(timeline, "c", { kind: "after", retainedMessageId: "provider-a" }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A"]); + + await rewind(timeline, "a", { kind: "after", retainedMessageId: null }); + expect(await timeline.readRows(scope)).toEqual([]); + }); + + it("keeps the rewind open when the provider's boundary is after the target", async () => { + const { timeline } = await conversation(); + const intent = await timeline.beginRewind(scope, "a"); + await timeline.recordRewindCut(scope, intent.id, { kind: "after", retainedMessageId: "a" }); + + await expect(timeline.completeRewind(scope, intent.id)).rejects.toBeInstanceOf( + TimelineRewindBoundaryError, + ); + expect((await timeline.describe(scope))?.rewind).toMatchObject({ id: intent.id }); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); + }); + + it("keeps active history and the open rewind when the retirement fails to commit", async () => { + const { backend, timeline } = await conversation(); + const intent = await timeline.beginRewind(scope, "b"); + await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); + backend.failNextRetire = true; + + await expect(timeline.completeRewind(scope, intent.id)).rejects.toThrow("disk full"); + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); + expect(await timeline.describe(scope)).toMatchObject({ retirementCount: 0 }); + + expect(await timeline.completeRewind(scope, intent.id)).toMatchObject({ + kind: "retired", + retirement: { seq: 6, retireFromSeq: 3 }, + }); + }); + + it("judges an unsettled rewind from the provider's own history", async () => { + const { timeline } = await conversation(); + await timeline.append(scope, [prompt("c", "t3"), reply("C", "t3")]); + const intent = await timeline.beginRewind(scope, "b"); + + expect(await timeline.judgeRewind(scope, intent.id, ["a", "b", "c"])).toEqual({ + kind: "unchanged", + }); + expect(await timeline.judgeRewind(scope, intent.id, ["a"])).toEqual({ + kind: "applied", + cut: { kind: "after", retainedMessageId: "a" }, + }); + expect(await timeline.judgeRewind(scope, intent.id, ["a", "b"])).toMatchObject({ + kind: "unknown", + reason: "The provider history ends after the rewind target", + }); + expect(await timeline.judgeRewind(scope, intent.id, ["a", "x"])).toMatchObject({ + kind: "unknown", + reason: "The provider history differs from Paseo's", + }); + expect(await timeline.judgeRewind(scope, intent.id, [])).toMatchObject({ + kind: "unknown", + reason: "The provider history has no prompts", + }); + expect(await timeline.judgeRewind(scope, "other", ["a"])).toEqual({ kind: "absent" }); + expect(texts(await timeline.readRows(scope))).toEqual([ + "a", + "A", + "b", + "tool_call", + "B", + "c", + "C", + ]); + expect((await timeline.describe(scope))?.rewind).toMatchObject({ id: intent.id }); + }); + + it("judges against provider ids and skips prompts the provider never acknowledged", async () => { + const timeline = new AgentTimeline(createBackend()); + await timeline.create(scope, { epoch: "e" }); + for (const id of ["a", "b", "c"]) { + await timeline.ingest(scope, { + item: { type: "user_message", text: id, messageId: id, clientMessageId: id }, + }); + if (id !== "b") await timeline.attachProviderMessageId(scope, id, `provider-${id}`); + } + const intent = await timeline.beginRewind(scope, "c"); + + expect(await timeline.judgeRewind(scope, intent.id, ["provider-a", "provider-c"])).toEqual({ + kind: "unchanged", + }); + const verdict = await timeline.judgeRewind(scope, intent.id, ["provider-a"]); + expect(verdict).toEqual({ + kind: "applied", + cut: { kind: "after", retainedMessageId: "provider-a" }, + }); + if (verdict.kind !== "applied") throw new Error("expected an applied verdict"); + await timeline.recordRewindCut(scope, intent.id, verdict.cut); + await timeline.completeRewind(scope, intent.id); + expect(texts(await timeline.readRows(scope))).toEqual(["a"]); + }); + + it("bounds pages across a thousand rewinds", async () => { + const backend = createBackend(); + const timeline = new AgentTimeline(backend); + await timeline.create(scope, { epoch: "e" }); + await timeline.append(scope, [prompt("kept", "t0"), reply("kept", "t0")]); + for (let index = 0; index < 1000; index++) { + await timeline.append(scope, [prompt(`p${index}`, `t${index + 1}`)]); + await rewind(timeline, `p${index}`); + } + + const tail = await timeline.fetch(scope, { limit: 50 }); + expect(texts(tail.rows)).toEqual(["kept", "kept"]); + expect(tail.retirements).toEqual([{ seq: 2002, retireFromSeq: 3 }]); + + let cursor = 2; + let pages = 0; + for (;;) { + const page = await timeline.fetch(scope, { + direction: "after", + cursor: { epoch: "e", seq: cursor }, + limit: 50, + }); + pages += 1; + expect(page.rows).toEqual([]); + expect(page.retirements.length).toBe(1); + if (page.endSeq === null) throw new Error("after-page lost its cursor"); + cursor = page.endSeq; + if (!page.hasNewer) break; + } + expect(cursor).toBe(2002); + // Each page covers 50 prompts and their 50 retirements. + expect(pages).toBe(20); + }); + + const child = (subagentId: string): TimelineScope => ({ agentId: "agent", subagentId }); + + function descriptor( + id: string, + fields: { toolCallId?: string; parentSubagentId?: string }, + ): ProviderSubagentDescriptor { + return { + id, + parentAgentId: "agent", + parentSubagentId: fields.parentSubagentId ?? null, + provider: "claude", + title: null, + description: null, + status: "running", + createdAt: "2026-07-12T10:00:00.000Z", + updatedAt: "2026-07-12T10:00:00.000Z", + toolCallId: fields.toolCallId ?? null, + cwd: null, + subtitle: null, + }; + } + + async function conversationWithChildren(): Promise<{ + backend: DeferredTimelineBackend; + timeline: AgentTimeline; + }> { + const { backend, timeline } = await conversation(); + await timeline.recordChild(child("kept"), () => descriptor("kept", {})); + await timeline.recordChild(child("rewound"), () => + descriptor("rewound", { toolCallId: "call-b" }), + ); + await timeline.recordChild(child("grandchild"), () => + descriptor("grandchild", { parentSubagentId: "rewound" }), + ); + await timeline.recordChildEntries(child("rewound"), [reply("working", "t2")]); + return { backend, timeline }; + } + + async function children(timeline: AgentTimeline): Promise<[string, boolean][]> { + return (await timeline.listChildren("agent")) + .map((stored): [string, boolean] => [stored.subagentId, stored.retired]) + .sort(([left], [right]) => left.localeCompare(right)); + } + + it("retires the children the rewound turn spawned, and their descendants, with the turn", async () => { + const { backend, timeline } = await conversationWithChildren(); + + expect(await rewind(timeline, "b")).toMatchObject({ + kind: "retired", + retiredChildIds: ["rewound", "grandchild"], + }); + + const stored: [string, boolean][] = [ + ["grandchild", true], + ["kept", false], + ["rewound", true], + ]; + expect(await children(timeline)).toEqual(stored); + // A retired child keeps its history, and a late provider event changes nothing. + expect(texts((await timeline.fetch(child("rewound"))).rows)).toEqual(["working"]); + expect( + await timeline.recordChild(child("rewound"), () => descriptor("rewound", {})), + ).toBeNull(); + expect(await timeline.recordChildEntries(child("rewound"), [reply("late", "t2")])).toBeNull(); + expect(await timeline.removeChild(child("rewound"))).toBe(false); + expect(texts((await timeline.fetch(child("rewound"))).rows)).toEqual(["working"]); + // The next owner over the same backend reads the same children. + expect(await children(new AgentTimeline(backend))).toEqual(stored); + }); + + it("leaves the conversation and its children as they were when the retirement fails", async () => { + const { backend, timeline } = await conversationWithChildren(); + const intent = await timeline.beginRewind(scope, "b"); + await timeline.recordRewindCut(scope, intent.id, { kind: "at_target" }); + backend.failNextRetire = true; + + await expect(timeline.completeRewind(scope, intent.id)).rejects.toThrow("disk full"); + + expect(texts(await timeline.readRows(scope))).toEqual(["a", "A", "b", "tool_call", "B"]); + expect(await children(timeline)).toEqual([ + ["grandchild", false], + ["kept", false], + ["rewound", false], + ]); + expect((await timeline.describe(scope))?.rewind?.id).toBe(intent.id); + expect(await timeline.completeRewind(scope, intent.id)).toMatchObject({ + kind: "retired", + retiredChildIds: ["rewound", "grandchild"], + }); + }); + + it("deletes every child history with the agent's, retired ones included", async () => { + const { backend, timeline } = await conversationWithChildren(); + await rewind(timeline, "b"); + + await timeline.delete(scope); + + const next = new AgentTimeline(backend); + expect(await next.listChildren("agent")).toEqual([]); + for (const subagentId of ["kept", "rewound", "grandchild"]) { + expect(await next.describe(child(subagentId))).toBeNull(); + } + }); + }, +); + +describe.each(TIMELINE_CONTRACT_BACKENDS)("provider history over a $name backend", ({ create }) => { const scope = root("agent"); function prompt(messageId: string): TimelineEntry { @@ -1069,7 +1188,7 @@ describe("provider history", () => { backend: DeferredTimelineBackend; timeline: AgentTimeline; }> { - const backend = new DeferredTimelineBackend(); + const backend = new DeferredTimelineBackend(create()); const timeline = new AgentTimeline(backend); await timeline.create(scope, { epoch: "e" }); await timeline.append(scope, entries); @@ -1077,7 +1196,7 @@ describe("provider history", () => { } it("imports history into a scope that has none, once", async () => { - const timeline = new AgentTimeline(new DeferredTimelineBackend()); + const timeline = new AgentTimeline(new DeferredTimelineBackend(create())); const history = [prompt("a"), reply("A"), prompt("b"), reply("B")]; const first = await timeline.importHistory(scope, history); diff --git a/packages/server/src/server/agent/timeline/index.ts b/packages/server/src/server/agent/timeline/index.ts index a9992da4cb..4b6bcb1ed3 100644 --- a/packages/server/src/server/agent/timeline/index.ts +++ b/packages/server/src/server/agent/timeline/index.ts @@ -4,6 +4,7 @@ export { TimelineRewindPendingError, TimelineRewindTargetError, type TimelineAppendResult, + type TimelineChild, type TimelineCommit, type TimelineEntry, type TimelineHead, diff --git a/packages/server/src/server/agent/timeline/internal/agent-timeline.ts b/packages/server/src/server/agent/timeline/internal/agent-timeline.ts index beea974a65..05a7cd4cdc 100644 --- a/packages/server/src/server/agent/timeline/internal/agent-timeline.ts +++ b/packages/server/src/server/agent/timeline/internal/agent-timeline.ts @@ -7,8 +7,10 @@ import type { AgentTimelineRow, } from "../../agent-timeline-store-types.js"; import type { ProjectedTimelineRow, TimelineRetirement } from "../../timeline-projection.js"; +import type { ProviderSubagentDescriptor } from "../../provider-subagents/store.js"; import type { AgentTimelineBackend, + TimelineChild, TimelineEntry, TimelineHead, TimelinePrompt, @@ -21,6 +23,7 @@ import { MemoryTimelineBackend } from "./memory-backend.js"; import { planProviderHistory } from "./provider-history.js"; export type { + TimelineChild, TimelineEntry, TimelineHead, TimelineRewindIntent, @@ -40,6 +43,8 @@ export interface TimelineRetirementCommit { retirement: TimelineRetirement; /** The rows that left the active conversation. They stay stored. */ retiredRows: ProjectedTimelineRow[]; + /** The provider children the retired rows spawned, and their descendants. They stay stored. */ + retiredChildIds: string[]; epoch: string; retirementCount: number; } @@ -211,10 +216,62 @@ export class AgentTimeline { }); } + /** The agent's provider children, retired ones included. */ + listChildren(agentId: string): Promise { + return this.run({ agentId }, () => this.backend.children(agentId)); + } + + /** + * Stores the descriptor `update` derives from the child's current one, creating the child when + * absent. A retired child is left as it is and resolves null. + */ + recordChild( + scope: TimelineScope, + update: (previous: ProviderSubagentDescriptor | null) => ProviderSubagentDescriptor, + ): Promise { + return this.run(scope, async () => { + const child = await this.backend.child(scope); + if (child?.retired) return null; + const descriptor = update(child?.descriptor ?? null); + await this.backend.setChild(scope, descriptor); + return descriptor; + }); + } + + /** Appends to a child's history, creating the child when absent. A retired child resolves null. */ + recordChildEntries( + scope: TimelineScope, + entries: readonly TimelineEntry[], + ): Promise { + return this.run(scope, async () => { + const child = await this.backend.child(scope); + if (child?.retired) return null; + if (!child) this.track(scope, await this.backend.create(scope, {})); + return entries.length === 0 ? [] : await this.commit(scope, entries); + }); + } + + /** Deletes a child's history. A retired child stays stored and resolves false. */ + removeChild(scope: TimelineScope): Promise { + return this.run(scope, async () => { + const child = await this.backend.child(scope); + if (child?.retired) return false; + await this.backend.delete(scope); + this.sizes.delete(sizeKey(scope)); + return true; + }); + } + + /** Deleting an agent's root history deletes all of its children's too. */ delete(scope: TimelineScope): Promise { return this.run(scope, async () => { await this.backend.delete(scope); this.sizes.delete(sizeKey(scope)); + if (scope.subagentId !== undefined) return; + const childPrefix = sizeKey({ agentId: scope.agentId, subagentId: "" }); + for (const key of this.sizes.keys()) { + if (key.startsWith(childPrefix)) this.sizes.delete(key); + } }); } @@ -379,11 +436,15 @@ export class AgentTimeline { if (!intent.cut) throw new Error("Cannot complete a rewind before its provider outcome"); const retireFromSeq = resolveRetireFromSeq(intent, await this.backend.prompts(scope)); if (retireFromSeq === null) throw new TimelineRewindBoundaryError(intent); - const result = await this.backend.retire(scope, { retireFromSeq }); + const result = await this.backend.retire(scope, { + retireFromSeq, + selectChildren: childrenSpawnedBy, + }); this.track(scope, result.head); const commit: TimelineRetirementCommit = { retirement: result.retirement, retiredRows: result.retiredRows, + retiredChildIds: result.retiredChildIds, epoch: result.head.epoch, retirementCount: result.head.retirementCount, }; @@ -523,6 +584,35 @@ function judgeRewindEvidence( }; } +/** The listed children that retired tool calls spawned, and their descendants. */ +function childrenSpawnedBy( + retiredRows: readonly ProjectedTimelineRow[], + children: readonly TimelineChild[], +): string[] { + const toolCallIds = new Set( + retiredRows.flatMap((row) => (row.item.type === "tool_call" ? [row.item.callId] : [])), + ); + const listed = children.flatMap((child) => + !child.retired && child.descriptor ? [child.descriptor] : [], + ); + const retired = new Set( + listed + .filter((child) => child.toolCallId !== null && toolCallIds.has(child.toolCallId)) + .map((child) => child.id), + ); + let grew = retired.size > 0; + while (grew) { + grew = false; + for (const child of listed) { + if (retired.has(child.id) || child.parentSubagentId === null) continue; + if (!retired.has(child.parentSubagentId)) continue; + retired.add(child.id); + grew = true; + } + } + return [...retired]; +} + function sizeKey(scope: TimelineScope): string { return scope.subagentId === undefined ? scope.agentId : `${scope.agentId}\0${scope.subagentId}`; } diff --git a/packages/server/src/server/agent/timeline/internal/backend.ts b/packages/server/src/server/agent/timeline/internal/backend.ts index 8b60701a9d..098897ef36 100644 --- a/packages/server/src/server/agent/timeline/internal/backend.ts +++ b/packages/server/src/server/agent/timeline/internal/backend.ts @@ -6,6 +6,7 @@ import type { } from "../../agent-timeline-store-types.js"; import type { ConversationRewindCut } from "../../agent-sdk-types.js"; import type { ProjectedTimelineRow, TimelineRetirement } from "../../timeline-projection.js"; +import type { ProviderSubagentDescriptor } from "../../provider-subagents/store.js"; /** Root history uses `agentId` alone; provider-child history adds the child's id. */ export interface TimelineScope { @@ -63,6 +64,21 @@ export interface TimelinePrompt { submitted: boolean; } +/** A provider child of an agent: its listing, stored with the child's scope. */ +export interface TimelineChild { + subagentId: string; + /** Null until the provider describes the child; such a child is not listed. */ + descriptor: ProviderSubagentDescriptor | null; + /** Spawned by a rewound turn. Its rows stay stored; it is no longer listed. */ + retired: boolean; +} + +/** Picks the children a retirement takes with it, from the rows it retires. */ +export type RetiredChildSelector = ( + retiredRows: readonly ProjectedTimelineRow[], + children: readonly TimelineChild[], +) => readonly string[]; + export interface TimelineBackendAppendResult { rows: AgentTimelineRow[]; head: TimelineHead; @@ -72,10 +88,12 @@ export interface TimelineBackendRetireResult { retirement: TimelineRetirement; /** The rows that left the active conversation. They stay stored. */ retiredRows: ProjectedTimelineRow[]; + /** The children the retirement marked retired. */ + retiredChildIds: string[]; head: TimelineHead; } -type MaybePromise = T | Promise; +export type MaybePromise = T | Promise; /** * Storage for the timeline owner. Each mutating call is one atomic commit: it either applies @@ -85,6 +103,7 @@ type MaybePromise = T | Promise; */ export interface AgentTimelineBackend { create(scope: TimelineScope, seed: TimelineSeed): MaybePromise; + /** Deleting a root scope deletes every child scope of the agent too, retired ones included. */ delete(scope: TimelineScope): MaybePromise; head(scope: TimelineScope): MaybePromise; /** Allocates the next sequence numbers and folds the entries into the projection. */ @@ -99,12 +118,19 @@ export interface AgentTimelineBackend { setRewind(scope: TimelineScope, rewind: TimelineRewindIntent | null): MaybePromise; /** * Allocates the next sequence number for a retirement of the active rows anchored at or after - * `retireFromSeq`, and clears the rewind intent, in one commit. + * `retireFromSeq`, marks the children `selectChildren` picks retired, and clears the rewind + * intent, in one commit. `selectChildren` is pure. */ retire( scope: TimelineScope, - input: { retireFromSeq: number }, + input: { retireFromSeq: number; selectChildren: RetiredChildSelector }, ): MaybePromise; + /** The agent's children, retired ones included, in no particular order. */ + children(agentId: string): MaybePromise; + /** The child of a child scope; null when the scope does not exist. */ + child(scope: TimelineScope): MaybePromise; + /** Stores a child's descriptor, creating its scope when absent. */ + setChild(scope: TimelineScope, descriptor: ProviderSubagentDescriptor): MaybePromise; attachProviderMessageId( scope: TimelineScope, clientMessageId: string, diff --git a/packages/server/src/server/agent/timeline/internal/memory-backend.ts b/packages/server/src/server/agent/timeline/internal/memory-backend.ts index 7c04a5c86d..7117a8d863 100644 --- a/packages/server/src/server/agent/timeline/internal/memory-backend.ts +++ b/packages/server/src/server/agent/timeline/internal/memory-backend.ts @@ -7,10 +7,13 @@ import type { AgentTimelineRow, } from "../../agent-timeline-store-types.js"; import { TimelineProjection, type ProjectedTimelineRow } from "../../timeline-projection.js"; +import type { ProviderSubagentDescriptor } from "../../provider-subagents/store.js"; import type { AgentTimelineBackend, TimelineBackendAppendResult, + RetiredChildSelector, TimelineBackendRetireResult, + TimelineChild, TimelineEntry, TimelineHead, TimelinePrompt, @@ -29,12 +32,21 @@ interface MemoryTimelineState { retired: ProjectedTimelineRow[]; retirementCount: number; rewind: TimelineRewindIntent | null; + /** Child scopes only. */ + descriptor: ProviderSubagentDescriptor | null; + childRetired: boolean; } const DEFAULT_TIMELINE_FETCH_LIMIT = 200; function scopeKey(scope: TimelineScope): string { - return scope.subagentId === undefined ? scope.agentId : `${scope.agentId}\0${scope.subagentId}`; + return scope.subagentId === undefined + ? scope.agentId + : childKeyPrefix(scope.agentId) + scope.subagentId; +} + +function childKeyPrefix(agentId: string): string { + return `${agentId}\0`; } function cloneRow(row: T): T { @@ -71,6 +83,8 @@ export class MemoryTimelineBackend implements AgentTimelineBackend { retired: [], retirementCount: 0, rewind: null, + descriptor: null, + childRetired: false, }; this.states.set(scopeKey(scope), state); return this.describe(state); @@ -78,6 +92,40 @@ export class MemoryTimelineBackend implements AgentTimelineBackend { delete(scope: TimelineScope): void { this.states.delete(scopeKey(scope)); + if (scope.subagentId !== undefined) return; + const prefix = childKeyPrefix(scope.agentId); + for (const key of this.states.keys()) { + if (key.startsWith(prefix)) this.states.delete(key); + } + } + + children(agentId: string): TimelineChild[] { + const prefix = childKeyPrefix(agentId); + const children: TimelineChild[] = []; + for (const [key, state] of this.states) { + if (!key.startsWith(prefix)) continue; + children.push({ + subagentId: key.slice(prefix.length), + descriptor: state.descriptor ? { ...state.descriptor } : null, + retired: state.childRetired, + }); + } + return children; + } + + child(scope: TimelineScope): TimelineChild | null { + const state = this.states.get(scopeKey(scope)); + if (!state || scope.subagentId === undefined) return null; + return { + subagentId: scope.subagentId, + descriptor: state.descriptor ? { ...state.descriptor } : null, + retired: state.childRetired, + }; + } + + setChild(scope: TimelineScope, descriptor: ProviderSubagentDescriptor): void { + if (!this.states.has(scopeKey(scope))) this.create(scope, {}); + this.requireState(scope).descriptor = { ...descriptor }; } head(scope: TimelineScope): TimelineHead | null { @@ -121,16 +169,29 @@ export class MemoryTimelineBackend implements AgentTimelineBackend { return this.describe(state); } - retire(scope: TimelineScope, input: { retireFromSeq: number }): TimelineBackendRetireResult { + retire( + scope: TimelineScope, + input: { retireFromSeq: number; selectChildren: RetiredChildSelector }, + ): TimelineBackendRetireResult { const state = this.requireState(scope); const retirement = { seq: state.nextSeq, retireFromSeq: input.retireFromSeq }; const retiredRows = state.projection.retire(retirement); + const retiredChildIds = [...input.selectChildren(retiredRows, this.children(scope.agentId))]; + for (const subagentId of retiredChildIds) { + const child = this.states.get(scopeKey({ agentId: scope.agentId, subagentId })); + if (child) child.childRetired = true; + } state.retired.push(...retiredRows); state.nextSeq = retirement.seq + 1; if (state.minSeq === 0) state.minSeq = retirement.seq; state.retirementCount += 1; state.rewind = null; - return { retirement, retiredRows: retiredRows.map(cloneRow), head: this.describe(state) }; + return { + retirement, + retiredRows: retiredRows.map(cloneRow), + retiredChildIds, + head: this.describe(state), + }; } /** Retired rows, oldest retirement first. Not part of the backend contract; tests read it. */ diff --git a/packages/server/src/server/agent/timeline/testing.ts b/packages/server/src/server/agent/timeline/testing.ts index 1933d9b35d..9f5055e873 100644 --- a/packages/server/src/server/agent/timeline/testing.ts +++ b/packages/server/src/server/agent/timeline/testing.ts @@ -6,10 +6,14 @@ import type { AgentTimelineRow, } from "../agent-timeline-store-types.js"; import type { ProjectedTimelineRow } from "../timeline-projection.js"; +import type { ProviderSubagentDescriptor } from "../provider-subagents/store.js"; import type { AgentTimelineBackend, + MaybePromise, + RetiredChildSelector, TimelineBackendAppendResult, TimelineBackendRetireResult, + TimelineChild, TimelineEntry, TimelineHead, TimelinePrompt, @@ -21,69 +25,96 @@ import { MemoryTimelineBackend } from "./internal/memory-backend.js"; /** * Testing entry point of the timeline module. Production code imports `./index.js`. - * - * Defers every call by a turn of the event loop, the shape of an I/O-backed store, and records - * the committed appends and whole-history reads. + */ + +/** A backend tests can also ask for the rows rewinds retired, which stay stored. */ +export interface InspectableTimelineBackend extends AgentTimelineBackend { + retiredRows(scope: TimelineScope): ProjectedTimelineRow[]; +} + +/** + * The backends the timeline contract suite (`agent-timeline.test.ts`) runs against. Every + * production backend is listed here. + */ +export const TIMELINE_CONTRACT_BACKENDS: ReadonlyArray<{ + name: string; + create: () => InspectableTimelineBackend; +}> = [{ name: "memory", create: () => new MemoryTimelineBackend() }]; + +/** + * Wraps a backend: defers every call by a turn of the event loop, the shape of an I/O-backed + * store, records the committed appends and whole-history reads, and fails the next append or + * retirement on request. */ export class DeferredTimelineBackend implements AgentTimelineBackend { - private readonly memory = new MemoryTimelineBackend(); readonly appends: TimelineEntry[][] = []; readRowsCalls = 0; failNextAppend = false; failNextRetire = false; - private async defer(call: () => T): Promise { + constructor(private readonly inner: InspectableTimelineBackend = new MemoryTimelineBackend()) {} + + private async defer(call: () => MaybePromise): Promise { await nextTurn(); - return call(); + return await call(); } create(scope: TimelineScope, seed: TimelineSeed): Promise { - return this.defer(() => this.memory.create(scope, seed)); + return this.defer(() => this.inner.create(scope, seed)); } delete(scope: TimelineScope): Promise { - return this.defer(() => this.memory.delete(scope)); + return this.defer(() => this.inner.delete(scope)); } head(scope: TimelineScope): Promise { - return this.defer(() => this.memory.head(scope)); + return this.defer(() => this.inner.head(scope)); } append( scope: TimelineScope, entries: readonly TimelineEntry[], ): Promise { - return this.defer(() => { + return this.defer(async () => { if (this.failNextAppend) { this.failNextAppend = false; throw new Error("disk full"); } - const result = this.memory.append(scope, entries); + const result = await this.inner.append(scope, entries); this.appends.push([...entries]); return result; }); } isRetired(scope: TimelineScope, entry: TimelineEntry): Promise { - return this.defer(() => this.memory.isRetired(scope, entry)); + return this.defer(() => this.inner.isRetired(scope, entry)); } prompts(scope: TimelineScope): Promise { - return this.defer(() => this.memory.prompts(scope)); + return this.defer(() => this.inner.prompts(scope)); } setRewind(scope: TimelineScope, rewind: TimelineRewindIntent | null): Promise { - return this.defer(() => this.memory.setRewind(scope, rewind)); + return this.defer(() => this.inner.setRewind(scope, rewind)); + } + children(agentId: string): Promise { + return this.defer(() => this.inner.children(agentId)); + } + child(scope: TimelineScope): Promise { + return this.defer(() => this.inner.child(scope)); + } + setChild(scope: TimelineScope, descriptor: ProviderSubagentDescriptor): Promise { + return this.defer(() => this.inner.setChild(scope, descriptor)); } retire( scope: TimelineScope, - input: { retireFromSeq: number }, + input: { retireFromSeq: number; selectChildren: RetiredChildSelector }, ): Promise { return this.defer(() => { if (this.failNextRetire) { this.failNextRetire = false; throw new Error("disk full"); } - return this.memory.retire(scope, input); + return this.inner.retire(scope, input); }); } /** Rows rewinds retired; they stay stored. */ retiredRows(scope: TimelineScope): ProjectedTimelineRow[] { - return this.memory.retiredRows(scope); + return this.inner.retiredRows(scope); } attachProviderMessageId( scope: TimelineScope, @@ -91,29 +122,29 @@ export class DeferredTimelineBackend implements AgentTimelineBackend { providerMessageId: string, ): Promise { return this.defer(() => - this.memory.attachProviderMessageId(scope, clientMessageId, providerMessageId), + this.inner.attachProviderMessageId(scope, clientMessageId, providerMessageId), ); } findSubmittedPrompt( scope: TimelineScope, clientMessageId: string, ): Promise { - return this.defer(() => this.memory.findSubmittedPrompt(scope, clientMessageId)); + return this.defer(() => this.inner.findSubmittedPrompt(scope, clientMessageId)); } fetch( scope: TimelineScope, options: AgentTimelineFetchOptions, ): Promise { - return this.defer(() => this.memory.fetch(scope, options)); + return this.defer(() => this.inner.fetch(scope, options)); } readRows(scope: TimelineScope): Promise { this.readRowsCalls += 1; - return this.defer(() => this.memory.readRows(scope)); + return this.defer(() => this.inner.readRows(scope)); } lastItem(scope: TimelineScope): Promise { - return this.defer(() => this.memory.lastItem(scope)); + return this.defer(() => this.inner.lastItem(scope)); } lastAssistantMessage(scope: TimelineScope): Promise { - return this.defer(() => this.memory.lastAssistantMessage(scope)); + return this.defer(() => this.inner.lastAssistantMessage(scope)); } } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 84c05d7f1b..167bc96605 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -8158,7 +8158,7 @@ export class Session { payload: { requestId: msg.requestId, parentAgentId: msg.parentAgentId, - subagents: this.agentManager.listProviderSubagents(msg.parentAgentId), + subagents: await this.agentManager.listProviderSubagents(msg.parentAgentId), error: null, }, }); @@ -8182,7 +8182,10 @@ export class Session { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); try { await this.loadProviderSubagentParent(msg.parentAgentId); - const descriptor = this.agentManager.getProviderSubagent(msg.parentAgentId, msg.subagentId); + const descriptor = await this.agentManager.getProviderSubagent( + msg.parentAgentId, + msg.subagentId, + ); if (!descriptor) { throw new Error("Provider subagent not found"); }