From 720964413fde85c6e35fc84130a715adbc97f92c Mon Sep 17 00:00:00 2001 From: Shivam Sharma <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:51:40 +0530 Subject: [PATCH 1/2] feat(orchestration-v2): populate subagent observability from providers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fills in the data model added by the previous PR. Every adapter now reports what its subagents are actually doing: role provenance, token usage, recent activity, and per-activation records. Usage needs two merge strategies because providers report it differently. Codex sends cumulative snapshots per thread, so those merge by maximum — duplicate and late out-of-order frames then leave the total unchanged instead of inflating it. Claude reports per activation, so lifetime usage accumulates only the delta since the previous snapshot. A subagent identity is reusable, so completing an activation leaves it at "idle" rather than "completed"; the timeline row for that activation is still recorded as completed via the status mapping. A Codex subAgentActivity frame carries no model of its own, so agents registered from one were recorded with none at all — the common case for Codex-native subagents, which then showed a role and a token count but no model. They are recorded against the turn's model selection, which is what they actually run on. Three activation-lifecycle fixes in the Claude adapter, found in review: - A resume that raced past settle was pre-opened by the wake buffer, which erased the terminal status the reopen check reads. The resume then reused the finished activation instead of starting a new one, so its ordinal stopped matching reality and its usage was measured against the previous activation's totals — clamping the delta to zero and silently dropping the resumed run's tokens. The pre-open now records that it happened, and the reopen check honours it. - A replayed "waiting" frame could resurrect a settled subagent; only pending/running were guarded before. - A duplicate terminal frame re-stamped completedAt, drifting the recorded completion time. Codex already preserved it; Claude now matches. Reuse across runs is deliberately not addressed here — a subagent row still carries the run it was spawned under. That is the next PR. --- .../orchestration-v2/Adapters/AcpAdapterV2.ts | 38 +- .../Adapters/ClaudeAdapterV2.test.ts | 292 +++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 350 ++++++++++++++++-- .../Adapters/CodexAdapterV2.test.ts | 102 ++++- .../Adapters/CodexAdapterV2.ts | 244 ++++++++++-- .../Adapters/CursorAdapterV2.ts | 29 +- .../Adapters/OpenCodeAdapterV2.ts | 38 +- .../ProviderRuntimeRecoveryService.test.ts | 3 + .../SubagentObservability.test.ts | 137 ++++++- .../orchestration-v2/SubagentObservability.ts | 156 ++++++++ .../testkit/fixtures/subagent/codex_output.ts | 3 +- .../subagent_continue/codex_output.ts | 6 +- .../fixtures/subagent_v2/codex_output.ts | 14 +- .../subagent_v2_nested/codex_output.ts | 18 +- 14 files changed, 1336 insertions(+), 94 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts index 2ae679fff3e..504f5616f0e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts @@ -12,6 +12,7 @@ import { type OrchestrationV2ProviderTurn, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivation, type OrchestrationV2TurnItem, type OrchestrationV2UserInputQuestion, type ProviderApprovalDecision, @@ -66,6 +67,7 @@ import { makeSubagentConversationArtifacts, subagentThreadTitle, } from "../SubagentProjection.ts"; +import { defaultSubagentRole, subagentActivationId } from "../SubagentObservability.ts"; import { ProviderAdapterEnsureThreadError, ProviderAdapterForkThreadError, @@ -1947,6 +1949,12 @@ export function makeAcpAdapterV2(options: AcpAdapterV2Options): ProviderAdapterV const turnItemOrdinal = existing?.turnItemOrdinal ?? (yield* resolveItemOrdinal(context, nativeTaskId)); const taskStatus = update.status; + const activationId = subagentActivationId(nodeId, 1); + const settled = + taskStatus === "completed" || + taskStatus === "failed" || + taskStatus === "cancelled" || + taskStatus === "interrupted"; const task: OrchestrationV2Subagent = { ...(existing?.task ?? { id: nodeId, @@ -1963,20 +1971,21 @@ export function makeAcpAdapterV2(options: AcpAdapterV2Options): ProviderAdapterV prompt: update.prompt, title: update.title, model: update.model, - kind: "subagent", - role: { name: "general-purpose", source: "app_default" }, + kind: "subagent" as const, + role: defaultSubagentRole(), + result: null, usage: null, - currentActivationId: null, + currentActivationId: activationId, activationCount: 1, workflow: null, workflowMembership: null, recentActivity: [], - result: null, startedAt: now, }), status: taskStatus, result: existing?.assistantText || update.result, - completedAt: taskStatus === "running" ? null : now, + currentActivationId: settled ? null : activationId, + completedAt: settled ? (existing?.task.completedAt ?? now) : null, updatedAt: now, }; const subagent: ActiveAcpSubagent = existing ?? { @@ -2094,7 +2103,7 @@ export function makeAcpAdapterV2(options: AcpAdapterV2Options): ProviderAdapterV ...subagent.task, status: taskStatus, result, - completedAt: taskStatus === "running" ? null : now, + completedAt: settled ? (subagent.task.completedAt ?? now) : null, updatedAt: now, }; const providerThreadId = subagent.task.providerThreadId; @@ -2141,6 +2150,23 @@ export function makeAcpAdapterV2(options: AcpAdapterV2Options): ProviderAdapterV }, }); yield* emitProviderEvent({ type: "subagent.updated", driver, subagent: subagent.task }); + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver, + activation: { + id: activationId, + threadId: subagent.task.threadId, + subagentId: subagent.task.id, + runId: subagent.task.runId, + providerTurnId: subagent.providerTurnId, + ordinal: 1, + status: taskStatus, + usage: null, + startedAt: subagent.task.startedAt, + completedAt: settled ? now : null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation, + }); yield* emitProviderEvent({ type: "turn_item.updated", driver, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 1834749bd15..b7027514b55 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1322,6 +1322,268 @@ describe("ClaudeAdapterV2 background wake turns", () => { }); const makeWakeHarness = makeWakeHarnessWithOptions(); + it.effect("projects Claude workflow phases and workers as structured subagents", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + const workflowTaskId = "task-workflow-observability"; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-workflow-observability"), + text: "Run a two-phase workflow.", + attachments: [], + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: workflowTaskId, + tool_use_id: "toolu-workflow-observability", + description: "Research and implement", + task_type: "local_workflow", + prompt: "Research, then implement.", + uuid: "00000000-0000-4000-8000-000000000121", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: workflowTaskId, + tool_use_id: "toolu-workflow-observability", + description: "Implementation started", + usage: { + input_tokens: 300, + output_tokens: 200, + total_tokens: 500, + tool_uses: 4, + }, + workflow_progress: [ + { type: "workflow_phase", index: 0, title: "Research" }, + { type: "workflow_phase", index: 1, title: "Implement" }, + { + type: "workflow_agent", + index: 0, + label: "Researcher", + state: "done", + phaseIndex: 0, + attempt: 0, + model: "claude-sonnet-4-6", + lastToolName: "Read", + tokens: 200, + toolCalls: 2, + }, + { + type: "workflow_agent", + index: 1, + label: "Implementer", + state: "active", + phase_index: 1, + attempt: 0, + model: "claude-opus-4-1", + last_tool_name: "Edit", + tokens: 300, + tool_calls: 2, + }, + ], + uuid: "00000000-0000-4000-8000-000000000122", + session_id: WAKE_NATIVE_SESSION, + }), + ); + + const subagentEvents = () => + harness.events.filter( + (event): event is Extract => + event.type === "subagent.updated", + ); + yield* awaitUntil( + () => + new Set( + subagentEvents() + .filter((event) => event.subagent.kind === "workflow_agent") + .map((event) => event.subagent.id), + ).size === 2, + "workflow workers projected", + ); + + const latestById = new Map( + subagentEvents().map((event) => [event.subagent.id, event.subagent]), + ); + const workflow = [...latestById.values()].find((agent) => agent.kind === "workflow"); + const workers = [...latestById.values()] + .filter((agent) => agent.kind === "workflow_agent") + .sort( + (left, right) => + (left.workflowMembership?.agentIndex ?? 0) - + (right.workflowMembership?.agentIndex ?? 0), + ); + assert.deepEqual(workflow?.role, { + name: "workflow-coordinator", + source: "app_default", + }); + assert.deepEqual(workflow?.workflow?.phases, [ + { index: 0, title: "Research" }, + { index: 1, title: "Implement" }, + ]); + assert.equal(workflow?.usage?.totalTokens, 500); + assert.deepEqual( + workers.map((worker) => worker.role), + [ + { name: "workflow-worker", source: "app_default" }, + { name: "workflow-worker", source: "app_default" }, + ], + ); + assert.deepEqual( + workers.map((worker) => [ + worker.status, + worker.workflowMembership?.phaseIndex, + worker.workflowMembership?.attempt, + worker.usage?.totalTokens, + ]), + [ + ["completed", 0, 1, 200], + ["running", 1, 1, 300], + ], + ); + + const workerUpdateCountBeforeStaleActive = subagentEvents().filter( + (event) => event.subagent.kind === "workflow_agent", + ).length; + const eventCountBeforeStaleActive = subagentEvents().length; + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: workflowTaskId, + tool_use_id: "toolu-workflow-observability", + description: "Provider replayed the completed attempt as active", + workflow_progress: [ + { + type: "workflow_agent", + index: 0, + label: "Researcher", + state: "active", + phaseIndex: 0, + attempt: 0, + tokens: 200, + }, + ], + uuid: "00000000-0000-4000-8000-000000000126", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil( + () => subagentEvents().length > eventCountBeforeStaleActive, + "stale workflow attempt inspected", + ); + assert.equal( + subagentEvents().filter((event) => event.subagent.kind === "workflow_agent").length, + workerUpdateCountBeforeStaleActive, + ); + + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: workflowTaskId, + tool_use_id: "toolu-workflow-observability", + description: "Research retry queued", + workflow_progress: [ + { type: "workflow_phase", index: 0, title: "Research" }, + { type: "workflow_phase", index: 1, title: "Implement" }, + { + type: "workflow_agent", + index: 0, + label: "Researcher", + state: "start", + phaseIndex: 0, + attempt: 1, + model: "claude-sonnet-4-6", + tokens: 50, + toolCalls: 1, + }, + ], + uuid: "00000000-0000-4000-8000-000000000123", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil( + () => + subagentEvents().at(-1)?.subagent.kind === "workflow_agent" && + subagentEvents().at(-1)?.subagent.status === "pending", + "workflow retry projected", + ); + const retried = subagentEvents().at(-1)?.subagent; + assert.equal(retried?.activationCount, 2); + assert.equal(retried?.workflowMembership?.attempt, 2); + assert.equal(retried?.usage?.totalTokens, 250); + assert.lengthOf( + new Set( + harness.events + .filter( + ( + event, + ): event is Extract< + ProviderAdapterV2Event, + { type: "subagent_activation.updated" } + > => event.type === "subagent_activation.updated", + ) + .filter((event) => event.activation.subagentId === retried?.id) + .map((event) => event.activation.id), + ), + 2, + ); + + const workerUpdateCount = subagentEvents().filter( + (event) => event.subagent.kind === "workflow_agent", + ).length; + const eventCount = subagentEvents().length; + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: workflowTaskId, + tool_use_id: "toolu-workflow-observability", + description: "Provider sent a future workflow state", + workflow_progress: [ + { + type: "workflow_agent", + index: 0, + label: "Researcher", + state: "future_state", + phaseIndex: 0, + attempt: 2, + tokens: 100, + }, + ], + uuid: "00000000-0000-4000-8000-000000000124", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* awaitUntil( + () => subagentEvents().length > eventCount, + "unknown workflow state inspected", + ); + assert.equal( + subagentEvents().filter((event) => event.subagent.kind === "workflow_agent").length, + workerUpdateCount, + ); + }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))), + ), + ); + it.effect("buffers wake output and requests a single continuation run", () => Effect.scoped( Effect.gen(function* () { @@ -2293,6 +2555,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { status: "completed", output_file: "/tmp/task-resume-subagent.output", summary: FIRST_SUMMARY, + usage: { + input_tokens: 70, + output_tokens: 30, + total_tokens: 100, + tool_uses: 2, + }, uuid: "00000000-0000-4000-8000-000000000402", session_id: WAKE_NATIVE_SESSION, }); @@ -2319,6 +2587,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { status: "completed", output_file: "/tmp/task-resume-subagent.output", summary: SECOND_SUMMARY, + usage: { + input_tokens: 50, + output_tokens: 30, + total_tokens: 80, + tool_uses: 1, + }, uuid: "00000000-0000-4000-8000-000000000407", session_id: WAKE_NATIVE_SESSION, }); @@ -2330,6 +2604,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { (event): event is Extract => event.type === "subagent.updated", ); + const activationEvents = () => + harness.events.filter( + ( + event, + ): event is Extract => + event.type === "subagent_activation.updated", + ); yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -2381,6 +2662,11 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* awaitUntil(() => harness.terminalEvents().length === 2, "continuation terminal"); assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); assert.equal(subagentEvents().at(-1)?.subagent.result, FIRST_SUMMARY); + assert.deepEqual(subagentEvents().at(-1)?.subagent.role, { + name: "general-purpose", + source: "provider", + }); + assert.equal(subagentEvents().at(-1)?.subagent.usage?.totalTokens, 100); assert.isFalse(yield* harness.hasPendingBackgroundWork); // A user turn nudges the completed subagent via SendMessage; the @@ -2506,6 +2792,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { const finalSubagent = subagentEvents().at(-1)?.subagent; assert.equal(finalSubagent?.status, "completed"); assert.equal(finalSubagent?.result, SECOND_SUMMARY); + assert.equal(finalSubagent?.activationCount, 2); + assert.equal(finalSubagent?.usage?.totalTokens, 180); + assert.equal(finalSubagent?.usage?.toolUses, 3); + assert.lengthOf(new Set(activationEvents().map((event) => event.activation.id)), 2); + assert.equal(activationEvents().at(-1)?.activation.ordinal, 2); + assert.equal(activationEvents().at(-1)?.activation.usage?.totalTokens, 80); // The completion keeps the resuming run's attribution. assert.equal(finalSubagent?.runId, "run-attempt-claude-wake-8c"); assert.isFalse(yield* harness.hasPendingBackgroundWork); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index c06c11e7c36..8e771204f12 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -32,6 +32,8 @@ import { type OrchestrationV2ProviderTurn, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivation, + type OrchestrationV2SubagentUsage, type OrchestrationV2TurnItem, type OrchestrationV2WebSearchResult, type ProviderApprovalDecision, @@ -111,6 +113,13 @@ import { makeSubagentConversationArtifacts, subagentThreadTitle, } from "../SubagentProjection.ts"; +import { + accumulateCumulativeSubagentUsage, + appendSubagentActivity, + mergeCumulativeSubagentUsage, + providerSubagentRole, + subagentActivationId, +} from "../SubagentObservability.ts"; export const CLAUDE_PROVIDER = ProviderDriverKind.make("claudeAgent"); export const CLAUDE_AGENT_SDK_QUERY_PROTOCOL = "claude-agent-sdk.query" as const; @@ -1401,6 +1410,99 @@ function claudeTaskTypeFromSdkMessage(message: SDKMessage): string | null { return typeof taskType === "string" ? taskType : null; } +const nonNegativeInteger = (value: unknown) => + typeof value === "number" && Number.isFinite(value) && value >= 0 ? Math.round(value) : undefined; + +const recordField = (value: unknown, key: string) => + value !== null && typeof value === "object" && !Array.isArray(value) + ? Reflect.get(value, key) + : undefined; + +const firstRecordField = (value: unknown, keys: ReadonlyArray) => { + for (const key of keys) { + const field = recordField(value, key); + if (field !== undefined) return field; + } + return undefined; +}; + +const nonNegativeRecordField = (value: unknown, keys: ReadonlyArray) => + nonNegativeInteger(firstRecordField(value, keys)); + +function claudeSubagentUsage(value: unknown): OrchestrationV2SubagentUsage | undefined { + if (value === null || typeof value !== "object" || Array.isArray(value)) return undefined; + const inputTokens = nonNegativeInteger(recordField(value, "input_tokens")); + const cacheCreationTokens = nonNegativeInteger(recordField(value, "cache_creation_input_tokens")); + const cachedInputTokens = nonNegativeInteger(recordField(value, "cache_read_input_tokens")); + const outputTokens = nonNegativeInteger(recordField(value, "output_tokens")); + const totalTokens = + nonNegativeInteger(recordField(value, "total_tokens")) ?? + nonNegativeInteger(recordField(value, "totalTokens")) ?? + (inputTokens ?? 0) + + (cacheCreationTokens ?? 0) + + (cachedInputTokens ?? 0) + + (outputTokens ?? 0); + const toolUses = + nonNegativeInteger(recordField(value, "tool_uses")) ?? + nonNegativeInteger(recordField(value, "totalToolUseCount")); + const durationMs = + nonNegativeInteger(recordField(value, "duration_ms")) ?? + nonNegativeInteger(recordField(value, "totalDurationMs")); + if ( + inputTokens === undefined && + cacheCreationTokens === undefined && + cachedInputTokens === undefined && + outputTokens === undefined && + nonNegativeInteger(recordField(value, "total_tokens")) === undefined && + nonNegativeInteger(recordField(value, "totalTokens")) === undefined && + toolUses === undefined && + durationMs === undefined + ) { + return undefined; + } + return { + totalTokens, + ...(inputTokens === undefined && cacheCreationTokens === undefined + ? {} + : { inputTokens: (inputTokens ?? 0) + (cacheCreationTokens ?? 0) }), + ...(cachedInputTokens === undefined ? {} : { cachedInputTokens }), + ...(outputTokens === undefined ? {} : { outputTokens }), + ...(toolUses === undefined ? {} : { toolUses }), + ...(durationMs === undefined ? {} : { durationMs }), + }; +} + +const claudeWorkflowAgentStatus = (entry: unknown) => { + const state = firstRecordField(entry, ["state", "status"]); + if (typeof state !== "string") return null; + switch (state.toLowerCase()) { + case "start": + return nonNegativeRecordField(entry, ["startedAt", "started_at"]) === undefined + ? ("pending" as const) + : ("running" as const); + case "pending": + case "queued": + return "pending" as const; + case "active": + case "in_progress": + case "running": + case "working": + return "running" as const; + case "blocked": + case "waiting": + return "waiting" as const; + case "completed": + case "done": + case "finished": + return "completed" as const; + case "error": + case "failed": + return "failed" as const; + default: + return null; + } +}; + function isClaudeNonSubagentTask(message: SDKMessage): boolean { return claudeTaskTypeFromSdkMessage(message) === "local_bash"; } @@ -1910,6 +2012,7 @@ interface ActiveClaudeTurnContext { interface ActiveClaudeSubagent { task: OrchestrationV2Subagent; + activation: OrchestrationV2SubagentActivation; readonly childThreadId: ThreadId; readonly childRootNodeId: OrchestrationV2ExecutionNode["id"]; readonly turnItemId: OrchestrationV2TurnItem["id"]; @@ -1918,6 +2021,13 @@ interface ActiveClaudeSubagent { progressItemOrdinal: number | null; progressStartedAt: DateTime.Utc | null; resultItemOrdinal: number | null; + /** + * Set when the wake buffer pre-opens a settled entry so a raced resume still + * counts as wake evidence. The drained task_started then no longer sees the + * terminal status it needs to recognise a resume, so this carries that fact + * across and keeps the resume on its own activation. + */ + resumePending: boolean; } interface ClaudeLiveQueryContext { @@ -2218,11 +2328,20 @@ export function makeClaudeAdapterV2( readonly toolUseId?: string; readonly prompt?: string; readonly title?: string; + readonly agentType?: string; + readonly roleFallback?: string; + readonly model?: string; + readonly kind?: OrchestrationV2Subagent["kind"]; + readonly workflow?: OrchestrationV2Subagent["workflow"]; + readonly workflowMembership?: OrchestrationV2Subagent["workflowMembership"]; + readonly parentNodeId?: OrchestrationV2ExecutionNode["id"]; + readonly allowCreateSettled?: boolean; readonly progress?: string; readonly result?: string; + readonly usage?: OrchestrationV2SubagentUsage; readonly status: Extract< - OrchestrationV2ExecutionNode["status"], - "running" | "completed" | "failed" | "cancelled" + OrchestrationV2Subagent["status"], + "pending" | "running" | "waiting" | "completed" | "failed" | "cancelled" >; readonly reopen?: boolean; }) { @@ -2234,7 +2353,11 @@ export function makeClaudeAdapterV2( ? undefined : input.context.subagentsByToolUseId.get(input.toolUseId)) ?? (yield* Ref.get(sessionSubagentsByTaskId)).get(input.taskId); - if (existingSubagent === undefined && input.status !== "running") { + if ( + existingSubagent === undefined && + input.status !== "running" && + input.allowCreateSettled !== true + ) { return; } // Status is monotone with one exception: task_started for a known @@ -2242,17 +2365,24 @@ export function makeClaudeAdapterV2( // completed subagent resumes it and re-emits task_started with the // same id), so it may re-open a terminal entry. A late or // out-of-order task_progress must not. - const isReopen = - input.reopen === true && - existingSubagent !== undefined && - existingSubagent.task.status !== "running" && - input.status === "running"; - if ( + const previousSettled = existingSubagent !== undefined && - existingSubagent.task.status !== "running" && - input.status === "running" && - !isReopen - ) { + (existingSubagent.task.status === "completed" || + existingSubagent.task.status === "failed" || + existingSubagent.task.status === "cancelled"); + // The wake buffer may already have pre-opened a settled entry, hiding + // the terminal status this check relies on. Treat that as settled too, + // or the resume silently reuses the finished activation: its ordinal + // stops matching reality and its usage is measured against the + // previous activation's totals, which clamps the delta to zero. + const resumePending = existingSubagent?.resumePending === true; + const wasSettled = previousSettled || resumePending; + // "waiting" reopens a settled entry just as readily as an active + // start, so a replayed blocked-worker snapshot must be refused too. + const activeStart = + input.status === "pending" || input.status === "running" || input.status === "waiting"; + const isReopen = input.reopen === true && wasSettled && activeStart; + if (existingSubagent !== undefined && wasSettled && activeStart && !isReopen) { return; } const lifecycleChanged = @@ -2262,9 +2392,13 @@ export function makeClaudeAdapterV2( // already pre-opened to running by bufferWakeMessage while the // projection node still holds the old terminal status; re-emit // the node lifecycle for authoritative task_started updates. - (input.reopen === true && input.status === "running"); + (input.reopen === true && activeStart); const now = yield* DateTime.now; + const settled = + input.status === "completed" || + input.status === "failed" || + input.status === "cancelled"; const nativeItemId = `task:${input.taskId}`; const nodeId = existingSubagent?.task.id ?? @@ -2301,12 +2435,13 @@ export function makeClaudeAdapterV2( existingSubagent.task, ) : existingSubagent.task; + const startsActivation = existingSubagent === undefined || isReopen; const task = { ...(priorTask ?? { id: nodeId, threadId: input.context.input.threadId, runId: input.context.input.runId, - parentNodeId: input.context.input.rootNodeId, + parentNodeId: input.parentNodeId ?? input.context.input.rootNodeId, origin: "provider_native" as const, createdBy: "agent" as const, driver: CLAUDE_PROVIDER, @@ -2320,16 +2455,16 @@ export function makeClaudeAdapterV2( }, prompt: input.prompt ?? "", title: input.title ?? null, - model: input.context.input.modelSelection.model, - kind: "subagent" as const, - role: { name: "general-purpose", source: "app_default" as const }, + model: input.model ?? input.context.input.modelSelection.model, + kind: input.kind ?? ("subagent" as const), + role: providerSubagentRole(input.agentType, input.roleFallback), + result: null, usage: null, currentActivationId: null, - activationCount: 1, + activationCount: 0, workflow: null, workflowMembership: null, recentActivity: [], - result: null, startedAt: now, }), status: input.status, @@ -2340,20 +2475,66 @@ export function makeClaudeAdapterV2( // subagents terminalize); attribution also enrolls the subagent // in the resuming run's active-child tracking so its fiber // outlives settle until the resumed task completes. - ...(input.reopen === true && - input.status === "running" && - existingSubagent !== undefined + ...(input.reopen === true && activeStart && existingSubagent !== undefined ? { runId: input.context.input.runId } : {}), ...(input.prompt === undefined ? {} : { prompt: input.prompt }), ...(input.title === undefined ? {} : { title: input.title }), + ...(input.model === undefined ? {} : { model: input.model }), + ...(input.kind === undefined ? {} : { kind: input.kind }), + ...(input.agentType === undefined + ? {} + : { role: providerSubagentRole(input.agentType, input.roleFallback) }), + ...(input.workflow === undefined ? {} : { workflow: input.workflow }), + ...(input.workflowMembership === undefined + ? {} + : { workflowMembership: input.workflowMembership }), ...(input.progress === undefined ? {} : { progress: input.progress }), ...(input.result === undefined ? {} : { result: input.result }), - completedAt: input.status === "running" ? null : now, + usage: accumulateCumulativeSubagentUsage( + priorTask?.usage ?? null, + startsActivation ? undefined : existingSubagent?.activation.usage, + input.usage, + ), + activationCount: + existingSubagent === undefined || isReopen + ? (priorTask?.activationCount ?? 0) + 1 + : (priorTask?.activationCount ?? 1), + recentActivity: appendSubagentActivity( + priorTask?.recentActivity ?? [], + input.progress, + now, + ), + // Keep the first completion time: a duplicate terminal frame is not + // a second completion, and re-stamping drifts the recorded end. + completedAt: settled ? (priorTask?.completedAt ?? now) : null, updatedAt: now, } satisfies OrchestrationV2Subagent; + const activation = startsActivation + ? ({ + id: subagentActivationId(nodeId, task.activationCount), + threadId: task.threadId, + subagentId: nodeId, + runId: task.runId, + providerTurnId: input.context.providerTurnId, + ordinal: task.activationCount, + status: input.status, + usage: input.usage ?? null, + startedAt: now, + completedAt: settled ? now : null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation) + : ({ + ...existingSubagent.activation, + status: input.status, + usage: mergeCumulativeSubagentUsage(existingSubagent.activation.usage, input.usage), + completedAt: settled ? now : null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation); + task.currentActivationId = settled ? null : activation.id; const subagent = { task, + activation, childThreadId, childRootNodeId, turnItemId: @@ -2367,6 +2548,8 @@ export function makeClaudeAdapterV2( progressItemOrdinal: existingSubagent?.progressItemOrdinal ?? null, progressStartedAt: existingSubagent?.progressStartedAt ?? null, resultItemOrdinal: existingSubagent?.resultItemOrdinal ?? null, + // Consumed once the reopen it was recording has been applied. + resumePending: isReopen ? false : (existingSubagent?.resumePending ?? false), } satisfies ActiveClaudeSubagent; input.context.subagentsByTaskId.set(input.taskId, subagent); if (input.toolUseId !== undefined) { @@ -2384,7 +2567,7 @@ export function makeClaudeAdapterV2( if ( registered !== undefined && registered.task.status !== "running" && - input.status === "running" && + activeStart && !(isReopen && registered === existingSubagent) ) { return current; @@ -2443,7 +2626,7 @@ export function makeClaudeAdapterV2( runtimeRequestId: null, checkpointScopeId: null, startedAt: task.startedAt, - completedAt: input.status === "running" ? null : now, + completedAt: settled ? now : null, }, }); yield* emitProviderEvent({ @@ -2464,7 +2647,7 @@ export function makeClaudeAdapterV2( runtimeRequestId: null, checkpointScopeId: null, startedAt: task.startedAt, - completedAt: input.status === "running" ? null : now, + completedAt: settled ? now : null, }, }); } @@ -2509,6 +2692,11 @@ export function makeClaudeAdapterV2( driver: CLAUDE_PROVIDER, subagent: task, }); + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CLAUDE_PROVIDER, + activation, + }); yield* emitProviderEvent({ type: "turn_item.updated", driver: CLAUDE_PROVIDER, @@ -3055,6 +3243,7 @@ export function makeClaudeAdapterV2( const { progress: _staleProgress, ...priorTask } = registered.task; return new Map(current).set(message.task_id, { ...registered, + resumePending: true, task: { ...priorTask, status: "running", @@ -3159,12 +3348,16 @@ export function makeClaudeAdapterV2( new Set(current).add(message.task_id), ); } else { + const agentType = recordField(message, "subagent_type"); + const isWorkflow = message.task_type === "local_workflow"; yield* updateClaudeSubagentNode({ context, taskId: message.task_id, ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), ...(message.prompt === undefined ? {} : { prompt: message.prompt }), title: message.description, + ...(typeof agentType === "string" ? { agentType } : {}), + ...(isWorkflow ? { kind: "workflow", roleFallback: "workflow-coordinator" } : {}), status: "running", reopen: true, }); @@ -3173,22 +3366,115 @@ export function makeClaudeAdapterV2( if (message.type === "system" && message.subtype === "task_progress") { const progress = message.description.trim(); + const usage = claudeSubagentUsage(message.usage); + const agentType = recordField(message, "subagent_type"); + const workflowProgress = recordField(message, "workflow_progress"); + const workflowEntries = Array.isArray(workflowProgress) ? workflowProgress : []; + const phases = workflowEntries.flatMap((entry) => { + const index = nonNegativeRecordField(entry, ["index", "phaseIndex", "phase_index"]); + const title = firstRecordField(entry, ["title", "name"]); + return recordField(entry, "type") === "workflow_phase" && + index !== undefined && + typeof title === "string" && + title.trim() + ? [{ index, title: title.trim() }] + : []; + }); const isBackgroundTask = (yield* Ref.get(pendingBackgroundTaskIds)).has( message.task_id, ); + const shouldTrackSubagent = + !context.ignoredTaskIds.has(message.task_id) && !isBackgroundTask; if ( - progress.length > 0 && - !context.ignoredTaskIds.has(message.task_id) && - !isBackgroundTask + (progress.length > 0 || usage !== undefined || workflowEntries.length > 0) && + shouldTrackSubagent ) { yield* updateClaudeSubagentNode({ context, taskId: message.task_id, ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), - progress, + ...(progress.length === 0 ? {} : { progress }), + ...(usage === undefined ? {} : { usage }), + ...(typeof agentType === "string" ? { agentType } : {}), + ...(phases.length === 0 + ? {} + : { + kind: "workflow", + roleFallback: "workflow-coordinator", + workflow: { phases }, + }), status: "running", }); } + + const workflowParent = shouldTrackSubagent + ? (context.subagentsByTaskId.get(message.task_id) ?? + (message.tool_use_id === undefined + ? undefined + : context.subagentsByToolUseId.get(message.tool_use_id)) ?? + (yield* Ref.get(sessionSubagentsByTaskId)).get(message.task_id)) + : undefined; + if (workflowParent !== undefined) { + for (const entry of workflowEntries) { + if (recordField(entry, "type") !== "workflow_agent") continue; + const index = nonNegativeInteger(recordField(entry, "index")); + const label = firstRecordField(entry, ["label", "name"]); + if (index === undefined || typeof label !== "string" || !label.trim()) continue; + const status = claudeWorkflowAgentStatus(entry); + if (status === null) continue; + const phaseIndex = nonNegativeRecordField(entry, ["phaseIndex", "phase_index"]); + const attempt = (nonNegativeInteger(recordField(entry, "attempt")) ?? 0) + 1; + const model = recordField(entry, "model"); + const lastToolName = firstRecordField(entry, ["lastToolName", "last_tool_name"]); + const tokens = nonNegativeRecordField(entry, ["tokens", "total_tokens"]); + const toolUses = nonNegativeRecordField(entry, [ + "toolCalls", + "tool_calls", + "tool_uses", + ]); + const childUsage = + tokens === undefined + ? undefined + : ({ + totalTokens: tokens, + ...(toolUses === undefined ? {} : { toolUses }), + } satisfies OrchestrationV2SubagentUsage); + const workerTaskId = `${message.task_id}:workflow:${index}`; + const existingWorker = + context.subagentsByTaskId.get(workerTaskId) ?? + (yield* Ref.get(sessionSubagentsByTaskId)).get(workerTaskId); + const activeWorker = status === "pending" || status === "running"; + const reopen = + activeWorker && + existingWorker !== undefined && + attempt > (existingWorker.task.workflowMembership?.attempt ?? 0); + yield* updateClaudeSubagentNode({ + context, + taskId: workerTaskId, + title: label.trim(), + parentNodeId: workflowParent.task.id, + kind: "workflow_agent", + roleFallback: "workflow-worker", + ...(typeof model === "string" && model.trim() ? { model: model.trim() } : {}), + workflowMembership: { + workflowSubagentId: workflowParent.task.id, + agentIndex: index, + phaseIndex: phaseIndex ?? null, + attempt, + }, + allowCreateSettled: true, + ...(typeof lastToolName === "string" && lastToolName.trim() + ? { progress: lastToolName.trim() } + : {}), + ...(childUsage === undefined ? {} : { usage: childUsage }), + ...(typeof recordField(entry, "error") === "string" + ? { result: String(recordField(entry, "error")) } + : {}), + status, + reopen, + }); + } + } } if (message.type === "system" && message.subtype === "task_notification") { @@ -3196,11 +3482,13 @@ export function makeClaudeAdapterV2( // background registry is the durable ignore signal across turns. const wasBackgroundTask = yield* clearPendingBackgroundTask(message.task_id); if (!wasBackgroundTask && !context.ignoredTaskIds.has(message.task_id)) { + const usage = claudeSubagentUsage(message.usage); yield* updateClaudeSubagentNode({ context, taskId: message.task_id, ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), result: message.summary, + ...(usage === undefined ? {} : { usage }), status: message.status === "completed" ? "completed" diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index f4b85199796..c3768fffcb7 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -49,6 +49,7 @@ import { CODEX_DEFAULT_INSTANCE_ID, CODEX_DRIVER_KIND, codexBackgroundCommandDetail, + codexCollabAgentStatus, codexThreadRuntimeParams, type CodexAgentMessageDeltaUpdate, type CodexAppServerClientFactoryShape, @@ -61,6 +62,17 @@ import { } from "./CodexAdapterV2.ts"; import { makeReplayServerConfig } from "./CodexAdapterV2.testkit.ts"; +describe("CodexAdapterV2 subagent status mapping", () => { + it("maps every provider state without treating terminal states as running", () => { + assert.deepEqual( + ["pendingInit", "running", "completed", "interrupted", "errored", "shutdown", "notFound"].map( + (status) => codexCollabAgentStatus(status as Parameters[0]), + ), + ["pending", "running", "idle", "interrupted", "failed", "cancelled", "failed"], + ); + }); +}); + describe("CodexAdapterV2 assistant message streaming", () => { it.effect("makes accumulated assistant text visible after the bounded flush interval", () => Effect.gen(function* () { @@ -986,6 +998,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { (event): event is Extract => event.type === "subagent.updated", ); + const subagentActivationUpdates = () => + events.filter( + ( + event, + ): event is Extract => + event.type === "subagent_activation.updated", + ); return { runtime, providerThread, @@ -994,6 +1013,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { continuationRequests, terminalEvents, subagentUpdates, + subagentActivationUpdates, hasPendingBackgroundWork, }; }); @@ -3184,6 +3204,38 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }); + const childTokenUsageUpdated = ( + turnId: string, + totalTokens: number, + lastTokens: number, + ): CodexReplay.CodexAppServerReplayEntry => ({ + type: "emit_inbound", + label: `thread/tokenUsage/updated/${turnId}`, + frame: { + method: "thread/tokenUsage/updated", + params: { + threadId: RESUME_CHILD_THREAD, + turnId, + tokenUsage: { + total: { + totalTokens, + inputTokens: totalTokens - 10, + cachedInputTokens: 0, + outputTokens: 10, + reasoningOutputTokens: 0, + }, + last: { + totalTokens: lastTokens, + inputTokens: lastTokens - 10, + cachedInputTokens: 0, + outputTokens: 10, + reasoningOutputTokens: 0, + }, + }, + }, + }, + }); + const resumeSubagentTranscript = makeCodexReplayTranscript({ scenario: RESUME_SCENARIO, entries: [ @@ -3232,6 +3284,18 @@ describe("CodexAdapterV2 post-settle continuation", () => { completedAtMs: 1782622442002, }), childTurnCompleted(RESUME_CHILD_TURN_1, 100), + childTokenUsageUpdated(RESUME_CHILD_TURN_1, 100, 100), + { + type: "emit_inbound", + label: "thread/status/changed/late-active", + frame: { + method: "thread/status/changed", + params: { + threadId: RESUME_CHILD_THREAD, + status: { type: "active", activeFlags: [] }, + }, + }, + }, { type: "emit_inbound", label: "item/completed/root-answer", @@ -3271,6 +3335,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { afterMs: 30_000, }), childTurnCompleted(RESUME_CHILD_TURN_2), + childTokenUsageUpdated(RESUME_CHILD_TURN_2, 180, 80), ], }); @@ -3303,8 +3368,20 @@ describe("CodexAdapterV2 post-settle continuation", () => { assert.equal(harness.terminalEvents()[0]?.status, "completed"); const settledUpdates = harness.subagentUpdates(); const firstCompletion = settledUpdates[settledUpdates.length - 1]; - assert.equal(firstCompletion?.subagent.status, "completed"); + assert.equal(firstCompletion?.subagent.status, "idle"); assert.equal(firstCompletion?.subagent.result, "CODEX_FIRST_DONE"); + assert.equal(firstCompletion?.subagent.activationCount, 1); + assert.equal(firstCompletion?.subagent.usage?.totalTokens, 100); + assert.isNull(firstCompletion?.subagent.currentActivationId); + assert.lengthOf( + new Set( + settledUpdates + .filter((event) => event.subagent.status === "idle") + .map((event) => event.subagent.completedAt), + ), + 1, + "post-idle usage updates preserve the completion timestamp", + ); assert.isFalse(yield* harness.hasPendingBackgroundWork); const settledUpdateCount = settledUpdates.length; @@ -3323,10 +3400,31 @@ describe("CodexAdapterV2 post-settle continuation", () => { const latest = updates[updates.length - 1]; return ( latest !== undefined && - latest.subagent.status === "completed" && + latest.subagent.status === "idle" && latest.subagent.result === "CODEX_RESUME_DONE" ); }, "resumed subagent completion"); + const finalSubagent = harness.subagentUpdates().at(-1)?.subagent; + assert.equal(finalSubagent?.activationCount, 2); + assert.equal(finalSubagent?.usage?.totalTokens, 180); + assert.isNull(finalSubagent?.currentActivationId); + const activations = harness.subagentActivationUpdates(); + assert.lengthOf(new Set(activations.map((event) => event.activation.id)), 2); + assert.sameMembers( + [...new Set(activations.map((event) => event.activation.ordinal))], + [1, 2], + ); + assert.deepEqual( + [ + ...new Set( + activations.map((event) => + event.activation.providerTurnId?.replace(/^.*native-turn:/, ""), + ), + ), + ], + [RESUME_CHILD_TURN_1, RESUME_CHILD_TURN_2], + ); + assert.equal(activations.at(-1)?.activation.status, "completed"); assert.isFalse(yield* harness.hasPendingBackgroundWork); assert.lengthOf(harness.terminalEvents(), 1); assert.lengthOf(harness.continuationRequests, 0); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 39b5a27b381..9c31a19ece4 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -21,6 +21,8 @@ import type { OrchestrationV2PlanStep, OrchestrationV2RuntimeRequest, OrchestrationV2Subagent, + OrchestrationV2SubagentActivation, + OrchestrationV2SubagentUsage, OrchestrationV2TurnItem, ProviderUserInputAnswers, ProviderApprovalDecision, @@ -105,13 +107,24 @@ import { makeSubagentConversationArtifacts, subagentThreadTitle, } from "../SubagentProjection.ts"; -import { defaultSubagentRole } from "../SubagentObservability.ts"; +import { + appendSubagentActivity, + defaultSubagentRole, + mergeCumulativeSubagentUsage, + subagentActivationId, +} from "../SubagentObservability.ts"; const CODEX_PROVIDER = ProviderDriverKind.make("codex"); export const CODEX_DRIVER_KIND = CODEX_PROVIDER; export const CODEX_DEFAULT_INSTANCE_ID = defaultInstanceIdForDriver(CODEX_DRIVER_KIND); const DEFAULT_CODEX_SETTINGS = Schema.decodeSync(CodexSettings)({}); const CODEX_ASSISTANT_DELTA_FLUSH_INTERVAL_MS = 50; +const isSettledCodexSubagent = (subagent: OrchestrationV2Subagent) => + subagent.status === "idle" || + subagent.status === "completed" || + subagent.status === "failed" || + subagent.status === "cancelled" || + subagent.status === "interrupted"; const CodexBackgroundTerminalTerminateResponse = Schema.Struct({ terminated: Schema.Boolean, }); @@ -807,6 +820,7 @@ interface ActiveCodexTurnContext { readonly rootNodeId: OrchestrationV2ExecutionNode["id"]; readonly subagent: CodexSubagentThreadContext | null; readonly startedAt: DateTime.Utc; + subagentActivation: OrchestrationV2SubagentActivation | null; } /** Snapshot of a still-running commandExecution item for interrupt/fail terminalization. */ @@ -891,6 +905,27 @@ type CodexCollabAgentToolCallItem = Extract< CodexSchema.V2ItemCompletedNotification__ThreadItem, { readonly type: "collabAgentToolCall" } >; +type CodexCollabAgentStatus = CodexCollabAgentToolCallItem["agentsStates"][string]["status"]; + +export function codexCollabAgentStatus( + status: CodexCollabAgentStatus, +): OrchestrationV2Subagent["status"] { + switch (status) { + case "pendingInit": + return "pending"; + case "running": + return "running"; + case "completed": + return "idle"; + case "interrupted": + return "interrupted"; + case "errored": + case "notFound": + return "failed"; + case "shutdown": + return "cancelled"; + } +} type CodexSubAgentActivityItem = Extract< | CodexSchema.V2ItemStartedNotification__ThreadItem @@ -898,6 +933,22 @@ type CodexSubAgentActivityItem = Extract< { readonly type: "subAgentActivity" } >; +const CODEX_SUBAGENT_ACTIVITY_LABELS: Readonly> = { + commandExecution: "Running command", + fileChange: "Editing files", + mcpToolCall: "Using MCP tool", + dynamicToolCall: "Using tool", + webSearch: "Searching the web", + reasoning: "Reasoning", + plan: "Updating plan", + todoList: "Updating tasks", + collabAgentToolCall: "Coordinating agents", + subAgentActivity: "Coordinating subagent", +}; + +const codexSubagentActivitySummary = (item: { readonly type: string }) => + CODEX_SUBAGENT_ACTIVITY_LABELS[item.type] ?? null; + export interface CodexAgentMessageDeltaUpdate { readonly turnId: string; readonly itemId: string; @@ -1481,6 +1532,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi rootNodeId: input.turnInput.rootNodeId, subagent: null, startedAt: input.startedAt, + subagentActivation: null, }; yield* Ref.update(activeTurns, (current) => { const updated = new Map(current); @@ -1690,20 +1742,42 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi readonly subagent: CodexSubagentThreadContext; readonly status: OrchestrationV2Subagent["status"]; readonly result?: string | null; + readonly progress?: string; + readonly activity?: string; + readonly usage?: OrchestrationV2SubagentUsage; + readonly currentActivationId?: OrchestrationV2Subagent["currentActivationId"]; + readonly activationCount?: number; readonly completedAt?: DateTime.Utc | null; }) => Effect.gen(function* () { const now = yield* DateTime.now; const terminal = + input.status === "idle" || input.status === "completed" || input.status === "failed" || input.status === "cancelled" || input.status === "interrupted"; - const completedAt = terminal ? (input.completedAt ?? now) : null; + const completedAt = terminal + ? (input.completedAt ?? input.subagent.task.completedAt ?? now) + : null; const task = { ...input.subagent.task, status: input.status, + ...(input.progress === undefined ? {} : { progress: input.progress }), result: input.result === undefined ? input.subagent.task.result : input.result, + usage: mergeCumulativeSubagentUsage(input.subagent.task.usage, input.usage), + currentActivationId: + input.currentActivationId !== undefined + ? input.currentActivationId + : terminal + ? null + : input.subagent.task.currentActivationId, + activationCount: input.activationCount ?? input.subagent.task.activationCount, + recentActivity: appendSubagentActivity( + input.subagent.task.recentActivity, + input.activity, + now, + ), completedAt, updatedAt: now, } satisfies OrchestrationV2Subagent; @@ -1814,19 +1888,40 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi rootNodeId: providerNodeId, subagent, startedAt: turn.startedAt, + subagentActivation: null, }; + const activationOrdinal = subagent.task.activationCount + 1; + const activation = { + id: subagentActivationId(subagent.task.id, activationOrdinal), + threadId: subagent.task.threadId, + subagentId: subagent.task.id, + runId: subagent.parentContext.projectionRunId, + providerTurnId, + ordinal: activationOrdinal, + status: "running", + usage: null, + startedAt: turn.startedAt, + completedAt: null, + updatedAt: turn.startedAt, + } satisfies OrchestrationV2SubagentActivation; + activeContext.subagentActivation = activation; yield* Ref.update(activeTurns, (current) => { const updated = new Map(current); updated.set(turn.nativeTurnId, activeContext); return updated; }); - if (providerTurnOrdinal > 1 && subagent.task.status !== "running") { - yield* emitSubagentTaskUpdate({ - subagent, - status: "running", - result: null, - }); - } + yield* emitSubagentTaskUpdate({ + subagent, + status: "running", + result: providerTurnOrdinal > 1 ? null : subagent.task.result, + currentActivationId: activation.id, + activationCount: activationOrdinal, + }); + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CODEX_PROVIDER, + activation, + }); const now = yield* DateTime.now; yield* emitProviderEvent({ type: "provider_thread.updated", @@ -1976,11 +2071,11 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi model: input.model, kind: "subagent", role: defaultSubagentRole(), - status: "running", + status: "pending", result: null, usage: null, currentActivationId: null, - activationCount: 1, + activationCount: 0, workflow: null, workflowMembership: null, recentActivity: [], @@ -2177,7 +2272,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi nativeToolCallId: input.item.id, prompt: "", title: input.item.agentPath, - model: null, + // A subAgentActivity frame carries no model of its own. The + // agent runs on the turn's selection, so recording null here + // left every Codex-native subagent without a model. + model: input.context.input.modelSelection.model, ordinal, emitInitialPrompt: false, }); @@ -2205,18 +2303,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi if (subagent === undefined) { continue; } - const nativeStatus = String(state.status); - const status: OrchestrationV2Subagent["status"] = - nativeStatus === "completed" - ? "completed" - : nativeStatus === "failed" || nativeStatus === "errored" - ? "failed" - : nativeStatus === "cancelled" || nativeStatus === "closed" - ? "cancelled" - : "running"; yield* emitSubagentTaskUpdate({ subagent, - status, + status: codexCollabAgentStatus(state.status), ...(state.message === null ? {} : { result: state.message }), }); } @@ -3150,12 +3239,107 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)), ); + yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) => + Effect.gen(function* () { + const subagent = (yield* Ref.get(subagentThreads)).get(payload.threadId); + if (subagent === undefined || payload.tokenUsage.total.totalTokens <= 0) { + return; + } + const total = payload.tokenUsage.total; + const usage = { + totalTokens: total.totalTokens, + inputTokens: total.inputTokens, + cachedInputTokens: total.cachedInputTokens, + outputTokens: total.outputTokens, + reasoningOutputTokens: total.reasoningOutputTokens, + } satisfies OrchestrationV2SubagentUsage; + yield* emitSubagentTaskUpdate({ + subagent, + status: subagent.task.status, + usage, + }); + + const activeContext = (yield* Ref.get(activeTurns)).get(payload.turnId); + if ( + activeContext === undefined || + activeContext.subagent !== subagent || + activeContext.subagentActivation === null + ) { + return; + } + const last = payload.tokenUsage.last; + const activation = { + ...activeContext.subagentActivation, + usage: { + totalTokens: last.totalTokens, + inputTokens: last.inputTokens, + cachedInputTokens: last.cachedInputTokens, + outputTokens: last.outputTokens, + reasoningOutputTokens: last.reasoningOutputTokens, + }, + updatedAt: yield* DateTime.now, + } satisfies OrchestrationV2SubagentActivation; + activeContext.subagentActivation = activation; + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CODEX_PROVIDER, + activation, + }); + }).pipe(Effect.orDie), + ); + + yield* client.handleServerNotification("thread/status/changed", (payload) => + Effect.gen(function* () { + const subagent = (yield* Ref.get(subagentThreads)).get(payload.threadId); + if (subagent === undefined || isSettledCodexSubagent(subagent.task)) return; + const status = + payload.status.type === "active" + ? payload.status.activeFlags.length > 0 + ? ("waiting" as const) + : ("running" as const) + : payload.status.type === "systemError" + ? ("failed" as const) + : null; + if (status === null) return; + yield* emitSubagentTaskUpdate({ subagent, status }); + + const activeContext = Array.from((yield* Ref.get(activeTurns)).values()).find( + (context) => context.subagent === subagent, + ); + if (activeContext === undefined || activeContext.subagentActivation === null) return; + const now = yield* DateTime.now; + const activation = { + ...activeContext.subagentActivation, + status, + completedAt: status === "failed" ? now : null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation; + activeContext.subagentActivation = activation; + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CODEX_PROVIDER, + activation, + }); + }).pipe(Effect.orDie), + ); + yield* client.handleServerNotification("item/started", (payload) => Effect.gen(function* () { const context = yield* awaitActiveTurn(payload.turnId); if (context === undefined) { return; } + if (context.subagent !== null) { + const activity = codexSubagentActivitySummary(payload.item); + if (activity !== null) { + yield* emitSubagentTaskUpdate({ + subagent: context.subagent, + status: "running", + progress: activity, + activity, + }); + } + } if (payload.item.type === "userMessage") { if (yield* emitSubagentUserMessage(context, payload.item)) { @@ -4036,9 +4220,23 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi completedAt: input.completedAt, }, }); + if (input.context.subagentActivation !== null) { + const activation = { + ...input.context.subagentActivation, + status: input.status, + completedAt: input.completedAt, + updatedAt: input.completedAt, + } satisfies OrchestrationV2SubagentActivation; + input.context.subagentActivation = activation; + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CODEX_PROVIDER, + activation, + }); + } yield* emitSubagentTaskUpdate({ subagent: input.context.subagent, - status: input.status, + status: input.status === "completed" ? "idle" : input.status, completedAt: input.completedAt, }); } diff --git a/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts index 01b144a0ab5..1162265631f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts @@ -23,6 +23,7 @@ import { type OrchestrationV2ProviderThread, type OrchestrationV2ProviderTurn, type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivation, type OrchestrationV2TurnItem, type ProviderInstanceId, type ThreadId, @@ -79,6 +80,7 @@ import { makeSubagentConversationArtifacts, subagentThreadTitle, } from "../SubagentProjection.ts"; +import { defaultSubagentRole, subagentActivationId } from "../SubagentObservability.ts"; import { CURSOR_PROVIDER, CursorAgentSdkRunner, @@ -1477,6 +1479,7 @@ export function makeCursorAdapterV2( driver: CURSOR_PROVIDER, nativeThreadId: `${input.context.run.runId}:task:${input.callId}`, }); + const activationId = subagentActivationId(nodeId, 1); const task: OrchestrationV2Subagent = { ...(existing?.task ?? { id: nodeId, @@ -1497,15 +1500,15 @@ export function makeCursorAdapterV2( prompt: args.prompt, title: args.description, model: args.model ?? input.context.input.modelSelection.model, - kind: "subagent", - role: { name: "general-purpose", source: "app_default" }, + kind: "subagent" as const, + role: defaultSubagentRole(), + result: null, usage: null, - currentActivationId: null, + currentActivationId: activationId, activationCount: 1, workflow: null, workflowMembership: null, recentActivity: [], - result: null, startedAt: now, }), nativeTaskRef: { @@ -1515,6 +1518,7 @@ export function makeCursorAdapterV2( }, status, result: resultText.length === 0 ? (existing?.task.result ?? null) : resultText, + currentActivationId: input.completed ? null : activationId, completedAt: input.completed ? now : null, updatedAt: now, }; @@ -1641,6 +1645,23 @@ export function makeCursorAdapterV2( driver: CURSOR_PROVIDER, subagent: task, }); + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: CURSOR_PROVIDER, + activation: { + id: activationId, + threadId: task.threadId, + subagentId: task.id, + runId: task.runId, + providerTurnId: input.context.providerTurnId, + ordinal: 1, + status, + usage: null, + startedAt: task.startedAt, + completedAt: input.completed ? now : null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation, + }); yield* emitProviderEvent({ type: "turn_item.updated", driver: CURSOR_PROVIDER, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts index 8254b8ea1e3..bd15ffa1467 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts @@ -28,6 +28,7 @@ import { type OrchestrationV2ProviderTurn, type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivation, type OrchestrationV2TurnItem, OpenCodeSettings as OpenCodeSettingsSchema, type PlanId, @@ -101,6 +102,11 @@ import { type ProviderAdapterDriverCreateInput, } from "../ProviderAdapterDriver.ts"; import { makeSubagentChildThread, subagentThreadTitle } from "../SubagentProjection.ts"; +import { + appendSubagentActivity, + defaultSubagentRole, + subagentActivationId, +} from "../SubagentObservability.ts"; export const OPENCODE_PROVIDER = ProviderDriverKind.make("opencode"); export const OPENCODE_DRIVER_KIND = OPENCODE_PROVIDER; @@ -248,6 +254,7 @@ interface OpenCodeSubagentContext { childProviderThreadId: OrchestrationV2ProviderThread["id"] | null; model: string | null; result: string | null; + recentActivity: OrchestrationV2Subagent["recentActivity"]; } interface OpenCodeThreadState { @@ -1086,6 +1093,7 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid childProviderThreadId: null, model: null, result: null, + recentActivity: [], }; subagentsByNativeItemId.set(part.id, context); } @@ -1195,6 +1203,9 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid : status.item === "pending" ? "pending" : "running"; + const activationId = subagentActivationId(nodeId, 1); + const settled = subagentStatus === "completed" || subagentStatus === "failed"; + context.recentActivity = appendSubagentActivity(context.recentActivity, title, now); const subagent: OrchestrationV2Subagent = { id: nodeId, threadId: turn.threadId, @@ -1211,15 +1222,15 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid title, model: context.model, kind: "subagent", - role: { name: "general-purpose", source: "app_default" }, + role: defaultSubagentRole(), + status: subagentStatus, + result: context.result, usage: null, - currentActivationId: null, + currentActivationId: settled ? null : activationId, activationCount: 1, workflow: null, workflowMembership: null, - recentActivity: [], - status: subagentStatus, - result: context.result, + recentActivity: context.recentActivity, startedAt: context.startedAt, completedAt, updatedAt: now, @@ -1250,6 +1261,23 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid driver: OPENCODE_PROVIDER, subagent, }); + yield* emitProviderEvent({ + type: "subagent_activation.updated", + driver: OPENCODE_PROVIDER, + activation: { + id: activationId, + threadId: subagent.threadId, + subagentId: subagent.id, + runId: subagent.runId, + providerTurnId: turn.providerTurnId, + ordinal: 1, + status: subagentStatus, + usage: null, + startedAt: context.startedAt, + completedAt, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation, + }); yield* emitProviderEvent({ type: "turn_item.updated", driver: OPENCODE_PROVIDER, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 932beec8995..853b204cf62 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -265,6 +265,7 @@ it.effect("cancels a stale waiting run when no checkpoint capture can finish it" attempts: [], nodes: [], subagents: [], + subagentActivations: [], messages: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; @@ -343,6 +344,7 @@ it.effect("cancels accepted queued work instead of replaying it after restart", }, ], subagents: [], + subagentActivations: [], messages: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; @@ -451,6 +453,7 @@ it.effect( ], nodes: [{ id: rootNodeId, runId, status: "running" }], subagents: [], + subagentActivations: [], messages: [{ id: MessageId.make("message_recovery_cancel"), runId, streaming: true }], turnItems: [ { diff --git a/apps/server/src/orchestration-v2/SubagentObservability.test.ts b/apps/server/src/orchestration-v2/SubagentObservability.test.ts index e2d679dcdcf..b18141b7ea9 100644 --- a/apps/server/src/orchestration-v2/SubagentObservability.test.ts +++ b/apps/server/src/orchestration-v2/SubagentObservability.test.ts @@ -1,23 +1,132 @@ import { describe, expect, it } from "vite-plus/test"; -import { NodeId } from "@t3tools/contracts"; -import { defaultSubagentRole, subagentActivationId } from "./SubagentObservability.ts"; +import { + accumulateCumulativeSubagentUsage, + appendSubagentActivity, + defaultSubagentRole, + mergeCumulativeSubagentUsage, + providerSubagentRole, +} from "./SubagentObservability.ts"; +import * as DateTime from "effect/DateTime"; describe("subagent observability", () => { - it("marks a fallback role as app-provided so provider roles stay distinguishable", () => { - expect(defaultSubagentRole()).toEqual({ name: "general-purpose", source: "app_default" }); - expect(defaultSubagentRole("workflow-worker")).toEqual({ - name: "workflow-worker", - source: "app_default", + it("preserves role provenance instead of using titles as roles", () => { + expect(providerSubagentRole(" reviewer ")).toEqual({ + name: "reviewer", + source: "provider", }); + expect(providerSubagentRole(undefined)).toEqual(defaultSubagentRole()); }); - it("derives activation ids that are stable and unique per ordinal", () => { - const subagentId = NodeId.make("node:provider:codex:native-item:call_abc"); - expect(subagentActivationId(subagentId, 1)).toBe(`${subagentId}:activation:1`); - // Stable derivation means a replayed activation updates its own row rather - // than inserting a duplicate. - expect(subagentActivationId(subagentId, 2)).toBe(subagentActivationId(subagentId, 2)); - expect(subagentActivationId(subagentId, 1)).not.toBe(subagentActivationId(subagentId, 2)); + it("accumulates activation deltas into stable lifetime usage", () => { + const first = accumulateCumulativeSubagentUsage(null, null, { + totalTokens: 100, + inputTokens: 70, + outputTokens: 30, + toolUses: 2, + }); + const firstAdvanced = accumulateCumulativeSubagentUsage( + first, + { + totalTokens: 100, + inputTokens: 70, + outputTokens: 30, + toolUses: 2, + }, + { + totalTokens: 140, + inputTokens: 95, + outputTokens: 45, + toolUses: 3, + }, + ); + const resumed = accumulateCumulativeSubagentUsage(firstAdvanced, null, { + totalTokens: 80, + inputTokens: 50, + outputTokens: 30, + toolUses: 1, + }); + + expect(firstAdvanced).toEqual({ + totalTokens: 140, + inputTokens: 95, + outputTokens: 45, + toolUses: 3, + }); + expect(resumed).toEqual({ + totalTokens: 220, + inputTokens: 145, + outputTokens: 75, + toolUses: 4, + }); + }); + + it("bounds recent activity and deduplicates adjacent updates", () => { + const now = DateTime.makeUnsafe("2026-07-26T00:00:00.000Z"); + const activity = Array.from({ length: 8 }, (_, index) => `step ${index}`).reduce( + (current, summary) => appendSubagentActivity(current, summary, now), + appendSubagentActivity([], "step 0", now), + ); + + expect(activity).toHaveLength(6); + expect(activity.map((entry) => entry.summary)).toEqual([ + "step 2", + "step 3", + "step 4", + "step 5", + "step 6", + "step 7", + ]); + expect(appendSubagentActivity([], " ", now)).toEqual([]); + }); + + it("merges cumulative provider counters without double-counting", () => { + expect( + mergeCumulativeSubagentUsage( + { totalTokens: 100, outputTokens: 25 }, + { totalTokens: 140, outputTokens: 20 }, + ), + ).toEqual({ totalTokens: 140, outputTokens: 25 }); + }); + + it("is idempotent for duplicate and late cumulative snapshots", () => { + const latest = { + totalTokens: 140, + inputTokens: 95, + outputTokens: 45, + toolUses: 3, + }; + const duplicate = mergeCumulativeSubagentUsage(latest, latest); + const late = mergeCumulativeSubagentUsage(duplicate, { + totalTokens: 100, + inputTokens: 70, + outputTokens: 30, + toolUses: 2, + }); + + expect(duplicate).toEqual(latest); + expect(late).toEqual(latest); + expect(accumulateCumulativeSubagentUsage(latest, latest, duplicate)).toEqual(latest); + }); + + it("preserves omitted activation counters when they reappear", () => { + const firstActivation = { totalTokens: 100, inputTokens: 70 }; + const withoutInput = mergeCumulativeSubagentUsage(firstActivation, { totalTokens: 140 }); + const lifetimeWithoutInput = accumulateCumulativeSubagentUsage( + firstActivation, + firstActivation, + { + totalTokens: 140, + }, + ); + const restoredInput = mergeCumulativeSubagentUsage(withoutInput, { + totalTokens: 180, + inputTokens: 100, + }); + + expect(withoutInput).toEqual({ totalTokens: 140, inputTokens: 70 }); + expect( + accumulateCumulativeSubagentUsage(lifetimeWithoutInput, withoutInput, restoredInput), + ).toEqual({ totalTokens: 180, inputTokens: 100 }); }); }); diff --git a/apps/server/src/orchestration-v2/SubagentObservability.ts b/apps/server/src/orchestration-v2/SubagentObservability.ts index 1da0281843d..eafdd347654 100644 --- a/apps/server/src/orchestration-v2/SubagentObservability.ts +++ b/apps/server/src/orchestration-v2/SubagentObservability.ts @@ -1,13 +1,169 @@ import { SubagentActivationId, type NodeId, + type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivity, type OrchestrationV2SubagentRole, + type OrchestrationV2SubagentUsage, } from "@t3tools/contracts"; +import type * as DateTime from "effect/DateTime"; + +export const SUBAGENT_RECENT_ACTIVITY_LIMIT = 6; +const SUBAGENT_ACTIVITY_TEXT_LIMIT = 180; export const defaultSubagentRole = (name = "general-purpose"): OrchestrationV2SubagentRole => ({ name, source: "app_default", }); +export const providerSubagentRole = ( + name: string | null | undefined, + fallback = "general-purpose", +): OrchestrationV2SubagentRole => { + const normalized = name?.trim(); + return normalized ? { name: normalized, source: "provider" } : defaultSubagentRole(fallback); +}; + export const subagentActivationId = (subagentId: NodeId, ordinal: number) => SubagentActivationId.make(`${subagentId}:activation:${ordinal}`); + +const boundedActivityText = (value: string) => { + const normalized = value.trim(); + return normalized.length > SUBAGENT_ACTIVITY_TEXT_LIMIT + ? `${normalized.slice(0, SUBAGENT_ACTIVITY_TEXT_LIMIT - 1)}…` + : normalized; +}; + +export const appendSubagentActivity = ( + current: OrchestrationV2Subagent["recentActivity"], + summary: string | null | undefined, + at: DateTime.Utc, +): ReadonlyArray => { + if (!summary?.trim()) return current; + const bounded = boundedActivityText(summary); + if (current.at(-1)?.summary === bounded) return current; + return [...current, { at, summary: bounded }].slice(-SUBAGENT_RECENT_ACTIVITY_LIMIT); +}; + +const cumulative = (previous: number | undefined, next: number | undefined) => + next === undefined ? previous : Math.max(previous ?? 0, next); + +// Provider progress frames are cumulative snapshots within one scope, not +// deltas. Taking the maximum makes duplicate and late out-of-order frames +// idempotent without inventing usage the provider did not report. +export const mergeCumulativeSubagentUsage = ( + previous: OrchestrationV2SubagentUsage | null, + next: OrchestrationV2SubagentUsage | null | undefined, +): OrchestrationV2SubagentUsage | null => { + if (next === undefined || next === null) return previous; + return { + totalTokens: Math.max(previous?.totalTokens ?? 0, next.totalTokens), + ...(cumulative(previous?.inputTokens, next.inputTokens) === undefined + ? {} + : { inputTokens: cumulative(previous?.inputTokens, next.inputTokens) }), + ...(cumulative(previous?.cachedInputTokens, next.cachedInputTokens) === undefined + ? {} + : { cachedInputTokens: cumulative(previous?.cachedInputTokens, next.cachedInputTokens) }), + ...(cumulative(previous?.outputTokens, next.outputTokens) === undefined + ? {} + : { outputTokens: cumulative(previous?.outputTokens, next.outputTokens) }), + ...(cumulative(previous?.reasoningOutputTokens, next.reasoningOutputTokens) === undefined + ? {} + : { + reasoningOutputTokens: cumulative( + previous?.reasoningOutputTokens, + next.reasoningOutputTokens, + ), + }), + ...(cumulative(previous?.toolUses, next.toolUses) === undefined + ? {} + : { toolUses: cumulative(previous?.toolUses, next.toolUses) }), + ...(cumulative(previous?.durationMs, next.durationMs) === undefined + ? {} + : { durationMs: cumulative(previous?.durationMs, next.durationMs) }), + }; +}; + +const accumulated = ( + previous: number | undefined, + previousActivation: number | undefined, + next: number | undefined, +) => + next === undefined ? previous : (previous ?? 0) + Math.max(0, next - (previousActivation ?? 0)); + +// Lifetime usage advances only by the change since the prior snapshot for the +// current activation. Pass no prior activation snapshot when a new activation +// starts so its first cumulative snapshot is added in full. +export const accumulateCumulativeSubagentUsage = ( + previous: OrchestrationV2SubagentUsage | null, + previousActivation: OrchestrationV2SubagentUsage | null | undefined, + next: OrchestrationV2SubagentUsage | null | undefined, +): OrchestrationV2SubagentUsage | null => { + if (next === undefined || next === null) return previous; + return { + totalTokens: + (previous?.totalTokens ?? 0) + + Math.max(0, next.totalTokens - (previousActivation?.totalTokens ?? 0)), + ...(accumulated(previous?.inputTokens, previousActivation?.inputTokens, next.inputTokens) === + undefined + ? {} + : { + inputTokens: accumulated( + previous?.inputTokens, + previousActivation?.inputTokens, + next.inputTokens, + ), + }), + ...(accumulated( + previous?.cachedInputTokens, + previousActivation?.cachedInputTokens, + next.cachedInputTokens, + ) === undefined + ? {} + : { + cachedInputTokens: accumulated( + previous?.cachedInputTokens, + previousActivation?.cachedInputTokens, + next.cachedInputTokens, + ), + }), + ...(accumulated(previous?.outputTokens, previousActivation?.outputTokens, next.outputTokens) === + undefined + ? {} + : { + outputTokens: accumulated( + previous?.outputTokens, + previousActivation?.outputTokens, + next.outputTokens, + ), + }), + ...(accumulated( + previous?.reasoningOutputTokens, + previousActivation?.reasoningOutputTokens, + next.reasoningOutputTokens, + ) === undefined + ? {} + : { + reasoningOutputTokens: accumulated( + previous?.reasoningOutputTokens, + previousActivation?.reasoningOutputTokens, + next.reasoningOutputTokens, + ), + }), + ...(accumulated(previous?.toolUses, previousActivation?.toolUses, next.toolUses) === undefined + ? {} + : { + toolUses: accumulated(previous?.toolUses, previousActivation?.toolUses, next.toolUses), + }), + ...(accumulated(previous?.durationMs, previousActivation?.durationMs, next.durationMs) === + undefined + ? {} + : { + durationMs: accumulated( + previous?.durationMs, + previousActivation?.durationMs, + next.durationMs, + ), + }), + }; +}; diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/subagent/codex_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/subagent/codex_output.ts index 6e9160624ba..5dde986bb7f 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/subagent/codex_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/subagent/codex_output.ts @@ -43,7 +43,7 @@ export function assertSubagentOutput( assert.lengthOf(result.shellSnapshot.threads, 3); assert.deepEqual( projection.subagents.map((subagent) => subagent.status), - ["completed", "completed"], + ["idle", "idle"], ); assert.isTrue( projection.subagents.some( @@ -76,6 +76,7 @@ export function assertSubagentOutput( throw new Error(`Missing parent lifecycle item for subagent ${subagent.id}`); } assert.equal(parentItem.result, subagent.result); + assert.equal(parentItem.status, "completed"); if (subagent.childThreadId === null) { throw new Error(`Subagent ${subagent.id} is missing its child thread`); } diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_continue/codex_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_continue/codex_output.ts index b631cbf565f..d9f3dfcd43a 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_continue/codex_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_continue/codex_output.ts @@ -34,8 +34,10 @@ export function assertSubagentContinueOutput( assert.lengthOf(projection.subagents, 1); const subagent = projection.subagents[0]!; - assert.equal(subagent.status, "completed"); - assert.equal(subagent.result, "initial subagent response"); + // A reusable identity rests at idle between activations rather than + // finishing outright, and carries the latest activation's result. + assert.equal(subagent.status, "idle"); + assert.equal(subagent.result, "continued subagent response"); assert.isNotNull(subagent.childThreadId); if (subagent.childThreadId === null) { throw new Error("Continued subagent is missing its child thread"); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2/codex_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2/codex_output.ts index aa73387fa84..a93589c1853 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2/codex_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2/codex_output.ts @@ -41,12 +41,23 @@ export function assertSubagentV2Output( assert.equal(subagent.driver, "codex"); assert.equal(subagent.title, "/root/hello_agent"); assert.equal(subagent.prompt, ""); - assert.equal(subagent.status, "completed"); + // A subAgentActivity frame carries no model of its own, so the agent is + // recorded against the turn's selection rather than left blank. + assert.equal(subagent.model, "gpt-5.4"); + assert.equal(subagent.status, "idle"); assert.equal(subagent.result, "Hello."); + assert.isNull(subagent.currentActivationId); + assert.equal(subagent.activationCount, 1); assert.isNotNull(subagent.childThreadId); assert.isNotNull(subagent.providerThreadId); assert.isNotNull(subagent.nativeTaskRef); assert.isNotNull(subagent.completedAt); + const activation = projection.subagentActivations.find( + (candidate) => candidate.subagentId === subagent.id, + ); + assert.isDefined(activation); + assert.equal(activation.status, "completed"); + assert.isNotNull(activation.completedAt); const parentItem = projection.turnItems.find( (item) => item.type === "subagent" && item.subagentId === subagent.id, ); @@ -54,6 +65,7 @@ export function assertSubagentV2Output( if (parentItem?.type !== "subagent") { throw new Error(`Missing parent lifecycle item for subagent ${subagent.id}`); } + assert.equal(parentItem.status, "completed"); assert.equal(parentItem.result, subagent.result); if (subagent.childThreadId === null) { throw new Error(`Subagent ${subagent.id} is missing its child thread`); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2_nested/codex_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2_nested/codex_output.ts index b01365cb5e6..df38709413c 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2_nested/codex_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/subagent_v2_nested/codex_output.ts @@ -26,7 +26,7 @@ function projectionById( return projection; } -function assertCompletedProviderNativeSubagent(input: { +function assertIdleProviderNativeSubagent(input: { readonly projection: OrchestrationV2ThreadProjection; readonly title: string; readonly result: string; @@ -43,12 +43,20 @@ function assertCompletedProviderNativeSubagent(input: { assert.equal(subagent.createdBy, "agent"); assert.equal(subagent.driver, "codex"); assert.equal(subagent.title, input.title); - assert.equal(subagent.status, "completed"); + assert.equal(subagent.status, "idle"); assert.equal(subagent.result, input.result); + assert.isNull(subagent.currentActivationId); + assert.equal(subagent.activationCount, 1); assert.isNotNull(subagent.childThreadId); assert.isNotNull(subagent.providerThreadId); assert.isNotNull(subagent.nativeTaskRef); assert.isNotNull(subagent.completedAt); + const activation = input.projection.subagentActivations.find( + (candidate) => candidate.subagentId === subagent.id, + ); + assert.isDefined(activation); + assert.equal(activation.status, "completed"); + assert.isNotNull(activation.completedAt); return subagent; } @@ -71,7 +79,7 @@ export function assertSubagentV2NestedOutput( assertUserMessagesInclude(rootProjection, [SUBAGENT_V2_PROMPT]); assert.lengthOf(result.shellSnapshot.threads, 4); - const first = assertCompletedProviderNativeSubagent({ + const first = assertIdleProviderNativeSubagent({ projection: rootProjection, title: "/root/hello_agent", result: "Subagent says: “Hello.”", @@ -88,7 +96,7 @@ export function assertSubagentV2NestedOutput( assert.lengthOf(firstProjection.providerTurns, 1); assertTurnItemTypes(firstProjection, ["subagent", "assistant_message"]); - const second = assertCompletedProviderNativeSubagent({ + const second = assertIdleProviderNativeSubagent({ projection: firstProjection, title: "/root/hello_agent/hello_agent", result: "Subagent says: “Hello.”", @@ -105,7 +113,7 @@ export function assertSubagentV2NestedOutput( assert.lengthOf(secondProjection.providerTurns, 1); assertTurnItemTypes(secondProjection, ["subagent", "assistant_message"]); - const third = assertCompletedProviderNativeSubagent({ + const third = assertIdleProviderNativeSubagent({ projection: secondProjection, title: "/root/hello_agent/hello_agent/hello_agent", result: "Hello.", From 2567faf4adf704f75567986493559f570c70f1f9 Mon Sep 17 00:00:00 2001 From: Shivam Sharma <91240327+shivamhwp@users.noreply.github.com> Date: Wed, 29 Jul 2026 01:11:51 +0530 Subject: [PATCH 2/2] test(orchestration-v2): replay a Claude workflow end to end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The adapter's workflow projection was covered in isolation and the panel derivation was covered against synthetic agents, but nothing drove real workflow frames through the orchestrator to the projection the client receives. Adds a claude_workflow replay fixture: a local_workflow task with two phases and two members, asserted on the resulting projection. Pins what the panel groups by — that members claim declared phases, that their membership points at the coordinator that actually reached the projection, and that the coordinator's usage still covers its members so subtracting theirs cannot go negative. Co-Authored-By: Claude Opus 5 (1M context) --- .../claude_workflow/claude_transcript.ndjson | 12 +++ .../testkit/fixtures/claude_workflow/input.ts | 18 ++++ .../fixtures/claude_workflow/output.ts | 91 +++++++++++++++++++ .../testkit/fixtures/index.ts | 14 +++ 4 files changed, 135 insertions(+) create mode 100644 apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson create mode 100644 apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/input.ts create mode 100644 apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson new file mode 100644 index 00000000000..9a5b478726a --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/claude_transcript.ndjson @@ -0,0 +1,12 @@ +{"type": "transcript_start", "provider": "claudeAgent", "protocol": "claude-agent-sdk.query", "version": "0.2.111", "scenario": "claude_workflow", "metadata": {"prompts": ["Run a two-phase workflow, then answer exactly: claude workflow fixture complete"], "model": "claude-sonnet-4-6", "nativeSessionId": "3c7f21a0-9d4e-4b21-8f60-000000000201", "queryMode": "streaming", "tools": "claude_code", "permissionMode": "bypassPermissions", "generatedBy": "hand-authored-workflow-observability"}} +{"type": "expect_outbound", "label": "query.open", "frame": {"type": "query.open", "options": {"model": "claude-sonnet-4-6", "tools": {"type": "preset", "preset": "claude_code"}, "permissionMode": "bypassPermissions", "allowDangerouslySkipPermissions": true, "sessionId": "3c7f21a0-9d4e-4b21-8f60-000000000201"}}} +{"type": "expect_outbound", "label": "prompt.offer:1", "frame": {"type": "prompt.offer", "message": {"type": "user", "message": {"role": "user", "content": "Run a two-phase workflow, then answer exactly: claude workflow fixture complete"}, "parent_tool_use_id": null}}} +{"type": "emit_inbound", "label": "system", "frame": {"type": "system", "subtype": "init", "agents": [], "apiKeySource": "none", "claude_code_version": "2.1.183", "cwd": "/tmp/claude-replay-claude_workflow", "tools": [], "mcp_servers": [], "model": "claude-sonnet-4-6", "permissionMode": "bypassPermissions", "slash_commands": [], "output_style": "default", "skills": [], "plugins": [], "fast_mode_state": "off", "uuid": "af3bf4cc-1385-4e19-a5fa-fcbe2a0ca201", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}} +{"type": "emit_inbound", "label": "assistant:commentary", "frame": {"type": "assistant", "message": {"model": "claude-sonnet-4-6", "id": "msg_01claudeworkflow1", "type": "message", "role": "assistant", "content": [{"type": "text", "text": "Starting the two-phase workflow."}], "stop_reason": null, "stop_sequence": null, "stop_details": null, "usage": {"input_tokens": 1, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0, "output_tokens": 1, "service_tier": "standard", "inference_geo": "not_available"}, "context_management": null}, "parent_tool_use_id": null, "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201", "uuid": "97898e28-ee4e-4fcf-80c3-5c790c502201"}} +{"type": "emit_inbound", "label": "task_started:local_workflow", "frame": {"type": "system", "subtype": "task_started", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Research and implement", "task_type": "local_workflow", "prompt": "Research the parser, then implement the fix.", "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73201", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}} +{"type": "emit_inbound", "label": "task_progress:phase-1", "frame": {"type": "system", "subtype": "task_progress", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Research phase running", "usage": {"input_tokens": 300, "output_tokens": 200, "total_tokens": 500, "tool_uses": 4}, "workflow_progress": [{"type": "workflow_phase", "index": 0, "title": "Research"}, {"type": "workflow_phase", "index": 1, "title": "Implement"}, {"type": "workflow_agent", "index": 0, "label": "Researcher", "state": "active", "phaseIndex": 0, "attempt": 0, "model": "claude-sonnet-4-6", "lastToolName": "Read", "tokens": 200, "toolCalls": 2}], "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73202", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}} +{"type": "emit_inbound", "label": "task_progress:phase-2", "frame": {"type": "system", "subtype": "task_progress", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "description": "Implement phase running", "usage": {"input_tokens": 700, "output_tokens": 500, "total_tokens": 1200, "tool_uses": 9}, "workflow_progress": [{"type": "workflow_phase", "index": 0, "title": "Research"}, {"type": "workflow_phase", "index": 1, "title": "Implement"}, {"type": "workflow_agent", "index": 0, "label": "Researcher", "state": "done", "phaseIndex": 0, "attempt": 0, "model": "claude-sonnet-4-6", "lastToolName": "Read", "tokens": 400, "toolCalls": 3}, {"type": "workflow_agent", "index": 1, "label": "Implementer", "state": "active", "phase_index": 1, "attempt": 0, "model": "claude-opus-4-1", "last_tool_name": "Edit", "tokens": 500, "tool_calls": 4}], "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73203", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}} +{"type": "emit_inbound", "label": "task_notification:local_workflow", "frame": {"type": "system", "subtype": "task_notification", "task_id": "wf7ka2mq1", "tool_use_id": "toolu_01ClaudeWorkflowFixture01", "status": "completed", "output_file": "", "summary": "Workflow finished both phases.", "usage": {"input_tokens": 900, "output_tokens": 600, "total_tokens": 1500, "tool_uses": 11}, "uuid": "4873c95b-1aa9-44a5-9783-dc56dac73204", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201"}} +{"type": "emit_inbound", "label": "assistant:commentary", "frame": {"type": "assistant", "message": {"model": "claude-sonnet-4-6", "id": "msg_01claudeworkflow2", "type": "message", "role": "assistant", "content": [{"type": "text", "text": "claude workflow fixture complete"}], "stop_reason": null, "stop_sequence": null, "stop_details": null, "usage": {"input_tokens": 1, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0, "output_tokens": 1, "service_tier": "standard", "inference_geo": "not_available"}, "context_management": null}, "parent_tool_use_id": null, "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201", "uuid": "89be9e65-097b-47e8-849b-be723386b201"}} +{"type": "emit_inbound", "label": "result", "frame": {"type": "result", "subtype": "success", "is_error": false, "api_error_status": null, "duration_ms": 4000, "duration_api_ms": 3500, "num_turns": 2, "result": "claude workflow fixture complete", "stop_reason": "end_turn", "session_id": "3c7f21a0-9d4e-4b21-8f60-000000000201", "total_cost_usd": 0.001, "usage": {"input_tokens": 2, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0, "output_tokens": 2, "server_tool_use": {"web_search_requests": 0, "web_fetch_requests": 0}, "service_tier": "standard", "cache_creation": {"ephemeral_1h_input_tokens": 0, "ephemeral_5m_input_tokens": 0}, "inference_geo": "", "iterations": [{"input_tokens": 1, "output_tokens": 1, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 0, "cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 0}, "type": "message"}], "speed": "standard"}, "modelUsage": {"claude-sonnet-4-6": {"inputTokens": 2, "outputTokens": 2, "cacheReadInputTokens": 0, "cacheCreationInputTokens": 0, "webSearchRequests": 0, "costUSD": 0.001, "contextWindow": 200000, "maxOutputTokens": 32000}}, "permission_denials": [], "terminal_reason": "completed", "fast_mode_state": "off", "uuid": "89be9e65-097b-47e8-849b-be723386b202"}} +{"type": "runtime_exit", "status": "success"} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/input.ts new file mode 100644 index 00000000000..831019d7816 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/input.ts @@ -0,0 +1,18 @@ +import type { OrchestratorFixtureInput } from "../shared.ts"; + +export const CLAUDE_WORKFLOW_PROMPT = + "Run a two-phase workflow, then answer exactly: claude workflow fixture complete"; + +/** + * End-to-end cover for the Claude workflow coordinator. The adapter-level + * projection of phases and workers is unit-tested, and the panel derivation is + * unit-tested against synthetic agents — nothing joined the two. This drives + * real workflow frames through the orchestrator so the coordinator, its + * phases, and its members are asserted on the projection the client actually + * receives, then feeds that projection to the panel derivation. + */ +export function claudeWorkflowInput(): OrchestratorFixtureInput { + return { + steps: [{ type: "message", text: CLAUDE_WORKFLOW_PROMPT }], + }; +} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts new file mode 100644 index 00000000000..064ad572a43 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_workflow/output.ts @@ -0,0 +1,91 @@ +import { assert } from "@effect/vitest"; +import type { ProviderReplayTranscript } from "@t3tools/contracts"; + +import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts"; +import { + assertBaseProjection, + assertSemanticProjectionIntegrity, + assertUserMessagesInclude, + projectionFor, +} from "../shared.ts"; +import { CLAUDE_WORKFLOW_PROMPT } from "./input.ts"; + +export function assertClaudeWorkflowOutput( + result: OrchestratorV2ScenarioResult, + transcript: ProviderReplayTranscript, +) { + assertBaseProjection({ result, transcript, runCount: 1, runStatuses: ["completed"] }); + + const projection = projectionFor(result, transcript.scenario); + assertSemanticProjectionIntegrity(projection); + assertUserMessagesInclude(projection, [CLAUDE_WORKFLOW_PROMPT]); + + const coordinator = projection.subagents.find((subagent) => subagent.kind === "workflow"); + assert.isDefined(coordinator, "the workflow must reach the projection as a coordinator"); + assert.deepEqual(coordinator?.role, { name: "workflow-coordinator", source: "app_default" }); + assert.deepEqual(coordinator?.workflow?.phases, [ + { index: 0, title: "Research" }, + { index: 1, title: "Implement" }, + ]); + + const members = projection.subagents + .filter((subagent) => subagent.kind === "workflow_agent") + .toSorted( + (left, right) => + (left.workflowMembership?.agentIndex ?? 0) - (right.workflowMembership?.agentIndex ?? 0), + ); + assert.lengthOf(members, 2, "both workflow members must reach the projection"); + assert.deepEqual( + members.map((member) => [ + member.title, + member.status, + member.workflowMembership?.phaseIndex, + member.model, + member.usage?.totalTokens, + ]), + [ + ["Researcher", "completed", 0, "claude-sonnet-4-6", 400], + ["Implementer", "running", 1, "claude-opus-4-1", 500], + ], + ); + // Membership must point at the coordinator that actually reached the + // projection, not at an id the adapter invented independently. + assert.deepEqual( + [...new Set(members.map((member) => member.workflowMembership?.workflowSubagentId))], + [coordinator?.id], + ); + + // The panel groups by these three fields, so pin the values it reads rather + // than only the ones the coordinator carries. The derivation itself is + // covered in client-runtime; the schema is what joins the two sides, so what + // matters here is that the server emits a shape that groups cleanly: + // every member claims a declared phase, and no member is left ungrouped. + const declaredPhases = new Set((coordinator?.workflow?.phases ?? []).map((phase) => phase.index)); + assert.deepEqual( + members.map((member) => declaredPhases.has(member.workflowMembership?.phaseIndex ?? -1)), + [true, true], + ); + assert.lengthOf( + projection.subagents.filter( + (subagent) => subagent.kind === "workflow_agent" && subagent.workflowMembership === null, + ), + 0, + "a member without membership would render outside its workflow", + ); + // The coordinator's usage already covers its members, so the panel subtracts + // theirs from it. That only stays non-negative if the provider's totals are + // carried through unmodified. + const memberTotal = members.reduce( + (total, member) => total + (member.usage?.totalTokens ?? 0), + 0, + ); + assert.isAtLeast(coordinator?.usage?.totalTokens ?? 0, memberTotal); + + const assistantTexts = projection.turnItems.flatMap((item) => + item.type === "assistant_message" ? [item.text] : [], + ); + assert.deepEqual(assistantTexts, [ + "Starting the two-phase workflow.", + "claude workflow fixture complete", + ]); +} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts index 510b85c4c46..04c157b99cf 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts @@ -4,6 +4,8 @@ import { claudeIdleResumeInput } from "./claude_idle_resume/input.ts"; import { assertClaudeIdleResumeOutput } from "./claude_idle_resume/output.ts"; import { claudeLocalBashTaskInput } from "./claude_local_bash_task/input.ts"; import { assertClaudeLocalBashTaskOutput } from "./claude_local_bash_task/output.ts"; +import { claudeWorkflowInput } from "./claude_workflow/input.ts"; +import { assertClaudeWorkflowOutput } from "./claude_workflow/output.ts"; import { claudeResultIsErrorInput } from "./claude_result_is_error/input.ts"; import { assertClaudeResultIsErrorOutput } from "./claude_result_is_error/output.ts"; import { grokSubagentLineageInput } from "./grok_subagent_lineage/input.ts"; @@ -99,6 +101,18 @@ export const ORCHESTRATOR_REPLAY_FIXTURES = [ }, ], }, + { + name: "claude_workflow", + buildInput: claudeWorkflowInput, + providers: [ + { + driver: ProviderDriverKind.make("claudeAgent"), + transcriptFile: new URL("./claude_workflow/claude_transcript.ndjson", import.meta.url), + modelSelection: CLAUDE_MODEL_SELECTION, + assertOutput: assertClaudeWorkflowOutput, + }, + ], + }, { name: "claude_idle_resume", buildInput: claudeIdleResumeInput,