feat(ai): add a uniform usage telemetry channel (Phase 1: the generic seam)

[sroussey]

Refs #624 — Phase 1 of 2 (the generic seam only; no provider run-fn is touched).

Providers surface no token usage today, so nothing downstream can compute cost. This lands the provider-agnostic channel end to end and leaves the per-provider fill-in to Phase 2. Every provider currently emits usage: undefined (none sets the new field), so this is purely additive with no behavior change: absent usage leaves no key on any output, and all existing streaming/caching semantics are untouched.

The four design decisions

1. Shape — a normalized Usage, where undefined ≠ 0

Usage (in packages/task-graph/src/task/StreamTypes.ts) carries input, output, cached, cacheWrite, reasoning, total, plus an extra bag for provider-specific counters with no normalized slot.

Every counter is number | undefined, and undefined means "the provider did not report this", never "zero". This is the load-bearing detail for cost math: a model that billed 0 cached tokens and a model that never tells you about caching are different facts, and collapsing them to 0 silently understates spend. mergeUsage preserves the distinction field-wise (undefined + undefined === undefined, undefined + 5 === 5).

It lives in task-graph beside the other canonical stream types (StreamFinish is defined there, not in packages/ai), and is re-exported from @workglow/ai so a provider package can import type { Usage } from "@workglow/ai" without depending on task-graph directly.

2. Transport — a sibling field on finish

StreamFinish gains an optional usage, as a sibling of data, never inside it.

• Why not inside data? data is governed by the streaming conventions: {} for delta streams, the entire Output for one-shot run-fns, and finish.data.object for json-mode. Folding token counts into data would corrupt all three at once.
• Why not a new usage stream-event arm? Every switch (event.type) over StreamEvent in the codebase and in downstream consumers would need a new case, and a stray usage event arriving mid-stream would have ambiguous ordering against finish. A sibling on the event that already means "this request is over" has exactly the right lifetime.
• Why not the run result? AiProviderRunFn returns Promise<void> by explicit contract — there is no return value to put metadata on.
• The worker boundary (WorkerManager.callWorkerRunFunction) forwards the whole event object, so a sibling field survives structured clone with no protocol change. Covered by a test.

3. Surfacing — a reserved output key, mirroring refusal

There is already a proven precedent for a non-port output field: refusal. USAGE_OUTPUT_KEY follows it exactly —

	refusal	usage
Folded by StreamEventAccumulator	applyRefusal()	applyUsage(), chained into all three materialize() exits
Folded by StreamProcessor	yes, before stream_end	yes, next to the refusal fold
Declared on any outputSchema()	no (reserved by design)	no (reserved by design)
Kept off dataflow edges	yes	yes
Output cache	blocks the save	strips the field, saves the value

The cache difference is deliberate. A refusal must not be memoized because it would replay as "the answer". Usage is different: the value is still worth caching, but token counts are a fact about one execution — serving that entry later costs zero tokens, so replaying the original counts would invent spend that never happened. So CacheCoordinator.save strips USAGE_OUTPUT_KEY rather than bailing. This lands in the same commit as the accumulator/processor change, so there is no window in which cache hits report phantom tokens.

Telemetry is recorded in one place — AiTask.execute(), via a new packages/ai/src/capability/UsageTelemetry.ts — so provider packages never touch telemetry directly (they also run inside workers, which have a separate registry and no main-thread telemetry provider). It records OTel gen-ai semantic-convention attributes (gen_ai.usage.input_tokens, …output_tokens, …cached_input_tokens, …cache_write_input_tokens, …reasoning_tokens, …total_tokens) and one getLogger().debug(...). Guarded on isEnabled following the traced.ts pattern, so attribute flattening is never paid for when nothing is collecting. ITelemetryProvider is unchanged — SpanAttributes already accepts flat numbers.

4. Multi-turn — aggregated consumer-side, keeping providers stateless

The no-accumulation rule means a provider must never buffer. So summing happens at the two places that already own multi-request control flow, both of which drop the sibling today:

• AiChatWithKbTask deliberately swallows inner-turn finish events (the agent loop). It now merges each turn's usage into an accumulator and attaches the total to the outer finish, which already carries { iterations }.
• StructuredGenerationTask synthesizes a fresh finish across validation-retry attempts. It now sums across attempts — a rejected attempt is still a billed request, and without this a schema-flaky model would look exactly as cheap as a clean one.

mergeUsage sums numeric extra keys and takes last-wins on string keys (a string is a label — service tier, cache key — not a counter).

On output schemas

usage is not added to any task's outputSchema() — reserved keys are undeclared by design, exactly like refusal. Checked the additionalProperties: false exposure explicitly: task output is never validated against outputSchema at runtime (only input and config are), and Task.addInput copies only keys the target declares as ports, so usage cannot reach a strict downstream input over a normal edge. Both are pinned by tests — one runs StructuredGenerationTask (whose output schema is additionalProperties: false, required: ["object"]) with usage attached, and one asserts usage does not leak onto a downstream task's input.

