1import type {
2 OrchestrationV2SubagentWorkflow,
3 OrchestrationV2WorkflowAgent,
4 OrchestrationV2WorkflowPhase,
5 OrchestrationV2WorkflowRunHandles,
6} from "@t3tools/contracts";
7
8import { field, text } from "../unknownField.ts";
9
10export const CLAUDE_WORKFLOW_TASK_TYPE = "local_workflow";
11
12export function hasClaudeWorkflowSnapshot(
13 message: unknown,
14): message is { readonly workflow_progress: ReadonlyArray<unknown> } {
15 const entries = field(message, "workflow_progress");
16 return Array.isArray(entries) && entries.length > 0;
17}
18
19const AGENT_STATES: Record<string, OrchestrationV2WorkflowAgent["state"] | undefined> = {
20 queued: "queued",
21 start: "running",
22 done: "completed",
23 error: "failed",
24};
25
26function count(value: unknown): number | undefined {
27 return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined;
28}
29
30function timestamp(value: unknown): number | undefined {
31 const milliseconds = count(value);
32 return milliseconds !== undefined && milliseconds <= 8_640_000_000_000_000
33 ? milliseconds
34 : undefined;
35}
36
37function preview(value: unknown): string | undefined {
38 return typeof value === "string" && value.length > 0 ? value : undefined;
39}
40
41function optional<T>(key: string, value: T | undefined): Record<string, T> {
42 return value === undefined ? {} : ({ [key]: value } as Record<string, T>);
43}
44
45function parsePhase(entry: unknown): OrchestrationV2WorkflowPhase | null {
46 const index = count(field(entry, "index"));
47 const title = text(field(entry, "title"));
48 return index === undefined || title === undefined ? null : { index, title };
49}
50
51function parseAgent(entry: unknown): OrchestrationV2WorkflowAgent | null {
52 const index = count(field(entry, "index"));
53 const label = text(field(entry, "label"));
54 if (index === undefined || label === undefined) return null;
55 // An unknown state is in-flight, not settled: guessing "completed" would
56 // freeze a live row, while guessing "running" self-corrects on the next frame.
57 const reportedState = text(field(entry, "state")) ?? "";
58 const state = Object.hasOwn(AGENT_STATES, reportedState)
59 ? (AGENT_STATES[reportedState] ?? "running")
60 : "running";
61 return {
62 index,
63 label,
64 state,
65 ...optional("agentId", text(field(entry, "agentId"))),
66 ...optional("phaseIndex", count(field(entry, "phaseIndex"))),
67 ...optional("phaseTitle", text(field(entry, "phaseTitle"))),
68 ...optional("model", text(field(entry, "model"))),
69 ...optional("attempt", count(field(entry, "attempt"))),
70 ...optional("totalTokens", count(field(entry, "tokens"))),
71 ...optional("toolCalls", count(field(entry, "toolCalls"))),
72 ...optional("durationMs", count(field(entry, "durationMs"))),
73 ...optional("queuedAt", timestamp(field(entry, "queuedAt"))),
74 ...optional("startedAt", timestamp(field(entry, "startedAt"))),
75 ...optional("prompt", preview(field(entry, "promptPreview"))),
76 ...optional("result", preview(field(entry, "resultPreview"))),
77 };
78}
79
80/**
81 * Reads the Workflow tool's structured result. A background launch answers the
82 * model with an acknowledgement rather than the run's output, and that
83 * acknowledgement is the only place the run's filesystem handles appear —
84 * nothing later on the event stream repeats them.
85 */
86export function parseClaudeWorkflowRunHandles(
87 toolUseResult: unknown,
88): OrchestrationV2WorkflowRunHandles | undefined {
89 if (text(field(toolUseResult, "taskType")) !== CLAUDE_WORKFLOW_TASK_TYPE) return undefined;
90 const handles = {
91 ...optional("runId", text(field(toolUseResult, "runId"))),
92 ...optional("transcriptDir", text(field(toolUseResult, "transcriptDir"))),
93 ...optional("scriptPath", text(field(toolUseResult, "scriptPath"))),
94 };
95 return Object.keys(handles).length === 0 ? undefined : handles;
96}
97
98function parseClaudeWorkflowProgress(message: unknown): {
99 readonly phases: ReadonlyArray<OrchestrationV2WorkflowPhase>;
100 readonly agents: ReadonlyArray<OrchestrationV2WorkflowAgent>;
101} | null {
102 if (!hasClaudeWorkflowSnapshot(message)) return null;
103 const phases: OrchestrationV2WorkflowPhase[] = [];
104 const agents: OrchestrationV2WorkflowAgent[] = [];
105 for (const entry of message.workflow_progress) {
106 const type = text(field(entry, "type"));
107 if (type === "workflow_phase") {
108 const phase = parsePhase(entry);
109 if (phase !== null) phases.push(phase);
110 } else if (type === "workflow_agent") {
111 const agent = parseAgent(entry);
112 if (agent !== null) agents.push(agent);
113 }
114 }
115 return phases.length === 0 && agents.length === 0 ? null : { phases, agents };
116}
117
118/**
119 * Folds one frame into the coordinator's workflow state. A frame without a
120 * snapshot still carries usage, so totals keep climbing between snapshots.
121 * A phase only ever joins the plan — the provider re-sends the full plan on
122 * every snapshot, but a late frame that omits an already-seen phase must not
123 * shorten the arc the user is looking at.
124 *
125 * Returns undefined when nothing identifies the task as a workflow. Progress
126 * frames do not repeat `task_type`, so this is the single place that decides
127 * whether a task is a coordinator; without it every ordinary subagent's usage
128 * frame would mint an empty workflow and render as a coordinator.
129 */
130export function mergeClaudeWorkflowProgress(input: {
131 readonly previous: OrchestrationV2SubagentWorkflow | undefined;
132 readonly name?: string | undefined;
133 readonly runHandles?: OrchestrationV2WorkflowRunHandles | undefined;
134 readonly message: unknown;
135}): OrchestrationV2SubagentWorkflow | undefined {
136 const { previous } = input;
137 const snapshot = parseClaudeWorkflowProgress(input.message);
138 if (
139 previous === undefined &&
140 input.name === undefined &&
141 input.runHandles === undefined &&
142 snapshot === null
143 ) {
144 return undefined;
145 }
146 const reported = field(input.message, "usage");
147 const phasesByIndex = new Map(
148 (previous?.phases ?? []).map((phase) => [phase.index, phase] as const),
149 );
150 for (const phase of snapshot?.phases ?? []) phasesByIndex.set(phase.index, phase);
151 const agentsByIndex = new Map(
152 (previous?.agents ?? []).map((agent) => [agent.index, agent] as const),
153 );
154 for (const agent of snapshot?.agents ?? []) {
155 const prior = agentsByIndex.get(agent.index);
156 const attempt = agent.attempt ?? prior?.attempt ?? 1;
157 if (prior !== undefined && attempt < (prior.attempt ?? 1)) continue;
158 const restarted = prior !== undefined && attempt > (prior.attempt ?? 1);
159 const regressed =
160 !restarted &&
161 (((prior?.state === "completed" || prior?.state === "failed") &&
162 agent.state !== prior.state) ||
163 (prior?.state === "running" && agent.state === "queued"));
164 agentsByIndex.set(
165 agent.index,
166 regressed
167 ? { ...agent, ...prior }
168 : {
169 ...(restarted
170 ? {
171 ...optional("phaseIndex", prior.phaseIndex),
172 ...optional("phaseTitle", prior.phaseTitle),
173 }
174 : prior),
175 ...agent,
176 },
177 );
178 }
179 const runHandles =
180 input.runHandles === undefined
181 ? previous?.runHandles
182 : { ...previous?.runHandles, ...input.runHandles };
183 return {
184 ...optional("name", input.name ?? previous?.name),
185 ...optional("runHandles", runHandles),
186 phases: Array.from(phasesByIndex.values()).sort((left, right) => left.index - right.index),
187 agents: Array.from(agentsByIndex.values()).sort((left, right) => left.index - right.index),
188 ...optional("totalTokens", count(field(reported, "total_tokens")) ?? previous?.totalTokens),
189 ...optional("toolCalls", count(field(reported, "tool_uses")) ?? previous?.toolCalls),
190 ...optional("durationMs", count(field(reported, "duration_ms")) ?? previous?.durationMs),
191 };
192}