diff --git a/src/client/app/KannaTranscript.tsx b/src/client/app/KannaTranscript.tsx index 9b736eeca..724dccf3b 100644 --- a/src/client/app/KannaTranscript.tsx +++ b/src/client/app/KannaTranscript.tsx @@ -8,6 +8,7 @@ import { AccountInfoMessage } from "../components/messages/AccountInfoMessage" import { TextMessage } from "../components/messages/TextMessage" import { AskUserQuestionMessage } from "../components/messages/AskUserQuestionMessage" import { ExitPlanModeMessage } from "../components/messages/ExitPlanModeMessage" +import { WorkflowMessage } from "../components/messages/WorkflowMessage" import { TodoWriteMessage } from "../components/messages/TodoWriteMessage" import { ToolCallMessage } from "../components/messages/ToolCallMessage" import { ResultMessage } from "../components/messages/ResultMessage" @@ -335,6 +336,15 @@ function sameMessage(left: HydratedTranscriptMessage, right: HydratedTranscriptM return right.kind === "handoff_boundary" && left.fromProvider === right.fromProvider && left.toProvider === right.toProvider + case "workflow_state": + // lastSnapshotId is the _id of the newest folded snapshot: any + // state/usage/agent change produces a new snapshot entry, so comparing + // it is sufficient (and far cheaper than deep-comparing agents). + // createdAt-based revisions are NOT safe here — two lifecycle snapshots + // can land in the same millisecond. + return right.kind === "workflow_state" + && left.taskId === right.taskId + && left.lastSnapshotId === right.lastSnapshotId case "unknown": return right.kind === "unknown" && left.json === right.json } @@ -511,6 +521,9 @@ const TranscriptSingleRow = memo(function TranscriptSingleRow({ case "interrupted": rendered = break + case "workflow_state": + rendered = + break case "compact_boundary": rendered = break diff --git a/src/client/components/messages/WorkflowMessage.tsx b/src/client/components/messages/WorkflowMessage.tsx new file mode 100644 index 000000000..779f5ea46 --- /dev/null +++ b/src/client/components/messages/WorkflowMessage.tsx @@ -0,0 +1,213 @@ +import { useEffect, useMemo, useState } from "react" +import { ChevronRight, CircleCheck, CircleX, Pause, Workflow as WorkflowIcon } from "lucide-react" +import type { HydratedTranscriptMessage, WorkflowAgentSnapshot, WorkflowRunStatus } from "../../../shared/types" +import { cn } from "../../lib/utils" +import { AnimatedShinyText } from "../ui/animated-shiny-text" + +type WorkflowStateMessage = Extract + +interface Props { + message: WorkflowStateMessage +} + +// Local copy: upstream has no shared formatDuration export. +function formatDuration(ms: number): string { + if (ms < 1000) return `${ms}ms` + const totalSeconds = Math.floor(ms / 1000) + const hours = Math.floor(totalSeconds / 3600) + const minutes = Math.floor((totalSeconds % 3600) / 60) + const seconds = totalSeconds % 60 + const parts: string[] = [] + if (hours > 0) parts.push(`${hours}h`) + if (minutes > 0) parts.push(`${minutes}m`) + if (seconds > 0) parts.push(`${seconds}s`) + return parts.join(" ") || "0s" +} + +function formatTokens(tokens: number): string { + if (tokens < 1_000) return `${Math.round(tokens)}` + if (tokens < 1_000_000) return `${(tokens / 1_000).toFixed(1).replace(/\.0$/, "")}k` + return `${(tokens / 1_000_000).toFixed(1).replace(/\.0$/, "")}m` +} + +function formatShortDuration(ms: number | undefined): string { + if (ms === undefined || ms < 0) return "—" + if (ms < 1000) return "<1s" + return formatDuration(ms) +} + +function statusIcon(status: WorkflowRunStatus) { + if (status === "completed") return + if (status === "failed" || status === "killed") return + if (status === "paused") return + return +} + +function agentDotClass(state: WorkflowAgentSnapshot["state"]): string { + switch (state) { + case "done": return "bg-emerald-500" + case "error": return "bg-destructive" + case "running": return "bg-amber-400 animate-pulse" + default: return "bg-muted-foreground/25" + } +} + +const MAX_GRID_DOTS = 200 + +/** Live elapsed while running; frozen at the last snapshot's age once terminal. */ +function useElapsedMs(message: WorkflowStateMessage): number { + const running = message.status === "running" || message.status === "pending" + const [now, setNow] = useState(() => Date.now()) + + useEffect(() => { + if (!running) return + const timer = setInterval(() => setNow(Date.now()), 1_000) + return () => clearInterval(timer) + }, [running]) + + if (!running) return Math.max(0, message.revision - message.startedAtMs) + return Math.max(0, now - message.startedAtMs) +} + +function AgentRow({ agent }: { agent: WorkflowAgentSnapshot }) { + return ( + + + + + {agent.label} + + {agent.error ? {agent.error} : null} + + + {agent.tokens !== undefined && agent.tokens > 0 ? formatTokens(agent.tokens) : "—"} + + + {agent.toolCalls ?? 0} + + + {formatShortDuration(agent.durationMs)} + + + ) +} + +export function WorkflowMessage({ message }: Props) { + const [expanded, setExpanded] = useState(false) + const elapsedMs = useElapsedMs(message) + const running = message.status === "running" || message.status === "pending" + + const agents = message.agents + const totalTokens = useMemo(() => { + const agentSum = agents.reduce((sum, agent) => sum + (agent.tokens ?? 0), 0) + return Math.max(message.usage?.totalTokens ?? 0, agentSum) + }, [agents, message.usage?.totalTokens]) + + const phaseGroups = useMemo(() => { + const groups = new Map() + for (const phase of message.phases) { + groups.set(phase.index, { title: phase.title, agents: [] }) + } + for (const agent of agents) { + const key = agent.phaseIndex ?? 0 + let group = groups.get(key) + if (!group) { + group = { title: agent.phaseTitle ?? null, agents: [] } + groups.set(key, group) + } + group.agents.push(agent) + } + return [...groups.entries()] + .sort(([left], [right]) => left - right) + .map(([, group]) => group) + .filter((group) => group.agents.length > 0) + }, [agents, message.phases]) + + const name = message.workflowName ?? "workflow" + const metaParts = [ + "Workflow", + agents.length > 0 ? `${agents.length} agent${agents.length === 1 ? "" : "s"}` : null, + elapsedMs > 0 ? formatShortDuration(elapsedMs) : null, + totalTokens > 0 ? `${formatTokens(totalTokens)} tokens` : null, + ].filter(Boolean) + + return ( +
+ + + {agents.length > 0 ? ( +
+ {agents.slice(0, MAX_GRID_DOTS).map((agent) => ( + + ))} + {agents.length > MAX_GRID_DOTS ? ( + +{agents.length - MAX_GRID_DOTS} + ) : null} +
+ ) : null} + + {expanded ? ( +
+ {message.description ? ( +

{message.description}

+ ) : null} + {phaseGroups.map((group, groupIndex) => ( +
0 && "mt-3")}> + {group.title ? ( +
{group.title}
+ ) : null} + + + + + + + + + + + {group.agents.map((agent) => )} + +
AgentTokensToolsTime
+
+ ))} + {message.summary ? ( +

{message.summary}

+ ) : null} +
+ ) : null} +
+ ) +} diff --git a/src/client/lib/parseTranscript.test.ts b/src/client/lib/parseTranscript.test.ts index fbca37c22..c16bf9df3 100644 --- a/src/client/lib/parseTranscript.test.ts +++ b/src/client/lib/parseTranscript.test.ts @@ -342,4 +342,53 @@ describe("getLatestToolIds", () => { TodoWrite: null, }) }) + + test("folds same-millisecond workflow snapshots without reusing stale state", () => { + const createdAt = Date.now() + const first: TranscriptEntry = { + _id: "workflow-snapshot-1", + createdAt, + kind: "workflow_state", + taskId: "workflow-task-1", + toolId: "workflow-tool-1", + workflowName: "demo", + status: "running", + phases: [], + agents: [{ index: 1, label: "agent-1", state: "running" }], + } + const second: TranscriptEntry = { + ...first, + _id: "workflow-snapshot-2", + status: "completed", + agents: [{ index: 1, label: "agent-1", state: "done", tokens: 500 }], + } + + const messages = processTranscriptMessages([ + entry({ + kind: "tool_call", + tool: { + kind: "tool", + toolKind: "unknown_tool", + toolName: "Workflow", + toolId: "workflow-tool-1", + input: {}, + }, + }), + first, + second, + ]) + + const toolCall = messages.find((message) => message.kind === "tool") + expect(toolCall?.hidden).toBe(true) + + const workflowMessages = messages.filter((message) => message.kind === "workflow_state") + expect(workflowMessages).toHaveLength(1) + const workflow = workflowMessages[0] + if (workflow?.kind !== "workflow_state") throw new Error("unexpected message") + expect(workflow.status).toBe("completed") + expect(workflow.agents[0]).toMatchObject({ state: "done", tokens: 500 }) + expect(workflow.lastSnapshotId).toBe("workflow-snapshot-2") + expect(workflow.startedAtMs).toBe(createdAt) + expect(workflow.id).toBe("workflow-snapshot-1") + }) }) diff --git a/src/client/lib/parseTranscript.ts b/src/client/lib/parseTranscript.ts index 32acc0803..12052a673 100644 --- a/src/client/lib/parseTranscript.ts +++ b/src/client/lib/parseTranscript.ts @@ -39,8 +39,13 @@ function getStructuredToolResultFromDebug(entry: Extract + export function processTranscriptMessages(entries: TranscriptEntry[]): HydratedTranscriptMessage[] { const pendingToolCalls = new Map() + // Latest workflow snapshot per background-task id: the server appends + // snapshots, the client keeps last-write-wins anchored at first occurrence. + const workflowMessages = new Map() const messages: HydratedTranscriptMessage[] = [] for (const entry of entries) { @@ -161,6 +166,42 @@ export function processTranscriptMessages(entries: TranscriptEntry[]): HydratedT kind: "interrupted", }) break + case "workflow_state": { + // The raw Workflow tool card is superseded by the workflow card once + // lifecycle snapshots exist for its tool-use id. + const spawningCall = entry.toolId ? pendingToolCalls.get(entry.toolId) : undefined + if (spawningCall) spawningCall.hydrated.hidden = true + + const fields = { + status: entry.status, + usage: entry.usage, + phases: entry.phases, + agents: entry.agents, + workflowName: entry.workflowName, + description: entry.description, + summary: entry.summary, + // _id is the change marker (unique per snapshot); createdAt feeds + // elapsed-time math but can collide within a millisecond. + lastSnapshotId: entry._id, + revision: entry.createdAt, + } + const existing = workflowMessages.get(entry.taskId) + if (existing) { + Object.assign(existing, fields) + } else { + const message: WorkflowStateMessage = { + ...createBaseMessage(entry), + kind: "workflow_state", + taskId: entry.taskId, + toolId: entry.toolId, + startedAtMs: entry.createdAt, + ...fields, + } + workflowMessages.set(entry.taskId, message) + messages.push(message) + } + break + } default: messages.push({ ...createBaseMessage(entry), diff --git a/src/server/agent.ts b/src/server/agent.ts index 8a4304cd3..7098df83d 100644 --- a/src/server/agent.ts +++ b/src/server/agent.ts @@ -53,6 +53,7 @@ import { fallbackTitleFromMessage } from "./generate-title" import { asNumber, asRecord } from "../shared/json" import { buildHandoffContext, buildHandoffMessageContent, type HandoffContext } from "./handoff" import { timestamped } from "./transcript" +import { WorkflowTracker } from "./workflow-tracker" const CLAUDE_TOOLSET = [ "Skill", @@ -503,6 +504,7 @@ async function* createClaudeHarnessStream( let seenAssistantUsageIds = new Set() let latestUsageSnapshot: ContextWindowUsageSnapshot | null = null let lastKnownContextWindow: number | undefined + const workflowTracker = new WorkflowTracker() for await (const sdkMessage of q as AsyncIterable) { const sessionToken = typeof sdkMessage.session_id === "string" ? sdkMessage.session_id : null @@ -609,6 +611,13 @@ async function* createClaudeHarnessStream( for (const entry of normalizeClaudeStreamMessage(sdkMessage)) { yield { type: "transcript", entry } } + + // Background-task lifecycle (system/task_*) is invisible to the + // normalizer; the workflow tracker folds it into canonical + // workflow_state snapshots. + for (const entry of workflowTracker.process(sdkMessage)) { + yield { type: "transcript", entry } + } } } diff --git a/src/server/handoff.test.ts b/src/server/handoff.test.ts index 35d27d315..9f09aa5d0 100644 --- a/src/server/handoff.test.ts +++ b/src/server/handoff.test.ts @@ -75,6 +75,13 @@ describe("buildHandoffContext", () => { userPrompt("hello"), timestamped({ kind: "status", status: "compacting" }), timestamped({ kind: "context_window_updated", usage: { usedTokens: 10, compactsAutomatically: false } }), + timestamped({ + kind: "workflow_state", + taskId: "workflow-task-1", + status: "running", + phases: [], + agents: [], + }), timestamped({ kind: "user_prompt", content: "secret steering", hidden: true }), assistantText("hi"), timestamped({ kind: "result", subtype: "success", isError: false, durationMs: 1, result: "hi" }), diff --git a/src/server/handoff.ts b/src/server/handoff.ts index dbb0d4a7e..1a1923a7c 100644 --- a/src/server/handoff.ts +++ b/src/server/handoff.ts @@ -159,6 +159,7 @@ function blockFromEntry(entry: TranscriptEntry): Omit | case "context_window_updated": case "compact_boundary": case "context_cleared": + case "workflow_state": return null } } diff --git a/src/server/workflow-tracker.test.ts b/src/server/workflow-tracker.test.ts new file mode 100644 index 000000000..d1416649a --- /dev/null +++ b/src/server/workflow-tracker.test.ts @@ -0,0 +1,251 @@ +import { describe, expect, test } from "bun:test" +import { WorkflowTracker } from "./workflow-tracker" +import type { WorkflowStateEntry } from "../shared/types" + +const TOOL_ID = "toolu_01QeXqtJ8tXG7dfx9qwQf8an" +const TASK_ID = "wj6t8vti0" + +function taskStarted(overrides: Record = {}) { + return { + type: "system", + subtype: "task_started", + task_id: TASK_ID, + tool_use_id: TOOL_ID, + description: "Demo workflow", + task_type: "local_workflow", + workflow_name: "probe-demo", + prompt: "export const meta = {...}", + uuid: "u1", + session_id: "s1", + ...overrides, + } +} + +function taskProgress(workflowProgress: unknown[], overrides: Record = {}) { + return { + type: "system", + subtype: "task_progress", + task_id: TASK_ID, + tool_use_id: TOOL_ID, + description: "Greetings: agent-1", + usage: { total_tokens: 100, tool_uses: 2, duration_ms: 500 }, + summary: "Demo workflow", + workflow_progress: workflowProgress, + uuid: "u2", + session_id: "s1", + ...overrides, + } +} + +function agentEvent(overrides: Record = {}) { + return { + type: "workflow_agent", + index: 1, + label: "agent-1", + phaseIndex: 1, + phaseTitle: "Greetings", + model: "claude-opus-4-8", + state: "start", + queuedAt: 1000, + promptPreview: "Say hi", + ...overrides, + } +} + +function single(entries: unknown[]): WorkflowStateEntry { + expect(entries).toHaveLength(1) + const entry = entries[0] as WorkflowStateEntry + expect(entry.kind).toBe("workflow_state") + return entry +} + +describe("WorkflowTracker", () => { + test("task_started for a local_workflow emits an initial running snapshot", () => { + const tracker = new WorkflowTracker() + const entry = single(tracker.process(taskStarted())) + + expect(entry.taskId).toBe(TASK_ID) + expect(entry.toolId).toBe(TOOL_ID) + expect(entry.workflowName).toBe("probe-demo") + expect(entry.description).toBe("Demo workflow") + expect(entry.status).toBe("running") + expect(entry.phases).toEqual([]) + expect(entry.agents).toEqual([]) + }) + + test("non-workflow task_started is ignored", () => { + const tracker = new WorkflowTracker() + expect(tracker.process(taskStarted({ task_type: "subagent", workflow_name: undefined }))).toEqual([]) + // Progress for an untracked, non-workflow task is also ignored. + expect(tracker.process(taskProgress([], { workflow_progress: undefined }))).toEqual([]) + }) + + test("skip_transcript stamps hidden on snapshots", () => { + const tracker = new WorkflowTracker() + const entry = single(tracker.process(taskStarted({ skip_transcript: true }))) + expect(entry.hidden).toBe(true) + }) + + test("folds phases and agent lifecycle from workflow_progress", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + + const entry = single(tracker.process(taskProgress([ + { type: "workflow_phase", index: 1, title: "Greetings" }, + agentEvent(), + agentEvent({ index: 2, label: "agent-2", promptPreview: "Say hello" }), + agentEvent({ agentId: "a5a5ced", startedAt: 1010 }), + ]))) + + expect(entry.phases).toEqual([{ index: 1, title: "Greetings" }]) + expect(entry.agents).toHaveLength(2) + expect(entry.agents[0]).toMatchObject({ + index: 1, + label: "agent-1", + state: "running", + agentId: "a5a5ced", + phaseIndex: 1, + phaseTitle: "Greetings", + model: "claude-opus-4-8", + promptPreview: "Say hi", + }) + // agent-2 was queued (start without startedAt). + expect(entry.agents[1]).toMatchObject({ index: 2, state: "queued" }) + expect(entry.usage).toEqual({ totalTokens: 100, toolUses: 2, durationMs: 500 }) + }) + + test("agent progress and terminal states merge without losing earlier fields", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + tracker.process(taskProgress([agentEvent({ agentId: "a1", startedAt: 1010 })])) + + const entry = single(tracker.process(taskProgress([ + agentEvent({ state: "progress", tokens: 5000, toolCalls: 3 }), + agentEvent({ state: "done", tokens: 8000, toolCalls: 4, durationMs: 4200 }), + ]))) + + expect(entry.agents[0]).toMatchObject({ + index: 1, + state: "done", + agentId: "a1", + tokens: 8000, + toolCalls: 4, + durationMs: 4200, + model: "claude-opus-4-8", + }) + }) + + test("error state captures the error message", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + const entry = single(tracker.process(taskProgress([ + agentEvent({ state: "error", error: "model not available", durationMs: 799 }), + ]))) + expect(entry.agents[0]).toMatchObject({ state: "error", error: "model not available" }) + }) + + test("terminal agent state never regresses on a replayed start event", () => { + let nowMs = 0 + const tracker = new WorkflowTracker(() => nowMs) + tracker.process(taskStarted()) + tracker.process(taskProgress([agentEvent({ state: "done", tokens: 100 })])) + // A replayed start is a no-op (no structural change), so it flushes on the + // throttle interval rather than immediately. + nowMs = 5_000 + const entry = single(tracker.process(taskProgress([agentEvent({ state: "start", startedAt: 1010 })]))) + expect(entry.agents[0]!.state).toBe("done") + }) + + test("pure token ticks are throttled; structural changes emit immediately", () => { + let nowMs = 0 + const tracker = new WorkflowTracker(() => nowMs) + tracker.process(taskStarted()) + tracker.process(taskProgress([agentEvent({ startedAt: 1010 })])) + + // Token-only update right after the last emit: swallowed. + nowMs = 100 + expect(tracker.process(taskProgress([agentEvent({ state: "progress", tokens: 10 })]))).toEqual([]) + + // Past the throttle interval the buffered progress flushes. + nowMs = 5_000 + const flushed = single(tracker.process(taskProgress([agentEvent({ state: "progress", tokens: 20 })]))) + expect(flushed.agents[0]!.tokens).toBe(20) + + // A state transition emits immediately even inside the interval. + nowMs = 5_100 + const terminal = single(tracker.process(taskProgress([agentEvent({ state: "done", tokens: 30 })]))) + expect(terminal.agents[0]!.state).toBe("done") + }) + + test("task_updated patches run status", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + const entry = single(tracker.process({ + type: "system", + subtype: "task_updated", + task_id: TASK_ID, + patch: { status: "completed", end_time: 2000 }, + })) + expect(entry.status).toBe("completed") + }) + + test("task_notification finalizes status, summary, usage, and straggling agents", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + tracker.process(taskProgress([ + agentEvent({ agentId: "a1", startedAt: 1010 }), + agentEvent({ index: 2, label: "agent-2", state: "done", tokens: 500 }), + ])) + + const entry = single(tracker.process({ + type: "system", + subtype: "task_notification", + task_id: TASK_ID, + tool_use_id: TOOL_ID, + status: "completed", + summary: "Workflow completed", + usage: { total_tokens: 900, tool_uses: 7, duration_ms: 6000 }, + })) + + expect(entry.status).toBe("completed") + expect(entry.summary).toBe("Workflow completed") + expect(entry.usage).toEqual({ totalTokens: 900, toolUses: 7, durationMs: 6000 }) + // The still-running agent is coerced to a terminal state. + expect(entry.agents.every((agent) => agent.state === "done")).toBe(true) + }) + + test("stopped notification maps to killed and errors stragglers", () => { + const tracker = new WorkflowTracker() + tracker.process(taskStarted()) + tracker.process(taskProgress([agentEvent({ startedAt: 1010 })])) + const entry = single(tracker.process({ + type: "system", + subtype: "task_notification", + task_id: TASK_ID, + status: "stopped", + })) + expect(entry.status).toBe("killed") + expect(entry.agents[0]!.state).toBe("error") + }) + + test("mid-flight task_progress with workflow_progress lazily creates the run", () => { + const tracker = new WorkflowTracker() + const entry = single(tracker.process(taskProgress([agentEvent({ startedAt: 1010 })]))) + expect(entry.taskId).toBe(TASK_ID) + expect(entry.agents).toHaveLength(1) + }) + + test("unknown message shapes are ignored", () => { + const tracker = new WorkflowTracker() + expect(tracker.process({ type: "assistant" })).toEqual([]) + expect(tracker.process({ type: "system", subtype: "init" })).toEqual([]) + expect(tracker.process(null)).toEqual([]) + expect(tracker.process({ type: "system", subtype: "task_updated", task_id: "unknown", patch: { status: "completed" } })).toEqual([]) + }) + + test("workflow inside a subagent keeps the parent agent scope", () => { + const tracker = new WorkflowTracker() + const entry = single(tracker.process(taskStarted({ parent_tool_use_id: "toolu_parent" }))) + expect(entry.agentId).toBe("toolu_parent") + }) +}) diff --git a/src/server/workflow-tracker.ts b/src/server/workflow-tracker.ts new file mode 100644 index 000000000..e204e9dd5 --- /dev/null +++ b/src/server/workflow-tracker.ts @@ -0,0 +1,284 @@ +import { asRecord } from "../shared/json" +import type { + TranscriptEntry, + WorkflowAgentRunState, + WorkflowAgentSnapshot, + WorkflowPhaseSnapshot, + WorkflowRunStatus, + WorkflowUsageSnapshot, +} from "../shared/types" +import { timestamped } from "./transcript" + +/** + * Minimum interval between appended snapshots for pure token/progress ticks. + * Structural changes (new agents, state transitions, run status changes) + * always emit immediately — this only throttles the noisy middle. + */ +const PROGRESS_EMIT_INTERVAL_MS = 1_500 + +interface WorkflowRun { + taskId: string + toolId?: string + workflowName?: string + description?: string + status: WorkflowRunStatus + usage?: WorkflowUsageSnapshot + phases: Map + agents: Map + summary?: string + /** Parent subagent scope when the workflow was launched inside a child agent. */ + agentId?: string + hidden?: boolean + lastEmittedAt: number + pendingProgress: boolean +} + +function asNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" && value ? value : undefined +} + +function normalizeUsage(value: unknown): WorkflowUsageSnapshot | undefined { + const usage = asRecord(value) + if (!usage) return undefined + return { + totalTokens: asNumber(usage.total_tokens) ?? 0, + toolUses: asNumber(usage.tool_uses) ?? 0, + durationMs: asNumber(usage.duration_ms) ?? 0, + } +} + +function normalizeRunStatus(value: unknown): WorkflowRunStatus | null { + if ( + value === "pending" || value === "running" || value === "completed" + || value === "failed" || value === "killed" || value === "paused" + ) return value + // task_notification uses "stopped" for user-killed runs. + if (value === "stopped") return "killed" + return null +} + +/** + * The CLI's workflow_agent events use state "start" both for queued entries + * (no startedAt yet) and actually-started ones; "progress" for running ticks; + * "error"/"done" as terminals. Unknown states keep the previous value so a + * future CLI can add states without regressing existing runs to "queued". + */ +function normalizeAgentState(value: unknown, startedAt: unknown, previous?: WorkflowAgentRunState): WorkflowAgentRunState { + if (value === "error" || value === "failed") return "error" + if (value === "done" || value === "success" || value === "complete" || value === "completed") return "done" + if (value === "progress" || value === "running") return "running" + if (value === "start" || value === "queued") { + // Terminal states never regress on a late/replayed start event. + if (previous === "done" || previous === "error") return previous + return startedAt !== undefined && startedAt !== null ? "running" : "queued" + } + return previous ?? "running" +} + +function isTerminalAgentState(state: WorkflowAgentRunState): boolean { + return state === "done" || state === "error" +} + +/** + * Folds the Claude CLI's background-task lifecycle messages + * (system/task_started, task_progress, task_updated, task_notification) for + * `local_workflow` tasks into canonical `workflow_state` transcript snapshots. + * + * NOT the todo list: TaskCreate/TaskUpdate/... tool calls are handled by + * ClaudeTaskTracker. This tracker consumes system messages the normalizer + * otherwise drops, so it is purely additive to the entry stream. + */ +export class WorkflowTracker { + private readonly runs = new Map() + private readonly now: () => number + + constructor(now: () => number = Date.now) { + this.now = now + } + + process(message: any): TranscriptEntry[] { + if (message?.type !== "system") return [] + + const subtype = message.subtype + if (subtype === "task_started") return this.onTaskStarted(message) + if (subtype === "task_progress") return this.onTaskProgress(message) + if (subtype === "task_updated") return this.onTaskUpdated(message) + if (subtype === "task_notification") return this.onTaskNotification(message) + return [] + } + + private onTaskStarted(message: any): TranscriptEntry[] { + const taskId = asString(message.task_id) + if (!taskId) return [] + const isWorkflow = message.task_type === "local_workflow" || asString(message.workflow_name) !== undefined + if (!isWorkflow) return [] + + const run: WorkflowRun = { + taskId, + toolId: asString(message.tool_use_id), + workflowName: asString(message.workflow_name), + description: asString(message.description), + status: "running", + phases: new Map(), + agents: new Map(), + agentId: asString(message.parent_tool_use_id), + ...(message.skip_transcript === true ? { hidden: true } : {}), + lastEmittedAt: 0, + pendingProgress: false, + } + this.runs.set(taskId, run) + return [this.snapshot(run)] + } + + private onTaskProgress(message: any): TranscriptEntry[] { + const taskId = asString(message.task_id) + if (!taskId) return [] + + let run = this.runs.get(taskId) + // A run can surface mid-flight (session resume): create it lazily, but + // only when the payload proves it is a workflow task. + if (!run) { + if (!Array.isArray(message.workflow_progress)) return [] + run = { + taskId, + toolId: asString(message.tool_use_id), + description: asString(message.summary) ?? asString(message.description), + status: "running", + phases: new Map(), + agents: new Map(), + agentId: asString(message.parent_tool_use_id), + lastEmittedAt: 0, + pendingProgress: false, + } + this.runs.set(taskId, run) + } + + const usage = normalizeUsage(message.usage) + if (usage) run.usage = usage + + let structuralChange = false + if (Array.isArray(message.workflow_progress)) { + for (const rawEvent of message.workflow_progress) { + const event = asRecord(rawEvent) + if (!event) continue + + if (event.type === "workflow_phase") { + const index = asNumber(event.index) + const title = asString(event.title) + if (index === undefined || !title) continue + if (!run.phases.has(index)) structuralChange = true + run.phases.set(index, { index, title }) + continue + } + + if (event.type === "workflow_agent") { + const index = asNumber(event.index) + if (index === undefined) continue + const previous = run.agents.get(index) + const state = normalizeAgentState(event.state, event.startedAt, previous?.state) + const next: WorkflowAgentSnapshot = { + index, + label: asString(event.label) ?? previous?.label ?? `agent-${index}`, + state, + ...(asNumber(event.phaseIndex) !== undefined + ? { phaseIndex: asNumber(event.phaseIndex) } + : previous?.phaseIndex !== undefined ? { phaseIndex: previous.phaseIndex } : {}), + ...(asString(event.phaseTitle) + ? { phaseTitle: asString(event.phaseTitle) } + : previous?.phaseTitle ? { phaseTitle: previous.phaseTitle } : {}), + ...(asString(event.agentId) + ? { agentId: asString(event.agentId) } + : previous?.agentId ? { agentId: previous.agentId } : {}), + ...(asString(event.model) + ? { model: asString(event.model) } + : previous?.model ? { model: previous.model } : {}), + ...(asString(event.promptPreview) + ? { promptPreview: asString(event.promptPreview) } + : previous?.promptPreview ? { promptPreview: previous.promptPreview } : {}), + ...(asNumber(event.tokens) !== undefined + ? { tokens: asNumber(event.tokens) } + : previous?.tokens !== undefined ? { tokens: previous.tokens } : {}), + ...(asNumber(event.toolCalls) !== undefined + ? { toolCalls: asNumber(event.toolCalls) } + : previous?.toolCalls !== undefined ? { toolCalls: previous.toolCalls } : {}), + ...(asNumber(event.durationMs) !== undefined + ? { durationMs: asNumber(event.durationMs) } + : previous?.durationMs !== undefined ? { durationMs: previous.durationMs } : {}), + ...(asString(event.error) + ? { error: asString(event.error) } + : previous?.error ? { error: previous.error } : {}), + } + if (!previous || previous.state !== next.state || isTerminalAgentState(next.state) && !isTerminalAgentState(previous.state)) { + structuralChange = true + } + run.agents.set(index, next) + } + } + } + + run.pendingProgress = true + if (structuralChange || this.now() - run.lastEmittedAt >= PROGRESS_EMIT_INTERVAL_MS) { + return [this.snapshot(run)] + } + return [] + } + + private onTaskUpdated(message: any): TranscriptEntry[] { + const taskId = asString(message.task_id) + const run = taskId ? this.runs.get(taskId) : undefined + if (!run) return [] + + const patch = asRecord(message.patch) + const status = normalizeRunStatus(patch?.status) + if (!status) return [] + if (status === run.status && !run.pendingProgress) return [] + run.status = status + return [this.snapshot(run)] + } + + private onTaskNotification(message: any): TranscriptEntry[] { + const taskId = asString(message.task_id) + const run = taskId ? this.runs.get(taskId) : undefined + if (!run) return [] + + const status = normalizeRunStatus(message.status) + if (status) run.status = status + const usage = normalizeUsage(message.usage) + if (usage) run.usage = usage + const summary = asString(message.summary) + if (summary) run.summary = summary + + // Terminal signal: no straggling agent may stay "running"/"queued" forever. + if (run.status === "completed" || run.status === "failed" || run.status === "killed") { + for (const [index, agent] of run.agents) { + if (!isTerminalAgentState(agent.state)) { + run.agents.set(index, { ...agent, state: run.status === "completed" ? "done" : "error" }) + } + } + } + return [this.snapshot(run)] + } + + private snapshot(run: WorkflowRun): TranscriptEntry { + run.lastEmittedAt = this.now() + run.pendingProgress = false + return timestamped({ + kind: "workflow_state", + ...(run.agentId ? { agentId: run.agentId } : {}), + ...(run.hidden ? { hidden: true } : {}), + taskId: run.taskId, + ...(run.toolId ? { toolId: run.toolId } : {}), + ...(run.workflowName ? { workflowName: run.workflowName } : {}), + ...(run.description ? { description: run.description } : {}), + status: run.status, + ...(run.usage ? { usage: run.usage } : {}), + phases: [...run.phases.values()].sort((left, right) => left.index - right.index), + agents: [...run.agents.values()].sort((left, right) => left.index - right.index), + ...(run.summary ? { summary: run.summary } : {}), + }, this.now()) + } +} diff --git a/src/shared/types.ts b/src/shared/types.ts index 6599e739f..e1a230450 100644 --- a/src/shared/types.ts +++ b/src/shared/types.ts @@ -1475,6 +1475,67 @@ export interface HandoffBoundaryEntry extends TranscriptEntryBase { } } +/** + * Status of a CLI background-task registry entry (workflow runs). Distinct + * from any todo-list task status — background tasks are executing units, + * not plan bookkeeping. + */ +export type WorkflowRunStatus = "pending" | "running" | "completed" | "failed" | "killed" | "paused" + +export type WorkflowAgentRunState = "queued" | "running" | "done" | "error" + +/** One agent() call inside a Workflow run, folded from workflow_progress events. */ +export interface WorkflowAgentSnapshot { + /** 1-based agent() call index within the run — stable identity across progress events. */ + index: number + label: string + phaseIndex?: number + phaseTitle?: string + /** CLI-assigned agent id, present once the agent actually starts. */ + agentId?: string + model?: string + state: WorkflowAgentRunState + promptPreview?: string + tokens?: number + toolCalls?: number + durationMs?: number + error?: string +} + +export interface WorkflowPhaseSnapshot { + index: number + title: string +} + +export interface WorkflowUsageSnapshot { + totalTokens: number + toolUses: number + durationMs: number +} + +/** + * Canonical snapshot of one Workflow tool run, folded server-side from the + * SDK's system/task_started + task_progress + task_updated + task_notification + * messages. Append-only in the transcript: the client keeps the latest + * snapshot per taskId (last write wins), anchored at the first occurrence. + */ +export interface WorkflowStateEntry extends TranscriptEntryBase { + kind: "workflow_state" + /** Parent subagent scope when a workflow is launched inside a child agent. */ + agentId?: string + /** Background-task handle from the CLI task registry (NOT a todo-list task id). */ + taskId: string + /** Tool-use id of the spawning Workflow tool call, when known. */ + toolId?: string + workflowName?: string + description?: string + status: WorkflowRunStatus + usage?: WorkflowUsageSnapshot + phases: WorkflowPhaseSnapshot[] + agents: WorkflowAgentSnapshot[] + summary?: string +} + export type TranscriptEntry = | UserPromptEntry | SystemInitEntry @@ -1490,6 +1551,7 @@ export type TranscriptEntry = | ContextClearedEntry | InterruptedEntry | HandoffBoundaryEntry + | WorkflowStateEntry export interface HydratedToolCallBase { id: string @@ -1583,6 +1645,7 @@ export type HydratedTranscriptMessage = | ({ kind: "context_cleared"; id: string; messageId?: string; timestamp: string; hidden?: boolean }) | ({ kind: "handoff_boundary"; fromProvider: AgentProvider; toProvider: AgentProvider; id: string; messageId?: string; timestamp: string; hidden?: boolean }) | ({ kind: "interrupted"; id: string; messageId?: string; timestamp: string; hidden?: boolean }) + | ({ kind: "workflow_state"; taskId: string; toolId?: string; workflowName?: string; description?: string; status: WorkflowRunStatus; usage?: WorkflowUsageSnapshot; phases: WorkflowPhaseSnapshot[]; agents: WorkflowAgentSnapshot[]; summary?: string; /** _id of the latest folded snapshot entry — guaranteed-unique change marker for memoized rows (createdAt can collide within one millisecond). */ lastSnapshotId: string; /** createdAt of the latest folded snapshot — elapsed-time math only, NOT change detection. */ revision: number; /** createdAt of the first snapshot — anchor for live elapsed time. */ startedAtMs: number; id: string; messageId?: string; timestamp: string; hidden?: boolean }) | ({ kind: "unknown"; json: string; id: string; messageId?: string; timestamp: string; hidden?: boolean }) | ({ id: string; messageId?: string; hidden?: boolean } & HydratedToolCall)