Phase 2 is deliberately deferred

Per-provider fill-in must land after #641 (ai-provider-cache-checkpoints, 84 files) merges — that PR edits ~15 of the provider run-fns Phase 2 would touch. Phase 1 changes no provider file at all (all 16 files here are in packages/ai, packages/task-graph, packages/test).

Checked against #641's actual file list rather than assuming: it shares exactly 2 of my 16 files, and in both the change is a single line inside getJobInput (sessionId: x → session: { sessionId: x }) —

shared file	#641 hunk	this PR's hunks
packages/ai/src/task/base/AiTask.ts	getJobInput, ~L314	import; one call after the provider-error throw, ~L252
packages/ai/src/task/AiChatWithKbTask.ts	getJobInput, ~L368	imports; turn loop ~L434; emit factory ~L631; final yield ~L698

So the overlap is file-level only, with no overlapping hunks — these auto-merge. The deferral is about the ~15 provider run-fns Phase 2 needs, which this PR leaves untouched.

Because the seam is generic, the follow-up is mechanical: each run-fn maps its own finish payload into Usage and attaches it to the finish it already emits. Checklist:

Usage already flows past the loop and is discarded — no request change needed

• @workglow/anthropic — usage.input_tokens / output_tokens / cache_read_input_tokens → cached / cache_creation_input_tokens → cacheWrite
• @workglow/openai (Responses) — usage.input_tokens / output_tokens / input_tokens_details.cached_tokens / cache_write_tokens / output_tokens_details.reasoning_tokens
• @workglow/google-gemini — usageMetadata.promptTokenCount / candidatesTokenCount / cachedContentTokenCount
• @workglow/ollama — the final chunk's prompt/eval counts (exact field names to confirm against the SDK response at implementation time)

Needs stream_options: { include_usage: true } on the request

• @workglow/xai
• @workglow/deepseek
• @workglow/openrouter
• @workglow/llamacpp-server

No usage available — leave unreported (undefined, not 0)

• @workglow/node-llama-cpp
• @workglow/huggingface-transformers
• @workglow/chrome-ai
• @workglow/tf-mediapipe

Tests

Driven entirely from synthetic streams (no provider emits usage yet), extending the existing suites rather than adding parallel ones — collectStream.test.ts, StreamEventAccumulator.test.ts, StreamingAccumulation.test.ts, WorkerRunFn_roundtrip.test.ts, AiChatWithKbTask.test.ts, StructuredGenerationTask.test.ts.

Covered: usage on finish surfaces as result.usage; absent usage leaves no key at all; one-shot finish + usage keeps data as the payload; delta-mode finish + usage keeps deltas winning; json-mode keeps data.object intact; usage and refusal coexist (a refused turn still reports its billed tokens); mergeUsage unit tests (undefined preservation both directions, extras merged with numeric summed / string last-wins, no input mutation); parity invariant — run() and the streaming path produce identical usage; usage is not resurrected from the output cache (and the stored entry has no usage key); usage survives the worker boundary through structuredClone; a 2-turn agent loop and a 2-attempt structured-generation retry each aggregate correctly, with single-turn/single-attempt cases asserting no double-counting.

Verification

All commands run on this branch at 5fff2be.

$ bun run build
 Tasks:    84 successful, 84 total
  Time:    45.283s

$ bun run build:types --force
 Tasks:    41 successful, 41 total
Cached:    0 cached, 41 total
  Time:    56.46s

$ bun scripts/test.ts ai vitest
Running all tests in sections [ai] — 38 file(s)
 Test Files  38 passed (38)
      Tests  275 passed (275)

$ bun scripts/test.ts task graph vitest
Running all tests in sections [task+graph] — 148 file(s)
 Test Files  148 passed (148)
      Tests  1885 passed | 24 skipped (1909)

bun run format then reordered imports in two files, so both affected sections were re-run afterwards to confirm the formatted tree is still green:

$ bun scripts/test.ts ai graph vitest
Running all tests in sections [ai+graph] — 114 file(s)
 Test Files  114 passed (114)
      Tests  1028 passed (1028)
   Duration  322.53s

🤖 Generated with Claude Code

https://claude.ai/code/session_01PwyJuFrJnibKvrrk8Fa4Fn

---

Generated by Claude Code

[github-actions]

Coverage Report

Status	Category	Percentage	Covered / Total
🔵	Lines	62.32%	28143 / 45155
🔵	Statements	62.2%	29182 / 46914
🔵	Functions	62.58%	5363 / 8569
🔵	Branches	51.29%	13902 / 27104

File Coverage

No changed files found.

Generated in workflow #2855 for commit 8fe8a5d by the Vitest Coverage Report Action