From 706ae3d9bb5700c7f2ba1235009011c001393e82 Mon Sep 17 00:00:00 2001 From: Ethan Greenfeld <39423472+ebg1223@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:22:44 -0700 Subject: [PATCH] Read Claude subagents from the SDK task protocol (#2498) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(claude): derive subagents from the SDK task protocol Claude Code announces subagent identity and lifecycle on the SDK stream (task_started, task_updated, task_notification). Paseo consumed only task_notification and reconstructed the rest from sidechain frames, in two independent derivations that had drifted apart. Both paths now translate into one small observation vocabulary and fold it with one function, so identity and status are derived once. The fold is stateless: the task protocol is edge-triggered and the store's sticky merge already accumulates. - Replay links a subagent through the agent-.meta.json sidecar, which carries the same Task tool_use id the live stream keys on. The agentId text scrape stays as a fallback for sessions recorded without it. - The sidechain tracker keeps mapping frames to timeline items but no longer owns the descriptor when the task protocol is active, detected from the session rather than from a version string. - Adds replay coverage, which did not exist. Subagent ids are unchanged, so no protocol, store, or client change. Co-Authored-By: Claude Opus 5 (1M context) * fix(claude): keep non-subagent tasks out of the track, open the child timeline with its prompt Backgrounded shell commands announce as task_started with a tool_use_id but task_type local_bash and no subagent_type, so filtering on the id alone put `sleep 20` in the subagents track. Discriminate on task_type, falling back to subagent_type for releases that predate it. task_started also carries the prompt. Emitting it as the child's first timeline item means the pane no longer opens mid-conversation, and matches replay, where the sidechain transcript already begins with the prompt. Co-Authored-By: Claude Opus 5 (1M context) * fix(claude): drop status for tasks that were never declared as subagents task_notification also fires for backgrounded shell commands and carries a tool_use_id. Routing status off that id readmitted exactly what the declaration filter rejected, as a descriptor with a status and no identity — rendered in the track as a nameless "Loading..." row. Status now routes only through the task_id map populated at declaration. Co-Authored-By: Claude Opus 5 (1M context) * fix(app): name a subagent row after its task, not its type The descriptor carries the subagent type in title and the task in description. The client collapsed them and rendered only the type, so every child in a fan-out read "general-purpose". The task is what tells siblings apart, so it names the row and the type moves to the muted secondary line, which the track computed but never drew. Providers that report no task keep the previous behavior. Co-Authored-By: Claude Opus 5 (1M context) * fix(claude): label the parent's Task card from the subagent declaration A backgrounded Task subagent emits no frames carrying parent_tool_use_id, so the sidechain tracker never runs for one and never supplies the sub_agent detail. Its card in the parent transcript rendered as a bare "Task". The declaration already carries the type and the task, so the card is built from it. For a foreground subagent the tracker still emits the same card enriched with the child's action log; the two collapse on the shared tool-call id, so that case is unchanged. Co-Authored-By: Claude Opus 5 (1M context) * feat(subagents): surface a subagent's model, effort, and cost Adds model, effort and usage to the provider subagent descriptor, populated from observations the child itself reports. - model: read per assistant frame, live and on replay. Never snapshotted, since a provider can swap models mid-flight on overload or refusal fallback, and never inherited from the parent. - effort: replay only. The SDK's assistant message does not carry it and it can be silently downgraded for the chosen model, so a live value would be a guess. - usage: task_progress reports tokens and tool count once per tool use, and task_notification reports the terminal totals. This matters most for backgrounded subagents, which emit no sidechain frames and would otherwise show no sign of activity at all. Status becomes optional on the upsert event: a report that only carries a model or a cost says nothing about whether the child is still running, and must not revert a finished one. Usage merges rather than replaces, because duration only arrives at the end. Fields are optional on the wire, so an older daemon simply omits them and the client renders nothing rather than substituting the parent's settings. Co-Authored-By: Claude Opus 5 (1M context) * feat(claude): read subagent effort from hooks Effort appears nowhere on the SDK message stream — verified by scanning every message type at depth — and the level Paseo requests is not necessarily the one that runs, since a model that does not support it is silently downgraded. Hooks are the source. Every hook input carries agent_id and effort.level, the active level for that turn after any downgrade, and agent_id is the same id task_started calls task_id, so it routes through the table already built at declaration. PreToolUse, PostToolUse and SubagentStop are registered purely to observe; they record what they see and always return empty, so they cannot affect tool execution or turn control. Paseo's hooks merge with any the user configured rather than replacing them: extraClaudeOptions is spread after the base options, so assigning would have silently stopped effort being observed for anyone with their own hooks. Verified end to end through ClaudeAgentClient against a real session: a subagent descriptor now carries model, effort and usage together. Co-Authored-By: Claude Opus 5 (1M context) * feat(subagents): show a subagent's runtime facts in its pane, and its cost on replay The track row carried model, effort and cost; the pane carried none of it, so opening a subagent showed only its task. The pane gets a header strip rather than a longer tab subtitle: the subtitle feeds the width-constrained tab strip, whose job is to name the task. Duration appears in the pane, which has room, and stays out of the dense row. Replay had no usage source, so a reopened agent lost its subagents' token and tool counts permanently — descriptors are in-memory and cleared when an agent closes. Usage is now derived from the child's own transcript, matching the live producer exactly: total tokens are the four counters summed on the last assistant turn (a context reading, not a cumulative sum, which would multiply the cached prefix by the turn count), and tool uses are tool_use blocks across assistant entries. Duration is an approximation — the transcript's first entry is written after the task starts — and is the one field replay cannot match. Co-Authored-By: Claude Opus 5 (1M context) * docs(claude): record the task protocol's undocumented wire facts The subagent sources depend on Claude Code internals the SDK types do not explain: which announced tasks are subagents, why a status for an undeclared task is dropped, why task ids outlive a turn, why effort is only reachable through hooks, and why a replayed session must skip its grandchildren. Co-Authored-By: Claude Opus 5 (1M context) * Generalize provider subagent presentation --------- Co-authored-by: Claude Opus 5 (1M context) Co-authored-by: Mohamed Boudra --- docs/agent-lifecycle.md | 14 + .../src/panels/provider-subagent-panel.tsx | 8 +- packages/app/src/subagents/select.test.ts | 9 + packages/app/src/subagents/select.ts | 15 +- .../src/subagents/track-presentation.test.ts | 100 +++- .../app/src/subagents/track-presentation.ts | 12 +- packages/app/src/subagents/track.tsx | 13 + packages/protocol/src/messages.ts | 3 + .../server/agent/provider-subagents/store.ts | 30 +- .../claude/agent.background-subagent.test.ts | 183 +++++++ .../claude/agent.sub-agent-replay.test.ts | 226 ++++++++ .../claude/agent.subagent-interrupt.test.ts | 183 +++++++ .../server/agent/providers/claude/agent.ts | 323 ++++++++---- .../agent/providers/claude/hooks.test.ts | 36 ++ .../server/agent/providers/claude/hooks.ts | 21 + .../server/agent/providers/claude/models.ts | 29 + .../providers/claude/sidechain-tracker.ts | 47 +- .../claude/subagents/live-source.test.ts | 489 +++++++++++++++++ .../providers/claude/subagents/live-source.ts | 395 ++++++++++++++ .../claude/subagents/observation.test.ts | 139 +++++ .../providers/claude/subagents/observation.ts | 91 ++++ .../claude/subagents/presentation.test.ts | 31 ++ .../claude/subagents/presentation.ts | 53 ++ .../claude/subagents/replay-source.test.ts | 497 ++++++++++++++++++ .../claude/subagents/replay-source.ts | 312 +++++++++++ .../providers/omp/test-utils/omp-harness.ts | 4 +- 26 files changed, 3140 insertions(+), 123 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/agent.subagent-interrupt.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/hooks.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/hooks.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/live-source.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/observation.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/observation.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/presentation.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/presentation.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/subagents/replay-source.ts diff --git a/docs/agent-lifecycle.md b/docs/agent-lifecycle.md index 680122391..cc6f43626 100644 --- a/docs/agent-lifecycle.md +++ b/docs/agent-lifecycle.md @@ -123,6 +123,20 @@ Clicking either kind opens a workspace tab. A Paseo subagent tab is a normal int Provider timelines use the same structural timeline item format but deliberately have a separate lifecycle and transport. A provider thread/session identifier is not a Paseo agent identifier, and closing its tab is always layout-only. +Provider descriptors may include one compact subtitle. The provider owns its contents and formatting; clients display and truncate it without interpreting provider-specific model, thinking, or usage fields. + +### Claude provider subagents: the task protocol + +Claude Code announces subagent lifecycle on the SDK stream (`task_started` / `task_updated` / `task_notification` / `task_progress`), and Paseo reads those announcements rather than reconstructing them from sidechain frames. The live source (`subagents/live-source.ts`) and the replay source (`subagents/replay-source.ts`) both translate into one observation vocabulary (`subagents/observation.ts`), so a fact is derived once for both paths instead of once per path. Gotchas that are not obvious from the SDK types: + +- **Not every announced task is a subagent.** A backgrounded shell announces as `task_type: "local_bash"` with a `tool_use_id` and no `subagent_type`; workflows announce as `local_workflow`; ambient housekeeping sets `skip_transcript`. Filtering on the presence of a `tool_use_id` alone puts `sleep 20` in the subagents track. +- **A task that was never declared gets no descriptor, by any route.** Filtered tasks still emit `task_notification`s carrying a `tool_use_id`, and still emit frames carrying `parent_tool_use_id`. Attributing either produces a descriptor with no identity and a defaulted `running` status — a nameless row that never finishes. Status, presentation updates, and sidechain frames all route through the declaration table. +- **Task ids are session-scoped, not turn-scoped.** Cancelling a turn must not clear the routing table: a backgrounded child settles after the interrupt and needs its descriptor to still exist. Cancellation instead terminalizes the declared children that were running in the foreground, and a later `task_notification` is free to correct that guess. Backgrounded children are identified by `task_updated.patch.is_backgrounded`. +- **Effort is only reachable through hooks.** It appears nowhere on the message stream at any depth, and the level Paseo requests is not necessarily the level that runs — a model that does not support it is silently downgraded. A hook firing inside a subagent reports the active post-downgrade level next to its `agent_id`, which is the same id `task_started` calls `task_id`. +- **Backgrounded subagents emit no frames carrying `parent_tool_use_id` at all.** Everything keyed off that field sees nothing for one; they are visible only because the task protocol announces them. +- **On replay, `/subagents/` holds every descendant, not just this session's children.** `agent-.meta.json` carries `spawnDepth`: `1` is a direct child, `2+` was spawned by another subagent and its `toolUseId` names a Task call made inside its parent's session, which nothing in this transcript can resolve. Replaying those adds rows the live stream never showed, each with no Task card and no recoverable outcome, so they render as running forever. One recorded session showed 10 subagents live and would have replayed 22. +- **Replay `totalTokens` is a context-size reading, not cumulative spend.** Claude Code finalizes a subagent by summing the _last_ assistant message's usage block and shipping that as `usage.total_tokens`. Summing per-entry usage instead multiplies the cached prefix by the turn count and reports a number several times larger than the live path. + Archived Paseo subagents disappear from the track, by design. To remove one from the track without closing its tab, use the **archive button** on the row — it opens a confirm dialog and archives the subagent on confirm. Provider-owned rows have no individual Paseo lifecycle controls. The track header's **Archive finished** action hides finished provider-owned rows in the current app session. Their native sessions and timelines are untouched, and managed Paseo subagents are not archived by this bulk action. If a hidden provider child starts running again, the app brings it back to the track. diff --git a/packages/app/src/panels/provider-subagent-panel.tsx b/packages/app/src/panels/provider-subagent-panel.tsx index a5bac3b7b..f33f1beab 100644 --- a/packages/app/src/panels/provider-subagent-panel.tsx +++ b/packages/app/src/panels/provider-subagent-panel.tsx @@ -45,10 +45,14 @@ function useProviderSubagentDescriptor( (state) => state.sessions[context.serverId]?.agents.get(target.parentAgentId)?.provider, ); const provider = descriptor?.provider ?? parentProvider ?? "agent"; - const label = descriptor?.title?.trim() || descriptor?.description?.trim() || "Subagent"; + // The task names the tab; the subagent type is supporting detail beside the provider. + const subagentType = descriptor?.title?.trim(); + const label = descriptor?.description?.trim() || subagentType || "Subagent"; + const providerLabel = `${formatProviderLabel(provider)} subagent`; return { label, - subtitle: `${formatProviderLabel(provider)} subagent`, + subtitle: + subagentType && subagentType !== label ? `${subagentType} · ${providerLabel}` : providerLabel, tooltip: label, titleState: descriptor ? "ready" : "loading", icon: getProviderIcon(provider), diff --git a/packages/app/src/subagents/select.test.ts b/packages/app/src/subagents/select.test.ts index 7545ea27c..ae110167a 100644 --- a/packages/app/src/subagents/select.test.ts +++ b/packages/app/src/subagents/select.test.ts @@ -76,6 +76,7 @@ describe("selectSubagentsForParent", () => { provider: "codex", title: "Provider child", description: null, + subtitle: "Codex worker · 4.2k tokens", status: "completed", createdAt: "2026-03-08T10:01:00.000Z", updatedAt: "2026-03-08T10:02:00.000Z", @@ -92,6 +93,10 @@ describe("selectSubagentsForParent", () => { (row) => row.id, ), ).toEqual(["provider-child"]); + expect( + selectProviderSubagentsForParent(useProviderSubagentStore.getState(), params, true)[0] + ?.subtitle, + ).toBe("Codex worker · 4.2k tokens"); }); it("hides locally dismissed provider children while retaining their descriptor", () => { @@ -259,6 +264,8 @@ describe("selectSubagentsForParent", () => { id: "child", provider: "claude", title: "Review child", + description: null, + subtitle: null, status: "running", requiresAttention: true, createdAt, @@ -266,11 +273,13 @@ describe("selectSubagentsForParent", () => { ]); expect(Object.keys(rows[0] ?? {}).sort()).toEqual([ "createdAt", + "description", "id", "kind", "provider", "requiresAttention", "status", + "subtitle", "title", ]); expect(rows[0]).not.toHaveProperty("onOpen"); diff --git a/packages/app/src/subagents/select.ts b/packages/app/src/subagents/select.ts index d96c1628a..cc07480b3 100644 --- a/packages/app/src/subagents/select.ts +++ b/packages/app/src/subagents/select.ts @@ -11,6 +11,9 @@ export interface PaseoSubagentRow { id: Agent["id"]; provider: Agent["provider"]; title: Agent["title"]; + /** Managed agents have a real title, so the union's task line is always absent for them. */ + description: null; + subtitle: null; status: Agent["status"]; requiresAttention: Agent["requiresAttention"]; createdAt: Agent["createdAt"]; @@ -21,7 +24,13 @@ export interface ProviderSubagentRow { id: string; parentAgentId: string; provider: ProviderSubagentDescriptorPayload["provider"]; + // `title` is the subagent type ("Explore", "general-purpose") and repeats across a fan-out; + // `description` is the task it was given. Both are carried so presentation can choose which + // one names the row — collapsing them here is what makes every row read alike. title: string | null; + description: string | null; + /** Compact provider-owned context. The app displays it without interpreting its contents. */ + subtitle: string | null; status: ProviderSubagentDescriptorPayload["status"]; requiresAttention: boolean; createdAt: Date; @@ -46,6 +55,8 @@ function toSubagentRow(agent: Agent): SubagentRow { id: agent.id, provider: agent.provider, title: agent.title, + description: null, + subtitle: null, status: agent.status, requiresAttention: agent.requiresAttention, createdAt: agent.createdAt, @@ -97,7 +108,9 @@ export function selectProviderSubagentsForParent( id: subagent.id, parentAgentId: subagent.parentAgentId, provider: subagent.provider, - title: subagent.title ?? subagent.description, + title: subagent.title, + description: subagent.description, + subtitle: subagent.subtitle ?? null, status: subagent.status, requiresAttention: subagent.status === "failed", createdAt: new Date(subagent.createdAt), diff --git a/packages/app/src/subagents/track-presentation.test.ts b/packages/app/src/subagents/track-presentation.test.ts index 5f74fd6bd..0dfd1b3cf 100644 --- a/packages/app/src/subagents/track-presentation.test.ts +++ b/packages/app/src/subagents/track-presentation.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import type { PaseoSubagentRow, SubagentRow } from "./select"; +import type { PaseoSubagentRow, ProviderSubagentRow, SubagentRow } from "./select"; import { buildSubagentRowPresentationData, countFinishedSubagents, @@ -15,6 +15,8 @@ function row( id: overrides.id, provider: overrides.provider ?? "codex", title: overrides.title ?? `Agent ${overrides.id}`, + description: null, + subtitle: null, status: overrides.status ?? "idle", requiresAttention: overrides.requiresAttention ?? false, createdAt: overrides.createdAt ?? new Date("2026-04-20T00:00:00.000Z"), @@ -82,6 +84,8 @@ describe("countFinishedSubagents", () => { parentAgentId: "parent", provider: "claude", title: "running", + description: null, + subtitle: null, status: "running", requiresAttention: false, createdAt: new Date("2026-04-20T00:00:00.000Z"), @@ -92,6 +96,8 @@ describe("countFinishedSubagents", () => { parentAgentId: "parent", provider: "claude", title: "failed", + description: null, + subtitle: null, status: "failed", requiresAttention: true, createdAt: new Date("2026-04-20T00:00:01.000Z"), @@ -166,3 +172,95 @@ describe("buildSubagentRowPresentationData", () => { ).toBe("done"); }); }); + +describe("buildSubagentRowPresentationData for provider rows", () => { + function providerRow(overrides: Partial = {}): ProviderSubagentRow { + return { + kind: "provider", + id: overrides.id ?? "toolu_1", + parentAgentId: "parent", + provider: "claude", + title: "title" in overrides ? (overrides.title ?? null) : "general-purpose", + description: overrides.description ?? null, + subtitle: overrides.subtitle ?? null, + status: overrides.status ?? "running", + requiresAttention: false, + createdAt: overrides.createdAt ?? new Date("2026-07-26T00:00:00.000Z"), + }; + } + + it("names the row after the task and demotes the subagent type", () => { + const presentation = buildSubagentRowPresentationData( + providerRow({ title: "general-purpose", description: "Reply with banana" }), + ); + expect(presentation.label).toBe("Reply with banana"); + expect(presentation.subtitle).toBe("general-purpose"); + }); + + it("tells two siblings of the same type apart", () => { + const left = buildSubagentRowPresentationData( + providerRow({ id: "a", description: "Summarize the docs" }), + ); + const right = buildSubagentRowPresentationData( + providerRow({ id: "b", description: "Reply with banana" }), + ); + expect(left.label).not.toBe(right.label); + }); + + it("keeps type-as-label and an empty subtitle when a provider reports no task", () => { + const presentation = buildSubagentRowPresentationData( + providerRow({ title: "Provider child", description: null }), + ); + expect(presentation.label).toBe("Provider child"); + expect(presentation.subtitle).toBe(""); + }); + + it("stays in the loading state when neither field is known", () => { + const presentation = buildSubagentRowPresentationData( + providerRow({ title: null, description: null }), + ); + expect(presentation.titleState).toBe("loading"); + }); + + it("leaves managed subagent rows with no subtitle", () => { + expect(buildSubagentRowPresentationData(row({ id: "a", title: "Managed" })).subtitle).toBe(""); + }); +}); + +describe("provider-owned row subtitles", () => { + function providerRow(overrides: Partial = {}): ProviderSubagentRow { + return { + kind: "provider", + id: "toolu_1", + parentAgentId: "parent", + provider: "claude", + title: "general-purpose", + description: "Reply with banana", + subtitle: null, + status: "running", + requiresAttention: false, + createdAt: new Date("2026-07-26T00:00:00.000Z"), + ...overrides, + }; + } + + it("displays provider context without interpreting it", () => { + expect( + buildSubagentRowPresentationData( + providerRow({ subtitle: "general-purpose · Opus 5 · High · 16.5k tokens" }), + ).subtitle, + ).toBe("general-purpose · Opus 5 · High · 16.5k tokens"); + }); + + it("falls back to the type when an older provider sends no subtitle", () => { + expect(buildSubagentRowPresentationData(providerRow()).subtitle).toBe("general-purpose"); + }); + + it("does not duplicate the type when it is already the primary label", () => { + expect( + buildSubagentRowPresentationData( + providerRow({ description: null, subtitle: null, title: "general-purpose" }), + ).subtitle, + ).toBe(""); + }); +}); diff --git a/packages/app/src/subagents/track-presentation.ts b/packages/app/src/subagents/track-presentation.ts index c4baa42f8..b1a0fc3b7 100644 --- a/packages/app/src/subagents/track-presentation.ts +++ b/packages/app/src/subagents/track-presentation.ts @@ -18,13 +18,19 @@ export interface SubagentRowPresentationData { } export function buildSubagentRowPresentationData(row: SubagentRow): SubagentRowPresentationData { - const label = resolveRowLabel(row.title); + // The task distinguishes siblings in a fan-out, so it names the row when present. Providers + // own the compact secondary context because model, effort, and usage semantics differ. + const description = resolveRowLabel(row.description); + const title = resolveRowLabel(row.title); + const label = description ?? title; + const providerSubtitle = row.kind === "provider" ? resolveRowLabel(row.subtitle) : null; + const subtitle = providerSubtitle ?? (description ? title : null); const status = presentationStatus(row); return { key: `${row.kind}_subagent_${row.id}`, kind: "agent", label: label ?? "", - subtitle: "", + subtitle: subtitle ?? "", titleState: label ? "ready" : "loading", statusBucket: deriveSidebarStateBucket({ status, @@ -52,7 +58,7 @@ export function countFinishedSubagents(rows: readonly SubagentRow[]): number { return rows.filter((row) => row.kind === "provider" && row.status !== "running").length; } -export function resolveRowLabel(title: SubagentRow["title"]): string | null { +export function resolveRowLabel(title: string | null | undefined): string | null { if (typeof title !== "string") { return null; } diff --git a/packages/app/src/subagents/track.tsx b/packages/app/src/subagents/track.tsx index 1aaa59436..b140b999a 100644 --- a/packages/app/src/subagents/track.tsx +++ b/packages/app/src/subagents/track.tsx @@ -204,6 +204,11 @@ function SubagentsTrackRow({ {displayLabel} + {presentation.subtitle ? ( + + {presentation.subtitle} + + ) : null} {row.kind === "paseo" ? ( ({ fontSize: theme.fontSize.sm, color: theme.colors.foreground, }, + // Keep provider context secondary and bounded so the task remains readable on compact screens. + rowSubtitle: { + flexShrink: 1, + minWidth: 0, + maxWidth: "45%", + fontSize: theme.fontSize.xs, + color: theme.colors.foregroundMuted, + }, actionClusterVisible: { flexDirection: "row", alignItems: "center", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index af44e09f9..114e78d38 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3558,6 +3558,9 @@ export const ProviderSubagentDescriptorPayloadSchema = z.object({ updatedAt: z.string(), toolCallId: z.string().nullable(), cwd: z.string().nullable().optional(), + // Compact provider-owned context for the shared track. Providers choose what belongs here and + // format it for display; clients must not parse provider-specific facts out of this string. + subtitle: z.string().nullable().optional(), }); export type ProviderSubagentDescriptorPayload = z.infer< diff --git a/packages/server/src/server/agent/provider-subagents/store.ts b/packages/server/src/server/agent/provider-subagents/store.ts index 0d05e6426..b29d9f6a5 100644 --- a/packages/server/src/server/agent/provider-subagents/store.ts +++ b/packages/server/src/server/agent/provider-subagents/store.ts @@ -21,6 +21,7 @@ export interface ProviderSubagentDescriptor { updatedAt: string; toolCallId: string | null; cwd: string | null; + subtitle: string | null; } export type ProviderSubagentInputEvent = @@ -29,9 +30,14 @@ export type ProviderSubagentInputEvent = id: string; title?: string | null; description?: string | null; - status: ProviderSubagentStatus; + /** + * Omit to keep the stored status. A presentation-only upsert says nothing about whether the + * child is still running, and must not revert a finished one. + */ + status?: ProviderSubagentStatus; toolCallId?: string | null; cwd?: string | null; + subtitle?: string | null; timestamp?: string; } | { @@ -58,6 +64,15 @@ function storeKey(parentAgentId: string, subagentId: string): string { return `${parentAgentId}\0${subagentId}`; } +/** + * Sticky upsert semantics for a descriptor field: an omitted value preserves what is stored, an + * explicit `null` clears it. Providers observe these fields incrementally, so a partial upsert + * must never blank fields it says nothing about. + */ +function stickyField(next: T | undefined, previous: T | null | undefined): T | null { + return next === undefined ? (previous ?? null) : next; +} + export class ProviderSubagentStore { private readonly descriptors = new Map(); private readonly timelines = new InMemoryAgentTimelineStore(); @@ -100,15 +115,14 @@ export class ProviderSubagentStore { id: event.id, parentAgentId, provider, - title: event.title === undefined ? (previous?.title ?? null) : event.title, - description: - event.description === undefined ? (previous?.description ?? null) : event.description, - status: event.status, + 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: - event.toolCallId === undefined ? (previous?.toolCallId ?? null) : event.toolCallId, - cwd: event.cwd === undefined ? (previous?.cwd ?? null) : event.cwd, + toolCallId: stickyField(event.toolCallId, previous?.toolCallId), + cwd: stickyField(event.cwd, previous?.cwd), + subtitle: stickyField(event.subtitle, previous?.subtitle), }; this.descriptors.set(key, subagent); return { type: "upsert", subagent }; diff --git a/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts new file mode 100644 index 000000000..0a7759343 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/agent.background-subagent.test.ts @@ -0,0 +1,183 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import { ClaudeAgentClient } from "./agent.js"; +import { streamSession } from "../test-utils/session-stream-adapter.js"; + +/** + * A backgrounded Task subagent emits NO frames carrying parent_tool_use_id — verified against + * Claude Code 2.1.220. Everything keyed off that field therefore sees nothing for one: no + * descriptor, no timeline, and a Task card in the parent transcript with no type or task on it. + * + * The task protocol still announces the subagent, so these assert that a background child is + * visible on both surfaces from its declaration alone. + */ + +const queryFactory = vi.fn(); + +function buildQueryMock(events: unknown[]) { + let index = 0; + return { + next: vi.fn(async () => { + if (index >= events.length) return { done: true, value: undefined }; + const value = events[index]; + index += 1; + return { done: false, value }; + }), + interrupt: vi.fn(async () => undefined), + return: vi.fn(async () => undefined), + close: vi.fn(() => undefined), + setPermissionMode: vi.fn(async () => undefined), + setModel: vi.fn(async () => undefined), + supportedModels: vi.fn(async () => []), + supportedCommands: vi.fn(async () => []), + rewindFiles: vi.fn(async () => ({ canRewind: true })), + [Symbol.asyncIterator]() { + return this; + }, + }; +} + +// No frame carries parent_tool_use_id: that is the defining shape of a background subagent. +const BACKGROUND_SUBAGENT_STREAM = [ + { type: "system", subtype: "init", session_id: "bg-session", permissionMode: "default" }, + { + type: "system", + subtype: "task_started", + task_id: "aa482d02957fe96c8", + tool_use_id: "toolu_01TVF5JXom1yoZVoiaEWAoUV", + task_type: "local_agent", + subagent_type: "general-purpose", + description: "Reply with banana", + prompt: "Reply with just the word: banana", + }, + { + type: "system", + subtype: "task_updated", + task_id: "aa482d02957fe96c8", + patch: { status: "completed" }, + }, + { + type: "result", + subtype: "success", + usage: { input_tokens: 1, cache_read_input_tokens: 0, output_tokens: 1 }, + total_cost_usd: 0, + }, +]; + +async function collectUntilTerminal( + stream: AsyncGenerator, +): Promise { + const events: AgentStreamEvent[] = []; + for await (const event of stream) { + events.push(event); + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + break; + } + } + return events; +} + +describe("background Claude subagents", () => { + afterEach(() => { + queryFactory.mockReset(); + }); + + async function runStream(): Promise { + queryFactory.mockImplementation(() => buildQueryMock(BACKGROUND_SUBAGENT_STREAM)); + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }).createSession({ provider: "claude", cwd: process.cwd() }); + const events = await collectUntilTerminal(streamSession(session, "delegate work")); + await session.close(); + return events; + } + + test("appears in the subagents track despite emitting no sidechain frames", async () => { + const upserts = (await runStream()) + .filter((event) => event.type === "provider_subagent") + .map((event) => event.event) + .filter((event) => event.type === "upsert"); + + expect(upserts[0]).toMatchObject({ + id: "toolu_01TVF5JXom1yoZVoiaEWAoUV", + toolCallId: "toolu_01TVF5JXom1yoZVoiaEWAoUV", + title: "general-purpose", + description: "Reply with banana", + }); + expect(upserts.at(-1)).toMatchObject({ status: "completed" }); + }); + + test("opens the child timeline with the prompt it was given", async () => { + const firstTimelineItem = (await runStream()) + .filter((event) => event.type === "provider_subagent") + .map((event) => event.event) + .find((event) => event.type === "timeline"); + + expect(firstTimelineItem).toMatchObject({ + id: "toolu_01TVF5JXom1yoZVoiaEWAoUV", + item: { type: "user_message", text: "Reply with just the word: banana" }, + }); + }); + + test("keeps a filtered task out of the track even when it emits sidechain frames", async () => { + // A workflow child is refused at declaration, but its frames still carry a parent_tool_use_id. + // Attributing them would recreate exactly what the filter rejected: a descriptor with no + // title and a defaulted "running" status — a nameless row that never finishes. + queryFactory.mockImplementation(() => + buildQueryMock([ + { type: "system", subtype: "init", session_id: "bg-session", permissionMode: "default" }, + { + type: "system", + subtype: "task_started", + task_id: "wf-1", + tool_use_id: "toolu_workflow", + task_type: "local_workflow", + workflow_name: "spec", + description: "Run the spec workflow", + }, + { + type: "assistant", + parent_tool_use_id: "toolu_workflow", + message: { model: "claude-opus-5", content: [{ type: "text", text: "working" }] }, + }, + { type: "result", subtype: "success", usage: {}, total_cost_usd: 0 }, + ]), + ); + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }).createSession({ provider: "claude", cwd: process.cwd() }); + const events = await collectUntilTerminal(streamSession(session, "run the workflow")); + await session.close(); + + expect(events.filter((event) => event.type === "provider_subagent")).toEqual([]); + }); + + test("labels the parent's Task card with the type and task", async () => { + // Without this the card renders as a bare "Task": the tracker that normally supplies the + // sub_agent detail never runs, because no frame carries parent_tool_use_id. + const card = (await runStream()) + .filter((event) => event.type === "timeline") + .map((event) => event.item) + .find((item) => item.type === "tool_call" && item.detail?.type === "sub_agent"); + + expect(card).toMatchObject({ + type: "tool_call", + callId: "toolu_01TVF5JXom1yoZVoiaEWAoUV", + detail: { + type: "sub_agent", + subAgentType: "general-purpose", + description: "Reply with banana", + }, + }); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts new file mode 100644 index 000000000..0b7e96f73 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/agent.sub-agent-replay.test.ts @@ -0,0 +1,226 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import { ClaudeAgentClient } from "./agent.js"; +import { claudeProjectDirSync } from "./project-dir.js"; + +/** + * Replay coverage for provider subagents rebuilt from a persisted session. + * + * The live path had tests; this path did not, which is how its derivation drifted from the live + * one. These exercise the real on-disk layout Claude Code writes. + */ + +const TOOL_USE_ID = "toolu_01DgLoPMW9"; +const AGENT_ID = "a1730a6215e1f5cf6"; + +function parentEntry(content: unknown): string { + return JSON.stringify({ + type: "assistant", + sessionId: "replay-session", + timestamp: "2026-07-26T06:27:47.034Z", + message: { role: "assistant", content }, + }); +} + +function taskToolUse(): string { + return parentEntry([ + { + type: "tool_use", + id: TOOL_USE_ID, + name: "Task", + input: { subagent_type: "general-purpose", description: "Summarize the docs" }, + }, + ]); +} + +function taskToolResult(options: { isError?: boolean; mentionAgentId?: boolean } = {}): string { + return JSON.stringify({ + type: "user", + sessionId: "replay-session", + timestamp: "2026-07-26T06:28:00.000Z", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: TOOL_USE_ID, + content: options.mentionAgentId === false ? "done" : `agentId: ${AGENT_ID}\ndone`, + is_error: options.isError === true, + }, + ], + }, + }); +} + +function sidechainEntry(): string { + return JSON.stringify({ + type: "assistant", + isSidechain: true, + agentId: AGENT_ID, + sessionId: "replay-session", + timestamp: "2026-07-26T06:27:50.000Z", + message: { role: "assistant", content: [{ type: "text", text: "summary of the docs" }] }, + }); +} + +describe("ClaudeAgentSession persisted subagent replay", () => { + const logger = createTestLogger(); + const queryFactory = vi.fn(() => { + throw new Error("replay must not start a query"); + }); + let tempRoot: string; + let cwd: string; + let configDir: string; + + function writeSession(options: { + parentLines: string[]; + meta?: string | null; + sidechainLines?: string[]; + }): void { + const historyDir = claudeProjectDirSync(cwd, { configDir }); + mkdirSync(historyDir, { recursive: true }); + writeFileSync(path.join(historyDir, "replay-session.jsonl"), options.parentLines.join("\n")); + + const subagentDir = path.join(historyDir, "replay-session", "subagents"); + mkdirSync(subagentDir, { recursive: true }); + writeFileSync( + path.join(subagentDir, `agent-${AGENT_ID}.jsonl`), + (options.sidechainLines ?? [sidechainEntry()]).join("\n"), + ); + if (options.meta !== null && options.meta !== undefined) { + writeFileSync(path.join(subagentDir, `agent-${AGENT_ID}.meta.json`), options.meta); + } + } + + async function replayDescriptors(): Promise< + Extract[] + > { + const client = new ClaudeAgentClient({ + logger, + queryFactory, + resolveVersion: async () => "2.1.220", + }); + const session = await client.resumeSession( + { provider: "claude", sessionId: "replay-session" }, + { cwd }, + ); + const events: Extract[] = []; + for await (const event of session.streamHistory()) { + if (event.type === "provider_subagent") events.push(event); + } + await session.close(); + return events; + } + + function upserts(events: Extract[]) { + return events.map((e) => e.event).filter((e) => e.type === "upsert"); + } + + beforeEach(() => { + tempRoot = mkdtempSync(path.join(os.tmpdir(), "claude-subagent-replay-")); + cwd = path.join(tempRoot, "repo"); + configDir = path.join(tempRoot, "claude-config"); + mkdirSync(cwd, { recursive: true }); + vi.stubEnv("CLAUDE_CONFIG_DIR", configDir); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + rmSync(tempRoot, { recursive: true, force: true }); + }); + + test("links a subagent to its Task call through the meta sidecar", async () => { + writeSession({ + parentLines: [taskToolUse(), taskToolResult()], + meta: JSON.stringify({ + agentType: "general-purpose", + description: "Summarize the docs", + toolUseId: TOOL_USE_ID, + spawnDepth: 1, + }), + }); + + const declared = upserts(await replayDescriptors())[0]; + expect(declared).toMatchObject({ + id: TOOL_USE_ID, + toolCallId: TOOL_USE_ID, + title: "general-purpose", + description: "Summarize the docs", + }); + }); + + test("links through the meta sidecar even when the parent's tool_result is missing", async () => { + // The regression the sidecar exists to fix: with no tool_result there is nothing to scrape, + // and the subagent used to degrade to a generic title with no tool call attached. + writeSession({ + parentLines: [taskToolUse()], + meta: JSON.stringify({ toolUseId: TOOL_USE_ID, agentType: "Explore" }), + }); + + const declared = upserts(await replayDescriptors())[0]; + expect(declared).toMatchObject({ id: TOOL_USE_ID, toolCallId: TOOL_USE_ID }); + expect(declared).not.toMatchObject({ title: "Claude subagent" }); + }); + + test("falls back to the legacy scrape when no meta sidecar exists", async () => { + writeSession({ parentLines: [taskToolUse(), taskToolResult()], meta: null }); + + const events = upserts(await replayDescriptors()); + expect(events[0]).toMatchObject({ id: TOOL_USE_ID, title: "general-purpose" }); + expect(events.at(-1)).toMatchObject({ status: "completed" }); + }); + + test("carries a failed outcome through to the descriptor", async () => { + writeSession({ + parentLines: [taskToolUse(), taskToolResult({ isError: true })], + meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), + }); + + expect(upserts(await replayDescriptors()).at(-1)).toMatchObject({ status: "failed" }); + }); + + test("leaves a subagent running when the transcript records no outcome", async () => { + writeSession({ + parentLines: [taskToolUse()], + meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), + }); + + // Asserted as "nothing terminal" rather than "every upsert says running": a subtitle-only + // upsert carries no status, so presentation arriving after completion cannot revert a child. + const statuses = upserts(await replayDescriptors()).map((event) => event.status); + expect(statuses).not.toContain("completed"); + expect(statuses).not.toContain("failed"); + expect(statuses).toContain("running"); + }); + + test.each([ + ["truncated json", '{"toolUseId":"toolu_01DgLoPMW9"'], + ["not json", "toolUseId: toolu"], + ["empty file", ""], + ["a json array", "[1,2,3]"], + ])("survives a malformed meta sidecar (%s)", async (_label, contents) => { + writeSession({ parentLines: [taskToolUse(), taskToolResult()], meta: contents }); + + // Falls back to the scrape rather than losing the subagent. + const events = upserts(await replayDescriptors()); + expect(events[0]).toMatchObject({ id: TOOL_USE_ID }); + }); + + test("replays the subagent's own transcript onto its timeline", async () => { + writeSession({ + parentLines: [taskToolUse(), taskToolResult()], + meta: JSON.stringify({ toolUseId: TOOL_USE_ID }), + }); + + const timeline = (await replayDescriptors()) + .map((event) => event.event) + .filter((event) => event.type === "timeline"); + expect(timeline.length).toBeGreaterThan(0); + expect(timeline[0]).toMatchObject({ id: TOOL_USE_ID }); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.subagent-interrupt.test.ts b/packages/server/src/server/agent/providers/claude/agent.subagent-interrupt.test.ts new file mode 100644 index 000000000..d4fc0bcc0 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/agent.subagent-interrupt.test.ts @@ -0,0 +1,183 @@ +import { afterEach, describe, expect, test, vi } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import type { ProviderSubagentInputEvent } from "../../provider-subagents/store.js"; +import { ClaudeAgentClient } from "./agent.js"; + +/** + * A subagent's lifecycle when its turn is interrupted. + * + * Task ids are session-scoped, so a turn ending must not forget them: a backgrounded child settles + * after the interrupt and still needs the descriptor it was declared on. What a turn ending does + * warrant is terminalizing the children that were running in the foreground, which died with it. + */ + +const queryFactory = vi.fn(); + +/** + * A query the test feeds one message at a time and never ends, so a turn can be interrupted while + * the session stays open and keeps streaming — the shape of "background the child, interrupt, keep + * working" that the whole feature exists for. + */ +function buildOpenQueryMock() { + const pending: unknown[] = []; + let wake: (() => void) | null = null; + let finished = false; + + const settle = () => { + const resume = wake; + wake = null; + resume?.(); + }; + + const query = { + next: vi.fn(async () => { + for (;;) { + if (pending.length > 0) return { done: false, value: pending.shift() }; + if (finished) return { done: true, value: undefined }; + await new Promise((resolve) => { + wake = resolve; + }); + } + }), + interrupt: vi.fn(async () => undefined), + return: vi.fn(async () => { + finished = true; + settle(); + return undefined; + }), + close: vi.fn(() => { + finished = true; + settle(); + }), + setPermissionMode: vi.fn(async () => undefined), + setModel: vi.fn(async () => undefined), + supportedModels: vi.fn(async () => []), + supportedCommands: vi.fn(async () => []), + rewindFiles: vi.fn(async () => ({ canRewind: true })), + [Symbol.asyncIterator]() { + return this; + }, + }; + + return { + query, + push(message: unknown) { + pending.push(message); + settle(); + }, + }; +} + +const TASK_ID = "aa482d02957fe96c8"; +const SUBAGENT_ID = "toolu_01TVF5JXom1yoZVoiaEWAoUV"; +const SECOND_SUBAGENT_ID = "toolu_01SyncMarker"; + +describe("subagent lifecycle across an interrupt", () => { + afterEach(() => { + queryFactory.mockReset(); + }); + + async function startInterruptibleTurn() { + const channel = buildOpenQueryMock(); + queryFactory.mockImplementation(() => channel.query); + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }).createSession({ provider: "claude", cwd: process.cwd() }); + + const subagentEvents: ProviderSubagentInputEvent[] = []; + session.subscribe((event) => { + if (event.type === "provider_subagent") subagentEvents.push(event.event); + }); + + await session.startTurn("delegate work"); + channel.push({ + type: "system", + subtype: "init", + session_id: "bg-session", + permissionMode: "default", + }); + channel.push({ + type: "system", + subtype: "task_started", + task_id: TASK_ID, + tool_use_id: SUBAGENT_ID, + task_type: "local_agent", + subagent_type: "general-purpose", + description: "Reply with banana", + }); + await vi.waitFor(() => expect(subagentEvents.length).toBeGreaterThan(0)); + + return { channel, session, subagentEvents }; + } + + function statuses(events: ProviderSubagentInputEvent[], id: string): string[] { + return events.flatMap((event) => + event.type === "upsert" && event.id === id && event.status ? [event.status] : [], + ); + } + + test("a backgrounded child settles after the turn that spawned it is canceled", async () => { + const { channel, session, subagentEvents } = await startInterruptibleTurn(); + + channel.push({ + type: "system", + subtype: "task_updated", + task_id: TASK_ID, + patch: { is_backgrounded: true }, + }); + // Backgrounding is a silent patch — nothing observable follows from it alone. The stream is + // ordered, so declaring a second subagent behind it and waiting for that declaration proves + // the patch was consumed. + channel.push({ + type: "system", + subtype: "task_started", + task_id: "second-task", + tool_use_id: SECOND_SUBAGENT_ID, + task_type: "local_agent", + subagent_type: "general-purpose", + description: "Sync marker", + }); + await vi.waitFor(() => + expect(subagentEvents).toContainEqual(expect.objectContaining({ id: SECOND_SUBAGENT_ID })), + ); + + await session.interrupt(); + + // The child was told to outlive the turn, so cancellation says nothing about it... + channel.push({ + type: "system", + subtype: "task_notification", + task_id: TASK_ID, + tool_use_id: SUBAGENT_ID, + status: "completed", + usage: { total_tokens: 32271, tool_uses: 2, duration_ms: 10934 }, + }); + + // ...and its completion still routes, because the interrupt did not wipe the task table. + await vi.waitFor(() => + expect(subagentEvents).toContainEqual( + expect.objectContaining({ id: SUBAGENT_ID, status: "completed" }), + ), + ); + expect(statuses(subagentEvents, SUBAGENT_ID)).toEqual(["running", "completed"]); + // The sibling stayed in the foreground, so it died with the turn. + expect(statuses(subagentEvents, SECOND_SUBAGENT_ID)).toEqual(["running", "canceled"]); + + await session.close(); + }); + + test("a foreground child is canceled with the turn that spawned it", async () => { + const { session, subagentEvents } = await startInterruptibleTurn(); + + await session.interrupt(); + + await vi.waitFor(() => + expect(statuses(subagentEvents, SUBAGENT_ID)).toEqual(["running", "canceled"]), + ); + + await session.close(); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 463540825..c29f7ca27 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -41,7 +41,19 @@ import { resolveClaudeDisabledThinkingForModel, } from "./model-manifest.js"; import { parsePartialJsonObject } from "./partial-json.js"; +import { mergeClaudeHooks } from "./hooks.js"; import { ClaudeSidechainTracker } from "./sidechain-tracker.js"; +import { + ClaudeTaskProtocolSource, + type ClaudeHookObservationInput, +} from "./subagents/live-source.js"; +import { + observeReplaySubagents, + parseClaudeSubagentMeta, + type ClaudeReplayParentFacts, + type ClaudeSubagentMeta, +} from "./subagents/replay-source.js"; +import { foldSubagentObservations, type SubagentObservation } from "./subagents/observation.js"; import { buildClaudeFeatures, claudeModelSupportsFastMode } from "./feature-definitions.js"; import { buildBinaryDiagnosticRows, @@ -1983,8 +1995,15 @@ class ClaudeAgentSession implements AgentSession { private autonomousTurn: AutonomousTurnState | null = null; private readonly subscribers = new Set<(event: AgentStreamEvent) => void>(); private readonly timelineAssembler = new TimelineAssembler(); + private readonly taskProtocolSource = new ClaudeTaskProtocolSource({ + getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, + }); private readonly sidechainTracker = new ClaudeSidechainTracker({ getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, + // Releases that predate the task protocol announce nothing, so the tracker keeps deriving + // identity and status from frames for them. Detecting the capability beats comparing version + // strings: it reacts to what this session actually does. + isDescriptorOwnedElsewhere: () => this.taskProtocolSource.isActive, }); private persistedHistory: PersistedTimelineEntry[] = []; private persistedProviderSubagentEvents: Extract< @@ -2469,6 +2488,7 @@ class ClaudeAgentSession implements AgentSession { this.cancelCurrentTurn = null; this.turnState = "idle"; this.sidechainTracker.clear(); + this.taskProtocolSource.reset(); this.input?.end(); this.query?.close?.(); await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt"); @@ -3083,6 +3103,10 @@ class ClaudeAgentSession implements AgentSession { ...(effort ? { effort } : {}), ...extraClaudeOptions, ...settingsOptions, + // Merged rather than assigned above: extraClaudeOptions is spread after the base, so any + // user-configured hooks would otherwise replace these wholesale and silently stop effort + // from being observed. + hooks: mergeClaudeHooks(this.buildSubagentEffortHooks(), extraClaudeOptions?.hooks), ...(this.persistSession === undefined ? {} : { persistSession: this.persistSession }), env: sdkEnv, }; @@ -3712,10 +3736,23 @@ class ClaudeAgentSession implements AgentSession { ): AgentStreamEvent[] { const parentToolUseId = readClaudeParentToolUseId(message); if (parentToolUseId) { - return this.sidechainTracker.handleMessage(message, parentToolUseId); + return this.translateSidechainFrameToEvents(message, parentToolUseId); } const events: AgentStreamEvent[] = []; + + // Subagent identity and lifecycle are announced by Claude Code's task protocol, so they are + // read rather than inferred from sidechain frames. `task_started` precedes the child's first + // frame, so the descriptor exists before any timeline item lands on it. + const subagentObservations = this.taskProtocolSource.observe(message); + for (const event of foldSubagentObservations(subagentObservations)) { + events.push({ type: "provider_subagent", provider: "claude", event }); + } + for (const observation of subagentObservations) { + if (observation.kind !== "declared") continue; + const card = this.buildSubagentToolCallCard(observation); + if (card) events.push(card); + } if (message.type !== "system") { const sessionCapture = this.captureSessionIdFromMessage(message); if (sessionCapture.notice) { @@ -3766,6 +3803,67 @@ class ClaudeAgentSession implements AgentSession { return events; } + /** + * A frame from inside a subagent, routed to that child rather than the parent's transcript. + */ + private translateSidechainFrameToEvents( + message: SDKMessage, + parentToolUseId: string, + ): AgentStreamEvent[] { + // Once a CLI announces its tasks it announces all of them, so a frame for one that was never + // declared is work the filter already rejected — a workflow child, ambient housekeeping, a + // grandchild announced in someone else's session. Attributing it anyway materializes exactly + // what the filter prevents: a nameless descriptor stuck running, plus a timeline no surface + // can open, both held for the session's lifetime. + if ( + this.taskProtocolSource.announcesTasks && + !this.taskProtocolSource.isDeclared(parentToolUseId) + ) { + return []; + } + // The child's own frames are the only place its model appears; the task protocol does not + // announce it. Read per frame rather than snapshotting, since a provider can swap models + // mid-flight on overload or refusal fallback. + const runtimeEvents = foldSubagentObservations( + this.taskProtocolSource.observeSidechainFrame(message, parentToolUseId), + ).map((event): AgentStreamEvent => ({ type: "provider_subagent", provider: "claude", event })); + return [...runtimeEvents, ...this.sidechainTracker.handleMessage(message, parentToolUseId)]; + } + + /** + * The Task card in the parent transcript, built from the declaration. + * + * The sidechain tracker also emits this card, enriched with the child's action log, and the two + * collapse on the shared tool-call id. Emitting it at declaration matters because a + * backgrounded subagent produces no sidechain frames at all — verified on the wire — so the + * tracker never runs for one and its card would otherwise stay an unlabeled "Task". + */ + private buildSubagentToolCallCard( + declaration: Extract, + ): AgentStreamEvent | null { + const toolCall = mapClaudeRunningToolCall({ + name: "Task", + callId: declaration.id, + input: null, + output: null, + }); + if (!toolCall) return null; + return { + type: "timeline", + provider: "claude", + item: { + ...toolCall, + detail: { + type: "sub_agent", + ...(declaration.title ? { subAgentType: declaration.title } : {}), + ...(declaration.description ? { description: declaration.description } : {}), + log: "", + actions: [], + }, + }, + }; + } + private appendSidechainResultEvents(message: SDKMessage, events: AgentStreamEvent[]): void { const content = toObjectRecord(toObjectRecord(message)?.message)?.content; if (!Array.isArray(content)) return; @@ -4251,6 +4349,15 @@ class ClaudeAgentSession implements AgentSession { } this.toolUseCache.clear(); this.sidechainTracker.clear(); + // The task protocol's routing table is session-scoped, so it is deliberately NOT reset here: + // the turn ended, the session did not. Wiping it would strand every task id the session still + // holds — a backgrounded child that settles after the interrupt would find no descriptor to + // land on, and ownership would flip back to the legacy tracker mid-session. + for (const event of foldSubagentObservations( + this.taskProtocolSource.cancelRunningForegroundTasks(), + )) { + this.pushEvent({ type: "provider_subagent", provider: "claude", event }); + } } private pushToolCall( @@ -4271,6 +4378,40 @@ class ClaudeAgentSession implements AgentSession { this.notifySubscribers(event); } + /** + * Effort is reachable only through hooks. + * + * It appears nowhere on the message stream — verified by scanning every message type at depth + * — and the level Paseo requests is not necessarily the level that runs, because a model that + * does not support it is silently downgraded. A hook firing inside a subagent reports the + * active post-downgrade level alongside the subagent's `agent_id`. + * + * These are observation-only: they record what they see and always return an empty result, so + * they can never alter tool execution or turn control. + */ + private buildSubagentEffortHooks(): NonNullable { + const observe = async (input: unknown): Promise> => { + try { + for (const event of foldSubagentObservations( + this.taskProtocolSource.observeHook(input as ClaudeHookObservationInput), + )) { + this.notifySubscribers({ type: "provider_subagent", provider: "claude", event }); + } + } catch (error) { + this.logger.debug({ err: error }, "Failed to read subagent effort from hook"); + } + return {}; + }; + + // SubagentStart carries no effort (documented as absent for lifecycle hooks), so the value + // lands on the child's first tool use. SubagentStop covers a child that used no tools. + return { + PreToolUse: [{ hooks: [observe] }], + PostToolUse: [{ hooks: [observe] }], + SubagentStop: [{ hooks: [observe] }], + }; + } + private notifySubscribers(event: AgentStreamEvent): void { const turnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id; const tagged = turnId ? { ...event, turnId } : event; @@ -4341,19 +4482,39 @@ class ClaudeAgentSession implements AgentSession { } } - private ingestPersistedSidechains(parentContent: string, sidechainContents: string[]): void { + private ingestPersistedSidechains( + parentContent: string, + sidechains: ClaudeSidechainHistory, + ): void { const parentEntries = parseClaudeHistoryRecords(parentContent).filter( (entry) => entry.isSidechain !== true, ); - const sidechainEntries = [parentContent, ...sidechainContents] + const sidechainEntries = [parentContent, ...sidechains.contents] .flatMap(parseClaudeHistoryRecords) .filter((entry) => entry.isSidechain === true && typeof entry.agentId === "string"); if (sidechainEntries.length === 0) { return; } + + // Replay produces the same observations the live task protocol produces, then folds them + // with the same function, so identity and status are derived once for both paths. + const observations = observeReplaySubagents({ + subagents: [...groupClaudeSidechainEntries(sidechainEntries)].map(([agentId, entries]) => ({ + agentId, + meta: sidechains.metaByAgentId.get(agentId) ?? null, + entries, + })), + parent: readClaudeReplayParentFacts(parentEntries), + convertEntry: (entry) => this.convertHistoryEntry(entry as ClaudeHistoryEntry), + }); + this.persistedProviderSubagentEvents.push( - ...buildClaudePersistedSidechainEvents(parentEntries, sidechainEntries, (entry) => - this.convertHistoryEntry(entry), + ...foldSubagentObservations(observations).map( + (event): Extract => ({ + type: "provider_subagent", + provider: "claude", + event, + }), ), ); this.historyPending = true; @@ -5132,15 +5293,58 @@ function parseClaudeHistoryRecords(content: string): ClaudeHistoryEntry[] { return entries; } -function readClaudeSidechainHistory(historyPath: string): string[] { +/** + * Collect what the parent transcript knows about its Task calls, in the shape the shared replay + * source consumes. The agentId scrape is kept only as a fallback for sessions recorded before + * Claude Code wrote the meta sidecar. + */ +function readClaudeReplayParentFacts(parentEntries: ClaudeHistoryEntry[]): ClaudeReplayParentFacts { + const toolCalls = new Map(); + for (const [id, call] of readClaudeHistoricalSubagentToolCalls(parentEntries)) { + toolCalls.set(id, { + ...((call.name ?? call.subagentType) ? { title: call.name ?? call.subagentType } : {}), + ...(call.description ? { description: call.description } : {}), + }); + } + + // Read outcomes straight off the tool_result blocks rather than reusing the agentId scrape, + // so a subagent linked through its meta sidecar still gets a status when the scrape missed it. + const outcomesByToolCallId = new Map(); + for (const entry of parentEntries) { + const content = toObjectRecord(entry.message)?.content; + if (!Array.isArray(content)) continue; + for (const value of content) { + const block = toObjectRecord(value); + if (block?.type !== "tool_result" || typeof block.tool_use_id !== "string") continue; + if (!toolCalls.has(block.tool_use_id)) continue; + outcomesByToolCallId.set(block.tool_use_id, { failed: block.is_error === true }); + } + } + + return { + toolCalls, + linksByAgentId: readClaudeHistoricalSubagentToolResults(parentEntries), + outcomesByToolCallId, + }; +} + +interface ClaudeSidechainHistory { + contents: string[]; + /** agentId -> sidecar metadata, when Claude Code wrote one next to the transcript. */ + metaByAgentId: Map; +} + +const CLAUDE_SUBAGENT_META_FILE = /^agent-(.+)\.meta\.json$/; + +function readClaudeSidechainHistory(historyPath: string): ClaudeSidechainHistory { const sessionDirectory = path.join( path.dirname(historyPath), path.basename(historyPath, ".jsonl"), ); const sidechainDirectory = path.join(sessionDirectory, "subagents"); - if (!fs.existsSync(sidechainDirectory)) return []; + const history: ClaudeSidechainHistory = { contents: [], metaByAgentId: new Map() }; + if (!fs.existsSync(sidechainDirectory)) return history; - const contents: string[] = []; const directories = [sidechainDirectory]; while (directories.length > 0) { const directory = directories.pop(); @@ -5149,12 +5353,26 @@ function readClaudeSidechainHistory(historyPath: string): string[] { const entryPath = path.join(directory, entry.name); if (entry.isDirectory()) { directories.push(entryPath); - } else if (entry.isFile() && entry.name.endsWith(".jsonl")) { - contents.push(fs.readFileSync(entryPath, "utf8")); + continue; + } + if (!entry.isFile()) continue; + if (entry.name.endsWith(".jsonl")) { + history.contents.push(fs.readFileSync(entryPath, "utf8")); + continue; + } + // The sidecar carries the Task tool_use id, which is the same id the live stream keys on. + // Reading it is what lets replay and live agree instead of each deriving its own link. + const metaMatch = CLAUDE_SUBAGENT_META_FILE.exec(entry.name); + if (!metaMatch?.[1]) continue; + try { + const meta = parseClaudeSubagentMeta(fs.readFileSync(entryPath, "utf8")); + if (meta) history.metaByAgentId.set(metaMatch[1], meta); + } catch { + // Undocumented internals: a missing or unreadable sidecar must never fail ingestion. } } } - return contents; + return history; } interface ClaudeHistoricalSubagentToolCall { @@ -5224,89 +5442,6 @@ function groupClaudeSidechainEntries( return entriesByAgentId; } -function buildClaudePersistedSidechainEvents( - parentEntries: ClaudeHistoryEntry[], - sidechainEntries: ClaudeHistoryEntry[], - convertEntry: (entry: ClaudeHistoryEntry) => AgentTimelineItem[], -): Extract[] { - const events: Extract[] = []; - const toolCalls = readClaudeHistoricalSubagentToolCalls(parentEntries); - const toolResults = readClaudeHistoricalSubagentToolResults(parentEntries); - for (const [agentId, entries] of groupClaudeSidechainEntries(sidechainEntries)) { - events.push( - ...buildClaudePersistedSidechainAgentEvents( - agentId, - entries, - toolCalls, - toolResults, - convertEntry, - ), - ); - } - return events; -} - -function resolveClaudeHistoricalSubagentTitle( - toolCall: ClaudeHistoricalSubagentToolCall | undefined, -): string { - return toolCall?.name ?? toolCall?.subagentType ?? "Claude subagent"; -} - -function buildClaudePersistedSidechainAgentEvents( - agentId: string, - entries: ClaudeHistoryEntry[], - toolCalls: ReadonlyMap, - toolResults: ReadonlyMap, - convertEntry: (entry: ClaudeHistoryEntry) => AgentTimelineItem[], -): Extract[] { - const result = toolResults.get(agentId); - const id = result?.toolCallId ?? agentId; - const toolCall = result ? toolCalls.get(result.toolCallId) : undefined; - const firstTimestamp = normalizeProviderReplayTimestamp(entries[0]?.timestamp); - const events: Extract[] = [ - { - type: "provider_subagent", - provider: "claude", - event: { - type: "upsert", - id, - title: resolveClaudeHistoricalSubagentTitle(toolCall), - description: toolCall?.description ?? null, - status: "running", - toolCallId: result?.toolCallId ?? null, - ...(firstTimestamp ? { timestamp: firstTimestamp } : {}), - }, - }, - ]; - for (const entry of entries) { - const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); - for (const item of convertEntry(entry)) { - events.push({ - type: "provider_subagent", - provider: "claude", - event: { - type: "timeline", - id, - item, - ...(timestamp ? { timestamp } : {}), - }, - }); - } - } - const lastTimestamp = normalizeProviderReplayTimestamp(entries.at(-1)?.timestamp); - events.push({ - type: "provider_subagent", - provider: "claude", - event: { - type: "upsert", - id, - status: result?.failed ? "failed" : "completed", - ...(lastTimestamp ? { timestamp: lastTimestamp } : {}), - }, - }); - return events; -} - interface ClaudeHistoryEntry { type?: unknown; subtype?: unknown; diff --git a/packages/server/src/server/agent/providers/claude/hooks.test.ts b/packages/server/src/server/agent/providers/claude/hooks.test.ts new file mode 100644 index 000000000..9bb4915af --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/hooks.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it, vi } from "vitest"; + +import { mergeClaudeHooks } from "./hooks.js"; + +describe("mergeClaudeHooks", () => { + const own = { PreToolUse: [{ hooks: [vi.fn()] }] } as never; + + it("keeps Paseo's hooks when the user configured none", () => { + expect(mergeClaudeHooks(own, undefined)).toEqual(own); + expect(mergeClaudeHooks(own, null)).toEqual(own); + }); + + it("appends user hooks for the same event instead of replacing them", () => { + const userHook = vi.fn(); + const merged = mergeClaudeHooks(own, { PreToolUse: [{ hooks: [userHook] }] }) as { + PreToolUse: unknown[]; + }; + // Both must survive: dropping Paseo's would silently stop effort being observed, and + // dropping the user's would break their configuration. + expect(merged.PreToolUse).toHaveLength(2); + }); + + it("carries through events Paseo does not register", () => { + const merged = mergeClaudeHooks(own, { SessionStart: [{ hooks: [vi.fn()] }] }) as Record< + string, + unknown[] + >; + expect(merged.SessionStart).toHaveLength(1); + expect(merged.PreToolUse).toHaveLength(1); + }); + + it("ignores malformed entries rather than throwing", () => { + expect(mergeClaudeHooks(own, { PreToolUse: "not-an-array" })).toEqual(own); + expect(mergeClaudeHooks(own, "nonsense")).toEqual(own); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/hooks.ts b/packages/server/src/server/agent/providers/claude/hooks.ts new file mode 100644 index 000000000..5ccc2f344 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/hooks.ts @@ -0,0 +1,21 @@ +import type { ClaudeOptions } from "./query.js"; + +type ClaudeHooks = NonNullable; + +/** + * Combine Paseo's observation hooks with any the user configured, per event. + * + * Neither side wins: Claude Code runs every matcher registered for an event, so appending keeps + * a user's hooks working while Paseo keeps observing. Assigning either one would silently drop + * the other, and the failure would be invisible — effort would simply stop appearing. + */ +export function mergeClaudeHooks(own: ClaudeHooks, extra: unknown): ClaudeHooks { + if (!extra || typeof extra !== "object") return own; + const merged: ClaudeHooks = { ...own }; + for (const [event, matchers] of Object.entries(extra as Record)) { + if (!Array.isArray(matchers)) continue; + const key = event as keyof ClaudeHooks; + merged[key] = [...(merged[key] ?? []), ...matchers] as ClaudeHooks[typeof key]; + } + return merged; +} diff --git a/packages/server/src/server/agent/providers/claude/models.ts b/packages/server/src/server/agent/providers/claude/models.ts index 48bf90088..51601ecb9 100644 --- a/packages/server/src/server/agent/providers/claude/models.ts +++ b/packages/server/src/server/agent/providers/claude/models.ts @@ -132,3 +132,32 @@ function isRecord(value: unknown): value is Record { export function normalizeClaudeRuntimeModelId(value: string | null | undefined): string | null { return normalizeClaudeManifestRuntimeModelId(value); } + +/** + * Placeholder model values Claude Code writes on frames with no real inference behind them. + * These are not models and must never be displayed. + */ +const CLAUDE_PLACEHOLDER_MODEL_IDS = new Set([""]); + +/** + * Resolve a model id observed on a Claude assistant frame, for display. + * + * Prefers the manifest-normalized id so equivalent spellings collapse (a dated alias and a + * gateway prefix are the same model), but falls back to the raw string when the manifest does + * not know it. The fallback matters: Claude Code is an Anthropic-compatible client, so subagents + * routinely report models that are not Anthropic's — Z.AI GLM ids via `ANTHROPIC_BASE_URL` + * (docs/custom-providers.md) among them. Manifest-only resolution would blank the model for + * exactly those users. + * + * Note a `[1m]` suffix is preserved where it names its own manifest entry: a 1M-context variant + * is a distinct model with a distinct context window, not a spelling of the 200K one. + * + * Returns null for placeholders and empty values, meaning "not observed". + */ +export function resolveObservedClaudeModelId(value: string | null | undefined): string | null { + const trimmed = typeof value === "string" ? value.trim() : ""; + if (!trimmed || CLAUDE_PLACEHOLDER_MODEL_IDS.has(trimmed)) { + return null; + } + return normalizeClaudeManifestRuntimeModelId(trimmed) ?? trimmed; +} diff --git a/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts index d5257e3d1..131d8ee12 100644 --- a/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts +++ b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts @@ -58,9 +58,19 @@ function isClaudeContentChunk(value: unknown): value is ClaudeContentChunk { export class ClaudeSidechainTracker { private readonly activeSidechains = new Map(); private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + private readonly isDescriptorOwnedElsewhere: () => boolean; - constructor(input: { getToolInput: (toolUseId: string) => AgentMetadata | null | undefined }) { + constructor(input: { + getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + /** + * When the provider learns subagent identity and status declaratively, this tracker stops + * emitting descriptor upserts and maps frames to timeline items only. Two owners writing the + * same descriptor from different evidence is how the live and replay paths drifted apart. + */ + isDescriptorOwnedElsewhere?: () => boolean; + }) { this.getToolInput = input.getToolInput; + this.isDescriptorOwnedElsewhere = input.isDescriptorOwnedElsewhere ?? (() => false); } handleMessage(message: SDKMessage, parentToolUseId: string): AgentStreamEvent[] { @@ -124,19 +134,25 @@ export class ClaudeSidechainTracker { actions: [], }; + const descriptorEvents: AgentStreamEvent[] = this.isDescriptorOwnedElsewhere() + ? [] + : [ + { + type: "provider_subagent", + provider: "claude", + event: { + type: "upsert", + id: parentToolUseId, + title: state.name ?? state.subAgentType ?? "Claude subagent", + description: state.description ?? null, + status: "running", + toolCallId: parentToolUseId, + }, + }, + ]; + return [ - { - type: "provider_subagent", - provider: "claude", - event: { - type: "upsert", - id: parentToolUseId, - title: state.name ?? state.subAgentType ?? "Claude subagent", - description: state.description ?? null, - status: "running", - toolCallId: parentToolUseId, - }, - }, + ...descriptorEvents, ...childTimelineItems.map( (item): AgentStreamEvent => ({ type: "provider_subagent", @@ -157,6 +173,10 @@ export class ClaudeSidechainTracker { finishAll(status: "completed" | "failed" | "canceled"): AgentStreamEvent[] { const events: AgentStreamEvent[] = []; + if (this.isDescriptorOwnedElsewhere()) { + this.activeSidechains.clear(); + return events; + } for (const [id, state] of this.activeSidechains) { events.push({ type: "provider_subagent", @@ -179,6 +199,7 @@ export class ClaudeSidechainTracker { const state = this.activeSidechains.get(id); if (!state) return []; this.activeSidechains.delete(id); + if (this.isDescriptorOwnedElsewhere()) return []; return [ { type: "provider_subagent", diff --git a/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts new file mode 100644 index 000000000..65b118e98 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/live-source.test.ts @@ -0,0 +1,489 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import { describe, expect, it } from "vitest"; + +import { ClaudeTaskProtocolSource } from "./live-source.js"; + +// Shapes copied from real wire output (Claude Code 2.1.220) rather than the type declarations, +// so the fixtures stay honest about what actually arrives. +function taskStarted(overrides: Record = {}): SDKMessage { + return { + type: "system", + subtype: "task_started", + task_id: "a1730a6215e1f5cf6", + tool_use_id: "toolu_01DgLoPMW9", + description: "Summarize hover and unistyles docs", + subagent_type: "general-purpose", + task_type: "local_agent", + ...overrides, + } as unknown as SDKMessage; +} + +function taskUpdated(status: string, taskId = "a1730a6215e1f5cf6"): SDKMessage { + return { + type: "system", + subtype: "task_updated", + task_id: taskId, + patch: { status, end_time: 1785117455676 }, + } as unknown as SDKMessage; +} + +function taskNotification(status: string, taskId = "a1730a6215e1f5cf6"): SDKMessage { + return { + type: "system", + subtype: "task_notification", + task_id: taskId, + tool_use_id: "toolu_01DgLoPMW9", + status, + output_file: "/tmp/x.output", + summary: "done", + } as unknown as SDKMessage; +} + +describe("ClaudeTaskProtocolSource", () => { + it("declares a subagent from its announcement, keyed by the Task tool_use id", () => { + const source = new ClaudeTaskProtocolSource(); + + expect(source.observe(taskStarted())).toEqual([ + { + kind: "declared", + id: "toolu_01DgLoPMW9", + toolCallId: "toolu_01DgLoPMW9", + title: "general-purpose", + description: "Summarize hover and unistyles docs", + }, + ]); + }); + + it("prefers the name the Task call gave over the agent type", () => { + // The replay source titles a subagent `input.name ?? subagent_type`. Live reads the same + // field from the same Task call, so one subagent is named identically on both paths. + const source = new ClaudeTaskProtocolSource({ + getToolInput: (toolUseId) => + toolUseId === "toolu_01DgLoPMW9" ? { name: "docs-scout", subagent_type: "Explore" } : null, + }); + + expect(source.observe(taskStarted({ subagent_type: "Explore" }))[0]).toMatchObject({ + kind: "declared", + title: "docs-scout", + }); + }); + + it("falls back to the agent type when the Task call named nothing", () => { + const source = new ClaudeTaskProtocolSource({ getToolInput: () => ({ description: "x" }) }); + expect(source.observe(taskStarted())[0]).toMatchObject({ title: "general-purpose" }); + }); + + it("routes a status update through the task_id it learned at declaration", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + + // task_updated carries no tool_use_id — the mapping is the only way to route it. + expect(source.observe(taskUpdated("completed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + }); + + it("does not re-announce a status that already holds", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe(taskUpdated("completed")); + + // task_updated and task_notification arrive in the same millisecond on the real wire. + expect(source.observe(taskNotification("completed"))).toEqual([]); + }); + + it("treats a killed or stopped task as canceled, not failed", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observe(taskUpdated("killed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "canceled" }, + ]); + }); + + it("keeps a paused task running, since the descriptor has no paused state", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe(taskUpdated("completed")); + expect(source.observe(taskUpdated("paused"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "running" }, + ]); + }); + + it("ignores tasks that carry no tool_use id", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskStarted({ tool_use_id: undefined }))).toEqual([]); + expect(source.isActive).toBe(false); + }); + + it("ignores a backgrounded shell command", () => { + const source = new ClaudeTaskProtocolSource(); + // Real wire shape: a backgrounded Bash announces with a tool_use_id but no subagent_type, + // so filtering on the id alone would put `sleep 20` in the subagents track. + expect( + source.observe( + taskStarted({ + task_id: "b51skux0z", + tool_use_id: "toolu_01MgVdcGPYnqE8cJQuccFtkU", + task_type: "local_bash", + subagent_type: undefined, + description: "Sleep 20 seconds in background", + }), + ), + ).toEqual([]); + expect(source.isActive).toBe(false); + }); + + it("ignores a workflow run", () => { + const source = new ClaudeTaskProtocolSource(); + expect( + source.observe(taskStarted({ task_type: "local_workflow", subagent_type: undefined })), + ).toEqual([]); + }); + + it("accepts a subagent from a release that predates task_type", () => { + const source = new ClaudeTaskProtocolSource(); + const observations = source.observe(taskStarted({ task_type: undefined })); + expect(observations[0]).toMatchObject({ kind: "declared", title: "general-purpose" }); + }); + + it("ignores an untyped task with no subagent type", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskStarted({ task_type: undefined, subagent_type: undefined }))).toEqual( + [], + ); + }); + + it("opens the child timeline with the prompt it was given", () => { + const source = new ClaudeTaskProtocolSource(); + const observations = source.observe( + taskStarted({ prompt: "Reply with just the word: banana" }), + ); + + expect(observations[1]).toEqual({ + kind: "timeline", + id: "toolu_01DgLoPMW9", + item: { type: "user_message", text: "Reply with just the word: banana" }, + }); + }); + + it("declares without a timeline item when no prompt was announced", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskStarted())).toHaveLength(1); + }); + + it("ignores ambient housekeeping tasks", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskStarted({ skip_transcript: true }))).toEqual([]); + expect(source.isActive).toBe(false); + }); + + it("routes concurrent subagents independently", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe( + taskStarted({ + task_id: "a88cdd22e38fd1dcf", + tool_use_id: "toolu_01GMd4ob4m", + description: "Summarize forms and design docs", + }), + ); + + expect(source.observe(taskUpdated("completed", "a88cdd22e38fd1dcf"))).toEqual([ + { kind: "status", id: "toolu_01GMd4ob4m", status: "completed" }, + ]); + expect(source.observe(taskUpdated("failed", "a1730a6215e1f5cf6"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "failed" }, + ]); + }); + + it("drops a status for a task it never saw declared", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskUpdated("completed", "unknown-task"))).toEqual([]); + }); + + it("does not readmit a filtered task through its notification", () => { + const source = new ClaudeTaskProtocolSource(); + // A backgrounded shell is filtered at declaration, but it still gets a task_notification + // carrying a tool_use_id. Routing status off that id would recreate the descriptor with a + // status and no identity — a nameless row in the subagents track. + source.observe( + taskStarted({ + task_id: "b51skux0z", + tool_use_id: "toolu_01MgVdcGPYnqE8cJQuccFtkU", + task_type: "local_bash", + subagent_type: undefined, + }), + ); + + expect(source.observe(taskNotification("completed", "b51skux0z"))).toEqual([]); + }); + + it("drops a notification for a task it never declared", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskNotification("failed", "unknown-task"))).toEqual([]); + }); + + it("reports inactivity until a task is announced, for older CLIs", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.isActive).toBe(false); + source.observe(taskStarted()); + expect(source.isActive).toBe(true); + source.reset(); + expect(source.isActive).toBe(false); + }); + + it("cancels the foreground subagents a canceled turn was running", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + + expect(source.cancelRunningForegroundTasks()).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "canceled" }, + ]); + // Idempotent: the child is no longer running, so a second cancellation says nothing. + expect(source.cancelRunningForegroundTasks()).toEqual([]); + }); + + it("leaves an already-settled subagent alone when the turn is canceled", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe(taskUpdated("completed")); + + expect(source.cancelRunningForegroundTasks()).toEqual([]); + }); + + it("does not cancel a backgrounded subagent, which outlives the turn", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe({ + type: "system", + subtype: "task_updated", + task_id: "a1730a6215e1f5cf6", + patch: { is_backgrounded: true }, + } as unknown as SDKMessage); + + expect(source.cancelRunningForegroundTasks()).toEqual([]); + }); + + it("still routes a backgrounded subagent that settles after the interrupt", () => { + // The headline case: interrupt, continue, and the child that was told to outlive the turn + // reports completion later. Wiping the routing table on cancel drops this on the floor and + // leaves the row spinning forever. + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observe({ + type: "system", + subtype: "task_updated", + task_id: "a1730a6215e1f5cf6", + patch: { is_backgrounded: true }, + } as unknown as SDKMessage); + source.cancelRunningForegroundTasks(); + + expect(source.observe(taskNotification("completed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + expect(source.isActive).toBe(true); + }); + + it("lets a late notification correct a cancellation it guessed wrong", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.cancelRunningForegroundTasks(); + + expect(source.observe(taskNotification("completed"))).toEqual([ + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + }); + + it("ignores unrelated system messages", () => { + const source = new ClaudeTaskProtocolSource(); + expect( + source.observe({ type: "system", subtype: "init", model: "opus" } as unknown as SDKMessage), + ).toEqual([]); + }); +}); + +describe("ClaudeTaskProtocolSource usage and runtime", () => { + function taskProgress(usage: Record, taskId = "a1730a6215e1f5cf6"): SDKMessage { + return { + type: "system", + subtype: "task_progress", + task_id: taskId, + usage, + last_tool_name: "Read", + } as unknown as SDKMessage; + } + + function assistantFrame(model: string): SDKMessage { + return { type: "assistant", message: { model, content: [] } } as unknown as SDKMessage; + } + + it("reports usage as the child works", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + // Real shape: task_progress fires once per tool use, not on a timer. + expect( + source.observe(taskProgress({ total_tokens: 16484, tool_uses: 1, duration_ms: 2474 })), + ).toEqual([ + { + kind: "subtitle", + id: "toolu_01DgLoPMW9", + subtitle: "general-purpose · 16.5k tokens", + }, + ]); + }); + + it("reports terminal usage and status together", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + const observations = source.observe({ + type: "system", + subtype: "task_notification", + task_id: "a1730a6215e1f5cf6", + tool_use_id: "toolu_01DgLoPMW9", + status: "completed", + usage: { total_tokens: 32271, tool_uses: 2, duration_ms: 10934 }, + } as unknown as SDKMessage); + + expect(observations).toEqual([ + { + kind: "subtitle", + id: "toolu_01DgLoPMW9", + subtitle: "general-purpose · 32.3k tokens", + }, + { kind: "status", id: "toolu_01DgLoPMW9", status: "completed" }, + ]); + }); + + it("ignores usage for a task it never declared", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observe(taskProgress({ total_tokens: 1 }, "b51skux0z"))).toEqual([]); + }); + + it("reads the model off the child's own frames", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect( + source.observeSidechainFrame(assistantFrame("claude-opus-5"), "toolu_01DgLoPMW9"), + ).toEqual([{ kind: "subtitle", id: "toolu_01DgLoPMW9", subtitle: "general-purpose · Opus 5" }]); + }); + + it("re-reports only when the model actually changes", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observeSidechainFrame(assistantFrame("claude-opus-5"), "toolu_01DgLoPMW9"); + // A dated alias is the same model; a mid-flight fallback is not. + expect( + source.observeSidechainFrame(assistantFrame("claude-opus-5-20260724"), "toolu_01DgLoPMW9"), + ).toEqual([]); + expect( + source.observeSidechainFrame(assistantFrame("claude-sonnet-5"), "toolu_01DgLoPMW9"), + ).toEqual([ + { kind: "subtitle", id: "toolu_01DgLoPMW9", subtitle: "general-purpose · Sonnet 5" }, + ]); + }); + + it("never reports a placeholder model", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observeSidechainFrame(assistantFrame(""), "toolu_01DgLoPMW9")).toEqual( + [], + ); + }); + + it("ignores a frame from a task it never declared", () => { + // A filtered task still emits frames carrying its tool_use id. Reporting a model for one + // would create a descriptor the declaration filter deliberately refused — with no title and + // a defaulted "running" status, which the track renders as a nameless row that never ends. + const source = new ClaudeTaskProtocolSource(); + source.observe( + taskStarted({ + task_id: "b51skux0z", + tool_use_id: "toolu_01MgVdcGPYnqE8cJQuccFtkU", + task_type: "local_bash", + subagent_type: undefined, + }), + ); + + expect( + source.observeSidechainFrame( + assistantFrame("claude-opus-5"), + "toolu_01MgVdcGPYnqE8cJQuccFtkU", + ), + ).toEqual([]); + expect(source.isDeclared("toolu_01MgVdcGPYnqE8cJQuccFtkU")).toBe(false); + }); + + it("never reports effort, which the live stream does not carry", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + const all = [ + ...source.observe(taskProgress({ total_tokens: 1 })), + ...source.observeSidechainFrame(assistantFrame("claude-opus-5"), "toolu_01DgLoPMW9"), + ]; + expect(all.some((o) => "effort" in o && o.effort !== undefined)).toBe(false); + }); +}); + +describe("ClaudeTaskProtocolSource effort from hooks", () => { + // Real hook shapes (Claude Code 2.1.220): agent_id equals task_started's task_id, and + // effort.level is the ACTIVE level after any silent downgrade. + function hook(overrides: Record = {}) { + return { + agent_id: "a1730a6215e1f5cf6", + agent_type: "general-purpose", + effort: { level: "high" }, + ...overrides, + }; + } + + it("reports effort for a declared subagent, routed by agent_id", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observeHook(hook())).toEqual([ + { kind: "subtitle", id: "toolu_01DgLoPMW9", subtitle: "general-purpose · High" }, + ]); + }); + + it("does not re-report an unchanged effort", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observeHook(hook()); + // PreToolUse and PostToolUse both fire on every tool use. + expect(source.observeHook(hook())).toEqual([]); + }); + + it("reports a downgrade when the level actually changes", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + source.observeHook(hook()); + expect(source.observeHook(hook({ effort: { level: "medium" } }))).toEqual([ + { kind: "subtitle", id: "toolu_01DgLoPMW9", subtitle: "general-purpose · Medium" }, + ]); + }); + + it("ignores a hook from the parent turn", () => { + // Stop fires with effort but no agent_id: that is the parent, not a child. + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observeHook(hook({ agent_id: undefined }))).toEqual([]); + }); + + it("ignores a hook for a subagent it never declared", () => { + const source = new ClaudeTaskProtocolSource(); + expect(source.observeHook(hook())).toEqual([]); + }); + + it("ignores a hook that carries no effort", () => { + // SubagentStart fires without effort — documented as absent for lifecycle hooks. + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observeHook(hook({ effort: undefined }))).toEqual([]); + }); + + it("survives a hook input of an unexpected shape", () => { + const source = new ClaudeTaskProtocolSource(); + source.observe(taskStarted()); + expect(source.observeHook({} as never)).toEqual([]); + expect(source.observeHook({ agent_id: 42, effort: { level: 7 } } as never)).toEqual([]); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/live-source.ts b/packages/server/src/server/agent/providers/claude/subagents/live-source.ts new file mode 100644 index 000000000..c2b199632 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/live-source.ts @@ -0,0 +1,395 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; + +import type { AgentMetadata } from "../../../agent-sdk-types.js"; +import type { ProviderSubagentStatus } from "../../../provider-subagents/store.js"; +import { resolveObservedClaudeModelId } from "../models.js"; +import type { SubagentObservation } from "./observation.js"; +import { + buildClaudeSubagentSubtitle, + type ClaudeSubagentPresentationFacts, + type ClaudeSubagentUsage, +} from "./presentation.js"; + +/** + * Claude Code announces subagent lifecycle on the SDK stream. This reads those announcements + * instead of reconstructing them from sidechain frames. + * + * Verified on the wire (Claude Code 2.1.220, Paseo's own query options): + * + * task_started task_id, tool_use_id, description, subagent_type, task_type + * task_updated task_id, patch.status, patch.is_backgrounded + * task_notification task_id, tool_use_id, status + * + * Only `task_started` carries `tool_use_id`, so the mapping from task id to the canonical + * subagent id has to be remembered. The source also accumulates Claude's presentation facts so + * the shared descriptor receives one complete provider-owned subtitle rather than Claude fields. + * Neither is a lifecycle state machine: status comes directly from task announcements. + * + * The table is session-scoped, because task ids are. It survives a turn ending — the one thing a + * turn ending warrants is `cancelRunningForegroundTasks`, not forgetting the session. + */ + +interface TaskStartedMessage { + task_id: string; + tool_use_id?: string; + description?: string; + subagent_type?: string; + task_type?: string; + prompt?: string; + skip_transcript?: boolean; +} + +/** Task-tool subagents. Backgrounded shell commands announce as `local_bash`. */ +const CLAUDE_SUBAGENT_TASK_TYPE = "local_agent"; + +/** + * Not every announced task is a subagent. Verified on the wire: + * + * Task subagent task_type "local_agent", subagent_type "general-purpose" + * background shell task_type "local_bash", no subagent_type + * + * Both carry a `tool_use_id`, so presence of an id is not a discriminator — filtering on it + * alone puts `sleep 20` in the subagents track. Releases that predate `task_type` are covered + * by requiring a subagent type instead. + */ +function isSubagentTask(message: TaskStartedMessage): boolean { + if (message.task_type) return message.task_type === CLAUDE_SUBAGENT_TASK_TYPE; + return readString(message.subagent_type) !== undefined; +} + +interface TaskUpdatedMessage { + task_id: string; + /** + * `is_backgrounded` is declared on `SDKTaskUpdatedMessage["patch"]`: it flips when a foreground + * task is backgrounded, which is the only signal that separates a child that dies with its turn + * from one that was explicitly told to outlive it. + */ + patch?: { status?: string; is_backgrounded?: boolean }; +} + +interface TaskNotificationMessage { + task_id: string; + tool_use_id?: string; + status?: string; + usage?: { total_tokens?: number; tool_uses?: number; duration_ms?: number }; +} + +/** + * The subset of a hook input this source reads. Every hook shares these base fields: `agent_id` + * is set when the hook fires from inside a subagent, and `effort.level` is the ACTIVE level for + * that turn, after any silent downgrade for the selected model. + */ +export interface ClaudeHookObservationInput { + agent_id?: unknown; + agent_type?: unknown; + effort?: { level?: unknown } | undefined; +} + +interface TaskProgressMessage { + task_id: string; + usage?: { total_tokens?: number; tool_uses?: number; duration_ms?: number }; +} + +function readUsage( + usage: { total_tokens?: number; tool_uses?: number; duration_ms?: number } | undefined, +): ClaudeSubagentUsage | null { + if (!usage) return null; + const observed: ClaudeSubagentUsage = {}; + if (typeof usage.total_tokens === "number") observed.totalTokens = usage.total_tokens; + return Object.keys(observed).length > 0 ? observed : null; +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined; +} + +/** + * The descriptor has no paused or pending state, and neither is terminal, so both read as + * running. `killed` is a cancellation, not a failure — the child was stopped, it did not error. + */ +function mapTaskStatus(status: string | undefined): ProviderSubagentStatus | undefined { + switch (status) { + case "pending": + case "running": + case "paused": + return "running"; + case "completed": + return "completed"; + case "failed": + return "failed"; + case "killed": + case "stopped": + return "canceled"; + default: + return undefined; + } +} + +export interface ClaudeTaskProtocolSourceInput { + /** + * The parent's Task tool input, by tool_use id. + * + * `task_started` announces `subagent_type`, but the Task call itself may also carry an explicit + * `name`, and the replay source prefers that name over the type. Reading the same field from the + * same place is what keeps one subagent titled identically live and on reopen. + */ + getToolInput?: (toolUseId: string) => AgentMetadata | null | undefined; +} + +export class ClaudeTaskProtocolSource { + /** task_id -> canonical subagent id (the Task tool_use id). Populated by task_started. */ + private readonly subagentIdByTaskId = new Map(); + /** + * Every subagent id this source declared. It is the source's whole vocabulary: an id that is + * not in here was either filtered at declaration or never announced, and this source has + * nothing to say about it. + */ + private readonly declaredIds = new Set(); + /** + * Declared subagents that were moved to the background. They outlive the turn that spawned + * them, so a turn ending is not evidence that they stopped. + */ + private readonly backgroundedIds = new Set(); + /** Last status emitted per subagent, so a redundant announcement is not re-broadcast. */ + private readonly lastStatusById = new Map(); + /** Claude facts stay inside the provider boundary; clients receive one compact subtitle. */ + private readonly presentationById = new Map(); + private readonly lastSubtitleById = new Map(); + private sawTaskStarted = false; + private sawAnyTask = false; + private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + + constructor(input: ClaudeTaskProtocolSourceInput = {}) { + this.getToolInput = input.getToolInput ?? (() => null); + } + + /** + * True once this session has announced at least one task. Older Claude Code releases predate + * the task protocol; callers use this to decide whether the legacy derivation still has work + * to do, rather than assuming a version. + */ + get isActive(): boolean { + return this.sawTaskStarted; + } + + /** + * True once this session announced any task at all, including the ones filtered out. + * + * Deliberately not `isActive`: that one answers "does a declarative owner exist for subagent + * descriptors", and stays false in a session whose only tasks were rejected. This one answers + * "does this CLI announce its tasks" — and once it does, every task is announced, so a frame + * for an id that was never declared is provably work the filter refused rather than something + * this source has yet to hear about. + */ + get announcesTasks(): boolean { + return this.sawAnyTask; + } + + /** + * Whether this source declared the given subagent. Callers route frames through this before + * attributing anything to an id: a frame for a task that was never declared belongs to work + * the declaration filter already rejected. + */ + isDeclared(subagentId: string): boolean { + return this.declaredIds.has(subagentId); + } + + observe(message: SDKMessage): SubagentObservation[] { + if (message.type !== "system") return []; + switch (message.subtype) { + case "task_started": + return this.observeTaskStarted(message as unknown as TaskStartedMessage); + case "task_updated": + return this.observeTaskUpdated(message as unknown as TaskUpdatedMessage); + case "task_notification": + return this.observeTaskNotification(message as unknown as TaskNotificationMessage); + case "task_progress": + return this.observeTaskProgress(message as unknown as TaskProgressMessage); + default: + return []; + } + } + + /** + * Forget the session. + * + * Task ids are session-scoped and outlive any single turn, so this belongs to session teardown + * alone. Clearing it when a turn ends would drop the routing table while its tasks were still + * live — see `cancelRunningForegroundTasks`, which is what a turn ending actually warrants. + */ + reset(): void { + this.subagentIdByTaskId.clear(); + this.declaredIds.clear(); + this.backgroundedIds.clear(); + this.lastStatusById.clear(); + this.presentationById.clear(); + this.lastSubtitleById.clear(); + this.sawTaskStarted = false; + this.sawAnyTask = false; + } + + /** + * Terminalize the subagents that a canceled turn was running. + * + * Cancellation is a fact about the turn, not about the task table: the session, and every id in + * it, outlives the turn. So this reports statuses and leaves the routing intact — that is what + * lets a child which settles after the interrupt still find its descriptor, and what keeps a + * later `task_notification` free to correct this guess. + * + * Backgrounded children are skipped. Being backgrounded is precisely the declaration that they + * outlive the turn; everything else was running in the foreground and died with it. + */ + cancelRunningForegroundTasks(): SubagentObservation[] { + const observations: SubagentObservation[] = []; + for (const id of this.declaredIds) { + if (this.backgroundedIds.has(id)) continue; + if (this.lastStatusById.get(id) !== "running") continue; + this.lastStatusById.set(id, "canceled"); + observations.push({ kind: "status", id, status: "canceled" }); + } + return observations; + } + + private observeTaskStarted(message: TaskStartedMessage): SubagentObservation[] { + // Recorded before the filter: what this proves is that the CLI announces its tasks, which is + // true whether or not this particular one is a subagent. + this.sawAnyTask = true; + + const id = readString(message.tool_use_id); + // skip_transcript marks ambient housekeeping the transcript should not show. + if (!id || message.skip_transcript === true || !isSubagentTask(message)) return []; + + this.sawTaskStarted = true; + this.subagentIdByTaskId.set(message.task_id, id); + this.declaredIds.add(id); + this.lastStatusById.set(id, "running"); + + // An explicit `name` on the Task call wins over the agent type, matching how replay titles the + // same subagent. Without it a fan-out of five Explores reads as five identical rows. + const title = readString(this.getToolInput(id)?.name) ?? readString(message.subagent_type); + const description = readString(message.description); + const observations: SubagentObservation[] = [ + { + kind: "declared", + id, + toolCallId: id, + ...(title ? { title } : {}), + ...(description ? { description } : {}), + }, + ]; + const initialPresentation = title ? { title } : {}; + this.presentationById.set(id, initialPresentation); + const initialSubtitle = buildClaudeSubagentSubtitle(initialPresentation); + if (initialSubtitle) this.lastSubtitleById.set(id, initialSubtitle); + + // Open the child's timeline with the prompt it was actually given. Without this the pane + // starts mid-conversation, showing replies to a question the reader never sees. + const prompt = readString(message.prompt); + if (prompt) { + observations.push({ kind: "timeline", id, item: { type: "user_message", text: prompt } }); + } + return observations; + } + + private observeTaskUpdated(message: TaskUpdatedMessage): SubagentObservation[] { + const id = this.subagentIdByTaskId.get(message.task_id); + const backgrounded = message.patch?.is_backgrounded; + if (id && typeof backgrounded === "boolean") { + if (backgrounded) this.backgroundedIds.add(id); + else this.backgroundedIds.delete(id); + } + return this.observeStatus(message.task_id, message.patch?.status); + } + + private observeTaskNotification(message: TaskNotificationMessage): SubagentObservation[] { + const observations = this.observeUsage(message.task_id, message.usage); + observations.push(...this.observeStatus(message.task_id, message.status)); + return observations; + } + + /** + * `task_progress` fires once per tool use, not on a timer, so this is an accurate live cost + * signal rather than a sampled one. It matters most for backgrounded subagents, which emit no + * sidechain frames at all and would otherwise show no activity while they work. + */ + private observeTaskProgress(message: TaskProgressMessage): SubagentObservation[] { + return this.observeUsage(message.task_id, message.usage); + } + + private observeUsage( + taskId: string, + raw: { total_tokens?: number; tool_uses?: number; duration_ms?: number } | undefined, + ): SubagentObservation[] { + const id = this.subagentIdByTaskId.get(taskId); + const usage = readUsage(raw); + if (!id || !usage) return []; + return this.updatePresentation(id, { usage }); + } + + /** + * Effort as reported by a hook firing inside a subagent. + * + * This is the only live source: the message stream carries no effort at any depth, and the + * value Paseo requested is not necessarily the one that ran, because a model that does not + * support the requested level is silently downgraded. Hooks report the post-downgrade level. + * + * `agent_id` is the same id `task_started` calls `task_id`, so it routes through the table + * already built at declaration — no second identity to reconcile. + */ + observeHook(input: ClaudeHookObservationInput): SubagentObservation[] { + const taskId = readString(input.agent_id); + const effort = readString(input.effort?.level); + if (!taskId || !effort) return []; + const id = this.subagentIdByTaskId.get(taskId); + if (!id) return []; + return this.updatePresentation(id, { effort }); + } + + /** + * The model the child is actually running, read off its own assistant frames. + * + * Routed through the declaration for the same reason status is: a frame carrying the tool_use + * id of a task this source filtered out — ambient housekeeping, a workflow child, a nested + * grandchild announced in someone else's session — would otherwise fold to an upsert with no + * identity and a defaulted "running" status. That is the nameless, never-finishing row the + * declaration filter exists to prevent, arriving by a different door. + */ + observeSidechainFrame(message: SDKMessage, subagentId: string): SubagentObservation[] { + if (message.type !== "assistant" || !this.declaredIds.has(subagentId)) return []; + const model = resolveObservedClaudeModelId( + typeof message.message?.model === "string" ? message.message.model : undefined, + ); + if (!model) return []; + return this.updatePresentation(subagentId, { model }); + } + + private updatePresentation( + id: string, + patch: ClaudeSubagentPresentationFacts, + ): SubagentObservation[] { + const previous = this.presentationById.get(id) ?? {}; + const next: ClaudeSubagentPresentationFacts = { ...previous, ...patch }; + this.presentationById.set(id, next); + const subtitle = buildClaudeSubagentSubtitle(next); + if (!subtitle || this.lastSubtitleById.get(id) === subtitle) return []; + this.lastSubtitleById.set(id, subtitle); + return [{ kind: "subtitle", id, subtitle }]; + } + + /** + * Status is routed only through a task this source declared. + * + * `task_notification` also fires for the tasks deliberately filtered above, and it carries a + * `tool_use_id`. Falling back to that id would readmit exactly what the filter rejected — as a + * descriptor holding a status and no identity, which the track renders as a nameless row. + * A status without a declaration describes nothing, so it is dropped. + */ + private observeStatus(taskId: string, rawStatus: string | undefined): SubagentObservation[] { + const id = this.subagentIdByTaskId.get(taskId); + if (!id) return []; + const status = mapTaskStatus(rawStatus); + if (!status || this.lastStatusById.get(id) === status) return []; + this.lastStatusById.set(id, status); + return [{ kind: "status", id, status }]; + } +} diff --git a/packages/server/src/server/agent/providers/claude/subagents/observation.test.ts b/packages/server/src/server/agent/providers/claude/subagents/observation.test.ts new file mode 100644 index 000000000..3690334af --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/observation.test.ts @@ -0,0 +1,139 @@ +import { describe, expect, it } from "vitest"; + +import { ProviderSubagentStore } from "../../../provider-subagents/store.js"; +import { foldSubagentObservations, type SubagentObservation } from "./observation.js"; + +describe("foldSubagentObservations", () => { + it("turns a declaration into a running upsert carrying the task", () => { + expect( + foldSubagentObservations([ + { + kind: "declared", + id: "toolu_1", + title: "Explore", + description: "Find subagent title code", + toolCallId: "toolu_1", + }, + ]), + ).toEqual([ + { + type: "upsert", + id: "toolu_1", + status: "running", + title: "Explore", + description: "Find subagent title code", + toolCallId: "toolu_1", + }, + ]); + }); + + it("emits status alone so the store's sticky merge keeps identity", () => { + expect( + foldSubagentObservations([{ kind: "status", id: "toolu_1", status: "completed" }]), + ).toEqual([{ type: "upsert", id: "toolu_1", status: "completed" }]); + }); + + it("omits absent fields rather than clearing them", () => { + const [event] = foldSubagentObservations([{ kind: "declared", id: "toolu_1" }]); + expect(Object.keys(event ?? {}).sort()).toEqual(["id", "status", "type"]); + }); + + it("distinguishes an explicitly null-ish description from an absent one", () => { + // An empty description is a real observation of "no task text", not "unknown". + const [event] = foldSubagentObservations([ + { kind: "declared", id: "toolu_1", description: "" }, + ]); + expect(event).toMatchObject({ description: "" }); + }); + + it("passes timeline items through with their timestamp", () => { + const item = { type: "reasoning", text: "thinking" } as const; + expect( + foldSubagentObservations([ + { kind: "timeline", id: "toolu_1", item, timestamp: "2026-07-26T00:00:00.000Z" }, + ]), + ).toEqual([{ type: "timeline", id: "toolu_1", item, timestamp: "2026-07-26T00:00:00.000Z" }]); + }); + + it("is stateless: folding the same observation twice yields the same event", () => { + const observation: SubagentObservation = { + kind: "declared", + id: "toolu_1", + title: "Explore", + }; + expect(foldSubagentObservations([observation])).toEqual( + foldSubagentObservations([observation]), + ); + }); + + it("accumulates through the store so a later status keeps the declared identity", () => { + // The property the whole design rests on: sources emit sparse facts, the store merges them. + const store = new ProviderSubagentStore(); + const events = foldSubagentObservations([ + { kind: "declared", id: "toolu_1", title: "Explore", description: "Find the code" }, + { kind: "status", id: "toolu_1", status: "completed" }, + ]); + + let descriptor = null; + for (const event of events) { + const applied = store.apply("parent", "claude", event); + if (applied.type === "upsert") descriptor = applied.subagent; + } + + expect(descriptor).toMatchObject({ + id: "toolu_1", + title: "Explore", + description: "Find the code", + status: "completed", + }); + }); +}); + +describe("subtitle observations", () => { + it("reports provider context without asserting a status", () => { + const [event] = foldSubagentObservations([ + { kind: "subtitle", id: "toolu_1", subtitle: "Explore · Opus 5 · High" }, + ]); + expect(event).toEqual({ + type: "upsert", + id: "toolu_1", + subtitle: "Explore · Opus 5 · High", + }); + expect(event).not.toHaveProperty("status"); + }); + + it("does not revert a finished subagent when presentation changes", () => { + const store = new ProviderSubagentStore(); + let descriptor = null; + for (const event of foldSubagentObservations([ + { kind: "declared", id: "toolu_1", title: "Explore" }, + { kind: "status", id: "toolu_1", status: "completed" }, + { kind: "subtitle", id: "toolu_1", subtitle: "Explore · Opus 5 · 32.3k tokens" }, + ])) { + const applied = store.apply("parent", "claude", event); + if (applied.type === "upsert") descriptor = applied.subagent; + } + expect(descriptor).toMatchObject({ + status: "completed", + subtitle: "Explore · Opus 5 · 32.3k tokens", + }); + }); + + it("keeps identity when a later subtitle arrives", () => { + const store = new ProviderSubagentStore(); + let descriptor = null; + for (const event of foldSubagentObservations([ + { kind: "declared", id: "toolu_1", title: "Explore", description: "Find the code" }, + { kind: "subtitle", id: "toolu_1", subtitle: "Explore · Opus 5" }, + ])) { + const applied = store.apply("parent", "claude", event); + if (applied.type === "upsert") descriptor = applied.subagent; + } + expect(descriptor).toMatchObject({ + title: "Explore", + description: "Find the code", + subtitle: "Explore · Opus 5", + status: "running", + }); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/observation.ts b/packages/server/src/server/agent/providers/claude/subagents/observation.ts new file mode 100644 index 000000000..718a9fc10 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/observation.ts @@ -0,0 +1,91 @@ +import type { AgentTimelineItem } from "../../../agent-sdk-types.js"; +import type { + ProviderSubagentInputEvent, + ProviderSubagentStatus, +} from "../../../provider-subagents/store.js"; + +/** + * A single fact a provider observed about one subagent. + * + * Sources translate their wire format into observations. Claude-specific presentation is reduced + * to a subtitle before it crosses this boundary, while lifecycle and timeline facts stay + * structural. Live and on-disk replay therefore speak one vocabulary. + * + * The union stays small on purpose: a kind earns its place when a descriptor field exists to + * receive it and a source exists to report it. + */ +export type SubagentObservation = + | { + /** The subagent exists, and here is what it was asked to do. */ + kind: "declared"; + id: string; + /** Subagent type, e.g. "Explore". Identical across a fan-out — not the task. */ + title?: string; + /** The task this child was given. What actually distinguishes it from its siblings. */ + description?: string; + toolCallId?: string; + timestamp?: string; + } + | { kind: "status"; id: string; status: ProviderSubagentStatus; timestamp?: string } + /** Complete provider-owned secondary label. Clients display this without parsing it. */ + | { kind: "subtitle"; id: string; subtitle: string; timestamp?: string } + | { kind: "timeline"; id: string; item: AgentTimelineItem; timestamp?: string }; + +/** + * Turn observations into store events. + * + * Pure and stateless by construction. Both the task protocol and a persisted transcript are + * edge-triggered — a subagent is declared once and its status changes only when it changes — so + * there is nothing to debounce here. Accumulation across partial upserts is the store's sticky + * merge (omitted field preserves, explicit null clears), which is why a `status` observation can + * carry status alone without blanking the title. + */ +export function foldSubagentObservations( + observations: readonly SubagentObservation[], +): ProviderSubagentInputEvent[] { + const events: ProviderSubagentInputEvent[] = []; + for (const observation of observations) { + if (observation.kind === "timeline") { + events.push({ + type: "timeline", + id: observation.id, + item: observation.item, + ...(observation.timestamp ? { timestamp: observation.timestamp } : {}), + }); + continue; + } + + if (observation.kind === "subtitle") { + events.push({ + type: "upsert", + id: observation.id, + subtitle: observation.subtitle, + ...(observation.timestamp ? { timestamp: observation.timestamp } : {}), + }); + continue; + } + + if (observation.kind === "status") { + events.push({ + type: "upsert", + id: observation.id, + status: observation.status, + ...(observation.timestamp ? { timestamp: observation.timestamp } : {}), + }); + continue; + } + + // A declared subagent is running until something says otherwise. Its terminal status + // arrives as its own observation, so this never has to guess at completion. + events.push({ + type: "upsert", + id: observation.id, + status: "running", + ...(observation.title === undefined ? {} : { title: observation.title }), + ...(observation.description === undefined ? {} : { description: observation.description }), + ...(observation.toolCallId === undefined ? {} : { toolCallId: observation.toolCallId }), + ...(observation.timestamp ? { timestamp: observation.timestamp } : {}), + }); + } + return events; +} diff --git a/packages/server/src/server/agent/providers/claude/subagents/presentation.test.ts b/packages/server/src/server/agent/providers/claude/subagents/presentation.test.ts new file mode 100644 index 000000000..6fd2b191a --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/presentation.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; + +import { buildClaudeSubagentSubtitle } from "./presentation.js"; + +describe("buildClaudeSubagentSubtitle", () => { + it("formats Claude facts into one compact provider-owned label", () => { + expect( + buildClaudeSubagentSubtitle({ + title: "general-purpose", + model: "claude-opus-5", + effort: "high", + usage: { totalTokens: 16_484 }, + }), + ).toBe("general-purpose · Opus 5 · High · 16.5k tokens"); + }); + + it("uses the manifest label for dated and context-window model variants", () => { + expect(buildClaudeSubagentSubtitle({ model: "claude-opus-4-8[1m]" })).toBe("Opus 4.8 1M"); + }); + + it("keeps unknown compatible-provider model names visible", () => { + expect(buildClaudeSubagentSubtitle({ model: "glm-5.1" })).toBe("glm-5.1"); + }); + + it("omits facts that were not observed", () => { + expect(buildClaudeSubagentSubtitle({ title: "Explore", usage: { totalTokens: 0 } })).toBe( + "Explore", + ); + expect(buildClaudeSubagentSubtitle({})).toBeUndefined(); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/presentation.ts b/packages/server/src/server/agent/providers/claude/subagents/presentation.ts new file mode 100644 index 000000000..87fb73f29 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/presentation.ts @@ -0,0 +1,53 @@ +import { findClaudeModel } from "../models.js"; + +export interface ClaudeSubagentUsage { + totalTokens?: number; +} + +export interface ClaudeSubagentPresentationFacts { + title?: string; + model?: string; + effort?: string; + usage?: ClaudeSubagentUsage; +} + +/** Build the complete compact subtitle Claude exposes to provider-neutral clients. */ +export function buildClaudeSubagentSubtitle( + facts: ClaudeSubagentPresentationFacts, +): string | undefined { + const parts = [ + readPart(facts.title), + formatModel(facts.model), + formatEffort(facts.effort), + formatTokens(facts.usage?.totalTokens), + ].filter((part): part is string => part !== undefined); + return parts.length > 0 ? parts.join(" · ") : undefined; +} + +function readPart(value: string | undefined): string | undefined { + const trimmed = value?.trim(); + return trimmed ? trimmed : undefined; +} + +function formatModel(modelId: string | undefined): string | undefined { + const normalized = readPart(modelId); + if (!normalized) return undefined; + return findClaudeModel(normalized)?.label ?? normalized; +} + +function formatEffort(effort: string | undefined): string | undefined { + const normalized = readPart(effort); + if (!normalized) return undefined; + if (normalized === "xhigh") return "Extra High"; + return normalized + .split(/[-_\s]+/) + .filter(Boolean) + .map((part) => part.charAt(0).toUpperCase() + part.slice(1)) + .join(" "); +} + +function formatTokens(totalTokens: number | undefined): string | undefined { + if (typeof totalTokens !== "number" || totalTokens <= 0) return undefined; + if (totalTokens < 1000) return `${Math.round(totalTokens)} tokens`; + return `${Math.round(totalTokens / 100) / 10}k tokens`; +} diff --git a/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts b/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts new file mode 100644 index 000000000..e47fe817f --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/replay-source.test.ts @@ -0,0 +1,497 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import { describe, expect, it } from "vitest"; + +import { + ProviderSubagentStore, + type ProviderSubagentDescriptor, +} from "../../../provider-subagents/store.js"; +import { ClaudeTaskProtocolSource } from "./live-source.js"; +import { foldSubagentObservations, type SubagentObservation } from "./observation.js"; +import { + observeReplaySubagents, + parseClaudeSubagentMeta, + type ClaudeReplayEntry, + type ClaudeReplayParentFacts, +} from "./replay-source.js"; + +const TOOL_USE_ID = "toolu_01DgLoPMW9"; +const AGENT_ID = "a1730a6215e1f5cf6"; + +function emptyParent(): ClaudeReplayParentFacts { + return { toolCalls: new Map(), linksByAgentId: new Map(), outcomesByToolCallId: new Map() }; +} + +function parentWithTaskCall(failed = false): ClaudeReplayParentFacts { + return { + toolCalls: new Map([ + [ + TOOL_USE_ID, + { title: "general-purpose", description: "Summarize hover and unistyles docs" }, + ], + ]), + linksByAgentId: new Map(), + outcomesByToolCallId: new Map([[TOOL_USE_ID, { failed }]]), + }; +} + +function applyToStore(observations: SubagentObservation[]): ProviderSubagentDescriptor | null { + const store = new ProviderSubagentStore(); + let descriptor: ProviderSubagentDescriptor | null = null; + for (const event of foldSubagentObservations(observations)) { + const applied = store.apply("parent", "claude", event); + if (applied.type === "upsert") descriptor = applied.subagent; + } + return descriptor; +} + +describe("parseClaudeSubagentMeta", () => { + it("reads the real sidecar shape", () => { + expect( + parseClaudeSubagentMeta( + '{"agentType":"general-purpose","description":"Reply banana","toolUseId":"toolu_013i","spawnDepth":1}', + ), + ).toEqual({ + agentType: "general-purpose", + description: "Reply banana", + toolUseId: "toolu_013i", + spawnDepth: 1, + }); + }); + + it.each([ + ["truncated json", '{"agentType":"Explore"'], + ["not json at all", "agentType: Explore"], + ["empty", ""], + ["a json array", "[1,2,3]"], + ["an object with no known fields", '{"unrelated":true}'], + ])("treats %s as absent rather than throwing", (_label, contents) => { + expect(parseClaudeSubagentMeta(contents)).toBeNull(); + }); + + it("keeps the good fields when one field has the wrong type", () => { + expect(parseClaudeSubagentMeta('{"agentType":"Explore","toolUseId":42}')).toEqual({ + agentType: "Explore", + description: undefined, + toolUseId: undefined, + spawnDepth: undefined, + }); + }); +}); + +describe("observeReplaySubagents", () => { + it("links a subagent to its Task call through meta.toolUseId", () => { + const observations = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + expect(observations[0]).toMatchObject({ + kind: "declared", + id: TOOL_USE_ID, + toolCallId: TOOL_USE_ID, + title: "general-purpose", + description: "Summarize hover and unistyles docs", + }); + }); + + it("replays only this session's own children, not their descendants", () => { + // Claude Code writes every descendant into the same subagents/ directory. A grandchild's + // toolUseId names a Task call made inside its parent's session, so nothing here can resolve + // it: it would replay as an extra row the live stream never showed, with no Task card and no + // outcome, running forever. One recorded session showed 10 subagents live and 22 on reopen. + const observations = observeReplaySubagents({ + subagents: [ + { agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID, spawnDepth: 1 }, entries: [] }, + { + agentId: "a6acb4b898", + meta: { toolUseId: "toolu_012rzYnFZA", agentType: "Explore", spawnDepth: 2 }, + entries: [], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + expect([...new Set(observations.map((observation) => observation.id))]).toEqual([TOOL_USE_ID]); + }); + + it("keeps a subagent recorded before spawnDepth existed", () => { + const observations = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + expect(observations[0]).toMatchObject({ kind: "declared", id: TOOL_USE_ID }); + }); + + it("recovers identity from meta when the parent's Task call is missing", () => { + // The case the legacy scrape cannot handle: no tool_result, so no link and no identity. + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID, agentType: "Explore", description: "Find the code" }, + entries: [], + }, + ], + parent: emptyParent(), + convertEntry: () => [], + }); + + expect(observations[0]).toMatchObject({ + id: TOOL_USE_ID, + toolCallId: TOOL_USE_ID, + title: "Explore", + description: "Find the code", + }); + }); + + it("falls back to the scraped link when there is no meta file", () => { + const parent: ClaudeReplayParentFacts = { + toolCalls: new Map([[TOOL_USE_ID, { title: "Explore" }]]), + linksByAgentId: new Map([[AGENT_ID, { toolCallId: TOOL_USE_ID, failed: false }]]), + outcomesByToolCallId: new Map(), + }; + + const [declared, ...rest] = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }], + parent, + convertEntry: () => [], + }); + + expect(declared).toMatchObject({ id: TOOL_USE_ID, toolCallId: TOOL_USE_ID, title: "Explore" }); + expect(rest.at(-1)).toMatchObject({ kind: "status", status: "completed" }); + }); + + it("keeps an unlinkable subagent visible under its own id", () => { + const [declared] = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: null, entries: [] }], + parent: emptyParent(), + convertEntry: () => [], + }); + expect(declared).toMatchObject({ id: AGENT_ID }); + expect(declared).not.toHaveProperty("toolCallId"); + }); + + it("leaves a subagent with no recorded outcome running", () => { + const observations = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }], + parent: { ...parentWithTaskCall(), outcomesByToolCallId: new Map() }, + convertEntry: () => [], + }); + expect(observations.some((o) => o.kind === "status")).toBe(false); + }); + + it("reports a failed outcome as failed", () => { + const observations = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }], + parent: parentWithTaskCall(true), + convertEntry: () => [], + }); + expect(observations.at(-1)).toMatchObject({ kind: "status", status: "failed" }); + }); + + it("carries entry timestamps onto timeline observations", () => { + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [{ type: "assistant", timestamp: "2026-07-26T06:27:47.034Z" }], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [{ type: "reasoning", text: "thinking" }], + }); + + expect(observations.find((observation) => observation.kind === "timeline")).toMatchObject({ + kind: "timeline", + id: TOOL_USE_ID, + timestamp: "2026-07-26T06:27:47.034Z", + }); + }); +}); + +describe("replay runtime", () => { + it("reads the model and effort off the subagent's own transcript", () => { + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [{ type: "assistant", effort: "high", message: { model: "claude-opus-5" } }], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + expect(observations.find((o) => o.kind === "subtitle")).toEqual({ + kind: "subtitle", + id: TOOL_USE_ID, + subtitle: "general-purpose · Opus 5 · High", + }); + }); + + it("takes the last observed values when the model changes mid-transcript", () => { + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [ + { type: "assistant", effort: "high", message: { model: "claude-opus-5" } }, + { type: "assistant", effort: "low", message: { model: "claude-sonnet-5" } }, + ], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({ + subtitle: "general-purpose · Sonnet 5 · Low", + }); + }); + + it("reports nothing when the transcript never names a model or effort", () => { + const observations = observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries: [] }], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + expect(observations.some((o) => o.kind === "subtitle")).toBe(false); + }); + + it("passes through a model the manifest does not know", () => { + // Claude Code is an Anthropic-compatible client; subagents legitimately report other models. + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [{ type: "assistant", message: { model: "glm-5.1" } }], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + expect(observations.find((o) => o.kind === "subtitle")).toMatchObject({ + subtitle: "general-purpose · glm-5.1", + }); + }); +}); + +describe("replay usage", () => { + function subtitleOf(entries: ClaudeReplayEntry[]): SubagentObservation | undefined { + return observeReplaySubagents({ + subagents: [{ agentId: AGENT_ID, meta: { toolUseId: TOOL_USE_ID }, entries }], + parent: parentWithTaskCall(), + convertEntry: () => [], + }).find((observation) => observation.kind === "subtitle"); + } + + it("derives the counters Claude Code itself reports, from a real transcript's shape", () => { + // Shape and numbers taken from an on-disk subagent transcript. Claude Code finalizes a + // subagent from the LAST assistant entry's usage, so the earlier, smaller turn must not win + // and must not be added in. + expect( + subtitleOf([ + { + type: "assistant", + timestamp: "2026-07-23T23:18:43.023Z", + message: { + model: "claude-opus-5", + content: [ + { type: "tool_use", name: "Read" }, + { type: "text", text: "hi" }, + ], + usage: { + input_tokens: 4, + cache_creation_input_tokens: 15_000, + cache_read_input_tokens: 0, + output_tokens: 120, + }, + }, + }, + { type: "user", timestamp: "2026-07-23T23:20:00.000Z", message: { content: [] } }, + { + type: "assistant", + timestamp: "2026-07-23T23:49:05.068Z", + message: { + model: "claude-opus-5", + content: [{ type: "tool_use", name: "Grep" }], + usage: { + input_tokens: 2293, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 65_024, + output_tokens: 1376, + }, + }, + }, + ]), + ).toEqual({ + kind: "subtitle", + id: TOOL_USE_ID, + // 2293 + 0 + 65024 + 1376 — the last turn's context, matching the live path's total_tokens. + subtitle: "general-purpose · Opus 5 · 68.7k tokens", + timestamp: "2026-07-23T23:49:05.068Z", + }); + }); + + it("formats tokens for a child that only answered", () => { + expect( + subtitleOf([ + { + type: "assistant", + timestamp: "2026-07-27T16:42:14.698Z", + message: { content: [{ type: "text", text: "done" }], usage: { input_tokens: 16_434 } }, + }, + { + type: "user", + timestamp: "2026-07-27T16:42:16.926Z", + message: { content: "ignored" }, + }, + ]), + ).toMatchObject({ subtitle: "general-purpose · 16.4k tokens" }); + }); + + it("emits nothing for a transcript with no entries", () => { + expect(subtitleOf([])).toBeUndefined(); + }); + + it("keeps duration out of the compact subtitle", () => { + const observation = subtitleOf([ + { + type: "assistant", + timestamp: "2026-07-27T16:42:14.698Z", + message: { content: [], usage: { input_tokens: 10 } }, + }, + ]); + expect(observation).toMatchObject({ subtitle: "general-purpose · 10 tokens" }); + }); + + it("omits totalTokens when no entry carries a usage block", () => { + const observation = subtitleOf([ + { type: "assistant", timestamp: "2026-07-27T16:42:14.698Z", message: { content: [] } }, + { + type: "assistant", + timestamp: "2026-07-27T16:42:20.000Z", + message: { content: [{ type: "tool_use", name: "Bash" }], usage: "not-an-object" }, + }, + ]); + expect(observation).toBeUndefined(); + }); + + it("keeps the last usage-bearing entry when the final assistant entry has none", () => { + expect( + subtitleOf([ + { + type: "assistant", + timestamp: "2026-07-27T16:42:14.698Z", + message: { content: [], usage: { input_tokens: 900, output_tokens: 100 } }, + }, + { type: "assistant", timestamp: "2026-07-27T16:42:20.000Z", message: { content: [] } }, + ]), + ).toMatchObject({ subtitle: "general-purpose · 1k tokens" }); + }); + + it("only reports counters the transcript actually shows", () => { + // A user-only transcript says nothing about cost: no assistant turn, so no tool count either. + const observation = subtitleOf([ + { type: "user", timestamp: "2026-07-27T16:42:14.698Z", message: { content: "prompt" } }, + { type: "user", timestamp: "2026-07-27T16:42:20.000Z", message: { content: "more" } }, + ]); + expect(observation).toBeUndefined(); + }); + + it("lands the complete provider-owned subtitle on the descriptor", () => { + const store = new ProviderSubagentStore(); + const observations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { toolUseId: TOOL_USE_ID }, + entries: [ + { + type: "assistant", + timestamp: "2026-07-27T16:42:14.698Z", + message: { content: [], usage: { input_tokens: 7 } }, + }, + ], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + for (const event of foldSubagentObservations(observations)) { + store.apply("parent", "claude", event); + } + + expect(store.get("parent", TOOL_USE_ID)?.subtitle).toBe("general-purpose · 7 tokens"); + }); +}); + +describe("live and replay agree", () => { + // The property the design exists to guarantee: one session, two sources, one descriptor. + // This is inexpressible against the inference implementation, where the paths share no + // vocabulary and each derives identity and status its own way. + it("produces identical descriptor state from the wire and from disk", () => { + const live = new ClaudeTaskProtocolSource(); + const liveObservations = [ + ...live.observe({ + type: "system", + subtype: "task_started", + task_id: AGENT_ID, + tool_use_id: TOOL_USE_ID, + description: "Summarize hover and unistyles docs", + subagent_type: "general-purpose", + task_type: "local_agent", + } as unknown as SDKMessage), + ...live.observe({ + type: "system", + subtype: "task_updated", + task_id: AGENT_ID, + patch: { status: "completed" }, + } as unknown as SDKMessage), + ]; + + const replayObservations = observeReplaySubagents({ + subagents: [ + { + agentId: AGENT_ID, + meta: { + toolUseId: TOOL_USE_ID, + agentType: "general-purpose", + description: "Summarize hover and unistyles docs", + }, + entries: [], + }, + ], + parent: parentWithTaskCall(), + convertEntry: () => [], + }); + + const fromWire = applyToStore(liveObservations); + const fromDisk = applyToStore(replayObservations); + + const identity = (descriptor: ProviderSubagentDescriptor | null) => ({ + id: descriptor?.id, + title: descriptor?.title, + description: descriptor?.description, + status: descriptor?.status, + toolCallId: descriptor?.toolCallId, + }); + + expect(identity(fromWire)).toEqual(identity(fromDisk)); + expect(identity(fromWire)).toEqual({ + id: TOOL_USE_ID, + title: "general-purpose", + description: "Summarize hover and unistyles docs", + status: "completed", + toolCallId: TOOL_USE_ID, + }); + }); +}); diff --git a/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts new file mode 100644 index 000000000..61fc2c54f --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/subagents/replay-source.ts @@ -0,0 +1,312 @@ +import { z } from "zod"; + +import type { AgentTimelineItem } from "../../../agent-sdk-types.js"; +import { normalizeProviderReplayTimestamp } from "../../../provider-history-timestamps.js"; +import { resolveObservedClaudeModelId } from "../models.js"; +import type { SubagentObservation } from "./observation.js"; +import { buildClaudeSubagentSubtitle, type ClaudeSubagentUsage } from "./presentation.js"; + +/** + * Rebuilds subagent observations from a persisted session, producing the same vocabulary the + * live task-protocol source produces. Identical observations in means identical descriptor + * state out, so a fact is derived once for both paths rather than once per path. + * + * Claude Code writes each subagent as `/subagents/agent-.jsonl` beside an + * `agent-.meta.json`. The meta file carries the Task `tool_use_id`, which is the same + * canonical id the live stream uses — it is the bridge that lets the two paths agree. + */ + +const ClaudeSubagentMetaSchema = z.object({ + agentType: z.string().optional().catch(undefined), + description: z.string().optional().catch(undefined), + toolUseId: z.string().optional().catch(undefined), + spawnDepth: z.number().optional().catch(undefined), +}); + +export type ClaudeSubagentMeta = z.infer; + +/** + * Parse a sidecar meta file. This is undocumented Claude Code internals, so every field is + * optional and a malformed file is indistinguishable from an absent one: callers fall back to + * the parent-transcript derivation rather than losing the subagent. + */ +export function parseClaudeSubagentMeta(contents: string): ClaudeSubagentMeta | null { + let parsed: unknown; + try { + parsed = JSON.parse(contents); + } catch { + return null; + } + const result = ClaudeSubagentMetaSchema.safeParse(parsed); + if (!result.success) return null; + const meta = result.data; + const hasAnyField = + meta.agentType !== undefined || + meta.description !== undefined || + meta.toolUseId !== undefined || + meta.spawnDepth !== undefined; + return hasAnyField ? meta : null; +} + +export interface ClaudeReplayEntry { + type?: unknown; + agentId?: unknown; + timestamp?: unknown; + /** Claude Code stamps the active effort on each assistant entry. */ + effort?: unknown; + message?: { content?: unknown; [key: string]: unknown }; + [key: string]: unknown; +} + +/** + * Last-observed model and effort from a subagent's own assistant entries. + * + * Effort is replay-only: the SDK's live assistant message does not carry it, and it can be + * silently downgraded for the selected model, so any live value would be a guess. + */ +function readRuntime(entries: readonly ClaudeReplayEntry[]): { model?: string; effort?: string } { + const runtime: { model?: string; effort?: string } = {}; + for (const entry of entries) { + if (entry.type !== "assistant") continue; + const effort = typeof entry.effort === "string" ? entry.effort.trim() : ""; + if (effort) runtime.effort = effort; + const rawModel = (entry.message as { model?: unknown } | undefined)?.model; + const model = resolveObservedClaudeModelId(typeof rawModel === "string" ? rawModel : undefined); + if (model) runtime.model = model; + } + return runtime; +} + +/** + * Token counters as Claude Code stamps them onto an assistant entry. Undocumented internals, so + * every field is optional and a wrong-typed field reads as absent rather than failing the entry. + */ +const ClaudeReplayUsageSchema = z.object({ + input_tokens: z.number().optional().catch(undefined), + cache_creation_input_tokens: z.number().optional().catch(undefined), + cache_read_input_tokens: z.number().optional().catch(undefined), + output_tokens: z.number().optional().catch(undefined), +}); + +/** + * The one number the live path calls `total_tokens`. + * + * This is not a guess. Claude Code 2.1.220 finalizes a subagent by summing the usage block of the + * LAST assistant message — `input + (cache_creation ?? 0) + (cache_read ?? 0) + output` — and ships + * that as `usage.total_tokens` on `task_notification`; `task_progress` reports the same sum from + * the live tracker as it climbs. So it is a context-size reading at the final turn, not a + * cumulative spend across turns, which is why a small subagent reports a number in the tens of + * thousands: the reused prompt cache dominates it. Summing per-entry usage instead would multiply + * the cached prefix by the turn count and report a number several times larger than the live path. + */ +function readTotalTokens(raw: unknown): number | undefined { + const parsed = ClaudeReplayUsageSchema.safeParse(raw); + if (!parsed.success) return undefined; + const { input_tokens, cache_creation_input_tokens, cache_read_input_tokens, output_tokens } = + parsed.data; + const counters = [ + input_tokens, + cache_creation_input_tokens, + cache_read_input_tokens, + output_tokens, + ]; + // An object with none of the four counters is not a usage block; reporting 0 would claim the + // subagent spent nothing. + if (counters.every((counter) => counter === undefined)) return undefined; + return counters.reduce((total: number, counter) => total + (counter ?? 0), 0); +} + +/** + * Token context of a subagent's run, recovered from its own transcript. + * + * Descriptors are in-memory and dropped when the agent closes, so without this a reopened agent + * permanently loses its children's counters. The live path gets these numbers handed to it by + * Claude Code; here they are re-derived to match those definitions: + * + * The last assistant entry's summed usage matches Claude Code's live `total_tokens` definition. + */ +function readUsage(entries: readonly ClaudeReplayEntry[]): ClaudeSubagentUsage | null { + let totalTokens: number | undefined; + + for (const entry of entries) { + if (entry.type !== "assistant") continue; + const observed = readTotalTokens(entry.message?.usage); + if (observed !== undefined) totalTokens = observed; + } + + return totalTokens === undefined ? null : { totalTokens }; +} + +export interface ClaudeReplaySubagentInput { + agentId: string; + meta: ClaudeSubagentMeta | null; + entries: ClaudeReplayEntry[]; +} + +export interface ClaudeReplayParentFacts { + /** Task `tool_use` id -> identity declared in the parent's tool input. */ + toolCalls: ReadonlyMap; + /** agentId -> tool call, recovered by scraping tool-result text. Legacy fallback only. */ + linksByAgentId: ReadonlyMap; + /** Task `tool_use` id -> outcome. Usable once meta.json supplies the link directly. */ + outcomesByToolCallId: ReadonlyMap; +} + +interface ResolvedLink { + id: string; + toolCallId: string | null; + failed: boolean | null; +} + +/** + * Resolve the canonical subagent id. + * + * `meta.toolUseId` is authoritative: it is the same id the live stream keys on. The legacy path + * scrapes `agentId:` out of stringified tool-result content, which both misses the link when the + * parent's tool_result is absent and matches unrelated text — on a real five-subagent session it + * produced eight ids, inventing `z`, `string`, and `update`. It stays only as a fallback for + * sessions recorded before the meta file existed. + */ +function resolveLink( + subagent: ClaudeReplaySubagentInput, + parent: ClaudeReplayParentFacts, +): ResolvedLink { + const metaToolUseId = subagent.meta?.toolUseId?.trim(); + if (metaToolUseId) { + const outcome = parent.outcomesByToolCallId.get(metaToolUseId); + return { + id: metaToolUseId, + toolCallId: metaToolUseId, + failed: outcome ? outcome.failed : null, + }; + } + + const scraped = parent.linksByAgentId.get(subagent.agentId); + if (scraped) { + return { id: scraped.toolCallId, toolCallId: scraped.toolCallId, failed: scraped.failed }; + } + + // No link at all: keep the subagent visible under its own agent id rather than dropping it. + return { id: subagent.agentId, toolCallId: null, failed: null }; +} + +/** + * The parent's tool input is the same source the live path reads, so it wins; meta.json fills the + * gap when the parent's Task call is missing from the transcript. + */ +function declareSubagent( + subagent: ClaudeReplaySubagentInput, + link: ResolvedLink, + toolCall: { title?: string; description?: string } | undefined, +): SubagentObservation { + const title = toolCall?.title ?? subagent.meta?.agentType; + const description = toolCall?.description ?? subagent.meta?.description; + const firstTimestamp = normalizeProviderReplayTimestamp(subagent.entries[0]?.timestamp); + return { + kind: "declared", + id: link.id, + ...(title ? { title } : {}), + ...(description ? { description } : {}), + ...(link.toolCallId ? { toolCallId: link.toolCallId } : {}), + ...(firstTimestamp ? { timestamp: firstTimestamp } : {}), + }; +} + +function observeSubtitle( + subagent: ClaudeReplaySubagentInput, + link: ResolvedLink, + title: string | undefined, +): SubagentObservation | null { + const runtime = readRuntime(subagent.entries); + const usage = readUsage(subagent.entries); + const hasDetails = + runtime.model !== undefined || + runtime.effort !== undefined || + (usage?.totalTokens !== undefined && usage.totalTokens > 0); + if (!hasDetails) return null; + const subtitle = buildClaudeSubagentSubtitle({ + title, + ...runtime, + ...(usage ? { usage } : {}), + }); + if (!subtitle) return null; + const timestamp = normalizeProviderReplayTimestamp(subagent.entries.at(-1)?.timestamp); + return { + kind: "subtitle", + id: link.id, + subtitle, + ...(timestamp ? { timestamp } : {}), + }; +} + +function observeSubagent( + subagent: ClaudeReplaySubagentInput, + parent: ClaudeReplayParentFacts, + convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[], +): SubagentObservation[] { + const link = resolveLink(subagent, parent); + const toolCall = link.toolCallId ? parent.toolCalls.get(link.toolCallId) : undefined; + + const observations: SubagentObservation[] = [declareSubagent(subagent, link, toolCall)]; + const subtitle = observeSubtitle(subagent, link, toolCall?.title ?? subagent.meta?.agentType); + if (subtitle) observations.push(subtitle); + + for (const entry of subagent.entries) { + const timestamp = normalizeProviderReplayTimestamp(entry.timestamp); + for (const item of convertEntry(entry)) { + observations.push({ + kind: "timeline", + id: link.id, + item, + ...(timestamp ? { timestamp } : {}), + }); + } + } + + // A transcript with no recorded outcome is a subagent that never finished — leave it running + // rather than inventing a completion. + if (link.failed !== null) { + const lastTimestamp = normalizeProviderReplayTimestamp(subagent.entries.at(-1)?.timestamp); + observations.push({ + kind: "status", + id: link.id, + status: link.failed ? "failed" : "completed", + ...(lastTimestamp ? { timestamp: lastTimestamp } : {}), + }); + } + + return observations; +} + +/** + * Whether this transcript belongs to a child of the session being replayed. + * + * Claude Code writes every descendant into the same `subagents/` directory, not just the session's + * own children, and `spawnDepth` is what separates them: 1 is a direct child, 2+ was spawned by + * another subagent. A grandchild's `toolUseId` names a Task call made inside its parent's session, + * so nothing in this transcript can resolve it — surveyed across recorded sessions, every depth-1 + * id matched a Task tool_use block in the parent transcript and no deeper id did. Replaying them + * anyway adds rows the live stream never showed, each with no Task card and no recoverable + * outcome, which leaves them running forever. One session recorded 10 subagents live and would + * replay 22. + * + * Absent depth means a session recorded before the field existed: keep the subagent rather than + * lose it, matching how every other missing field here is treated. + */ +function isDirectChild(subagent: ClaudeReplaySubagentInput): boolean { + const depth = subagent.meta?.spawnDepth; + return depth === undefined || depth <= 1; +} + +export function observeReplaySubagents(input: { + subagents: readonly ClaudeReplaySubagentInput[]; + parent: ClaudeReplayParentFacts; + convertEntry: (entry: ClaudeReplayEntry) => AgentTimelineItem[]; +}): SubagentObservation[] { + const observations: SubagentObservation[] = []; + for (const subagent of input.subagents) { + if (!isDirectChild(subagent)) continue; + observations.push(...observeSubagent(subagent, input.parent, input.convertEntry)); + } + return observations; +} diff --git a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts index 00fed1ac8..39b4d10b4 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts @@ -479,7 +479,9 @@ export class OmpHarness { .map(([callId]) => callId); } - subagentUpserts(): Array<{ id: string; status: string }> { + // `status` is optional on the upsert event — an upsert may report only model or usage. OMP + // always sets one, so this stays a plain string for assertions. + subagentUpserts(): Array<{ id: string; status: string | undefined }> { return this.events.flatMap((event) => event.type === "provider_subagent" && event.event.type === "upsert" ? [{ id: event.event.id, status: event.event.status }]