fix(cost): read Workflow-tool agents and complete their output tokens

.agents/skills/bpmn2agent-cost/SKILL.md

@@ -18,8 +18,23 @@ report the numbers the script prints and never recompute or round them yourself.
 - **Run data**, in this order:
   1. `.claude/runs/<workflow>/ledger.jsonl`, written by the workflow's cost-ledger hook; survives
      the cleanup of old transcripts. No session named → the newest.
-  2. A session transcript `~/.claude/projects/<project>/<session>.jsonl` (subagents sit next to it);
-     use it when the workflow has no ledger yet or to cross-check one.
+  2. A session transcript `~/.claude/projects/<project>/<session>.jsonl` (subagents sit next to it,
+     Workflow-tool agents under `subagents/workflows/<run>/`); use it when the workflow has no ledger
+     yet or to cross-check one.
+- **Telemetry (optional, makes it exact):** the transcripts of Workflow-tool agents only hold a
+  streaming snapshot of the output tokens. Without telemetry the report completes them from
+  `cost-state` and marks that share as estimated. To get exact numbers, start Claude Code with
+  telemetry pointed at the bundled sink and pass its file as `--otel`:
+
+  ```bash
+  node ${CLAUDE_SKILL_DIR}/scripts/otel-sink.mjs otel.jsonl &        # OTLP/HTTP JSON receiver on :4318
+  CLAUDE_CODE_ENABLE_TELEMETRY=1 OTEL_LOGS_EXPORTER=otlp OTEL_METRICS_EXPORTER=otlp \
+  OTEL_EXPORTER_OTLP_PROTOCOL=http/json OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4318 \
+  OTEL_LOGS_EXPORT_INTERVAL=2000 claude
+  ```
+
+  Its `api_request` events carry the request id, so every request gets its exact usage, and the
+  web-search/fetch helper calls that have no transcript show up as *Hilfsaufrufe*.
 
 ## 2. Run the report
 
@@ -28,7 +43,7 @@ node ${CLAUDE_SKILL_DIR}/scripts/cost-report.mjs --map <cost-map.json> \
   --ledger .claude/runs/<workflow>/ledger.jsonl        # or: --session <session.jsonl>
 ```
 
-Optional: `--session-id <id>` (ledger with several runs), `--claude-json <file>` (result of
+Optional: `--otel <otel.jsonl>` (exact output tokens, see above), `--session-id <id>` (ledger with several runs), `--claude-json <file>` (result of
 `claude -p --output-format json`, its `total_cost_usd` is compared), `--out <dir>`. It writes
 `cost.json` and `cost.md` next to the ledger and prints the checks. Exit `1` means a reconciliation
 check failed: say which one and do not present the cost as reliable.
@@ -38,8 +53,11 @@ check failed: say which one and do not present the cost as reliable.
 - Lead with the total and the top elements; point at `cost.md` for the tables.
 - **nicht zugeordnet** over 5 % means the label convention or the cost map is off. Name where the
   requests came from (the report lists them); do not distribute the cost over elements.
-- **Hilfsaufrufe** are in the session total but in no transcript (web-search helper model and the
-  like). A warning that a model's tokens are missing means the session was resumed or cleared.
+- **Hilfsaufrufe** are in the session total but in no transcript (web-search/fetch helper calls). A
+  warning that a model's tokens are missing means that, or that the session was resumed or cleared.
+- **geschätzt**: say how much. The total is exact; only the split of the missing output tokens over
+  the Workflow agents follows text length (on a real run the per-element error summed to under 2 % of
+  the cost). Offer `--otel` for exact figures; never present the estimated share as measured.
 - MCP calls (Gemini/NotebookLM and others) are counted but carry no price: they are billed outside
   Claude.
 - Prices are those of `prices.json`; the report names its version. A failing "Preis reproduziert

.agents/skills/bpmn2agent-cost/scripts/fixture.mjs

@@ -34,8 +34,14 @@ const u = (c, o = {}) => ({
 })
 
 /** One API request = a thinking line with a streaming snapshot, then the final line. */
-function request(id, model, extra = {}, c = OPUS_USAGE) {
+function request(id, model, extra = {}, c = OPUS_USAGE, final = true) {
   const base = { type: 'assistant', requestId: id, ...extra }
+  // an agent of the Workflow tool: every line keeps the streaming snapshot, none gets the final output
+  if (!final) {
+    return [
+      { ...base, message: { model, content: [{ type: 'text', text: 'x'.repeat(40) }], stop_reason: null, usage: u(c, { output: 3 }) } },
+    ]
+  }
   return [
     { ...base, message: { model, content: [{ type: 'thinking' }], stop_reason: null, usage: u(c, { output: 2 }) } },
     { ...base, message: { model, content: [{ type: 'text' }], stop_reason: 'end_turn', usage: u(c) } },
@@ -52,7 +58,7 @@ const write = (file, lines) => {
  * 3 subagent requests (element K2 by description, lane QA by agent type, unassigned) and a cost-state
  * that also holds a haiku auxiliary call which no transcript has.
  */
-export function makeFixture(dir, { dropSubagent = false, sessionId = 'sess-1' } = {}) {
+export function makeFixture(dir, { dropSubagent = false, sessionId = 'sess-1', snapshotWorkflowAgents = false } = {}) {
   const S = sessionId
   const OPUS = 'claude-opus-5-5'
   write(path.join(dir, `${S}.jsonl`), [
@@ -78,11 +84,13 @@ export function makeFixture(dir, { dropSubagent = false, sessionId = 'sess-1' }
   const subs = [
     ['a1', { agentType: 'wf-qa', description: 'K2 Story prüfen' }],
     ['a2', { agentType: 'wf-qa', description: 'Nachfrage klären' }],
-    ['a3', { agentType: 'general-purpose', description: 'irgendwas' }],
+    // an agent of the Workflow tool: one level deeper, with the phase in its meta
+    ['a3', { agentType: 'general-purpose', description: 'irgendwas', workflowPhase: 'Eingang' }, 'workflows/wf_1'],
   ]
-  for (const [id, meta] of dropSubagent ? [] : subs) {
-    write(path.join(dir, S, 'subagents', `agent-${id}.jsonl`), request(`req_${id}`, OPUS))
-    fs.writeFileSync(path.join(dir, S, 'subagents', `agent-${id}.meta.json`), JSON.stringify({ ...meta, spawnDepth: 1 }))
+  for (const [id, meta, sub = ''] of dropSubagent ? [] : subs) {
+    const d = path.join(dir, S, 'subagents', sub)
+    write(path.join(d, `agent-${id}.jsonl`), request(`req_${id}`, OPUS, {}, OPUS_USAGE, !(snapshotWorkflowAgents && sub)))
+    fs.writeFileSync(path.join(d, `agent-${id}.meta.json`), JSON.stringify({ ...meta, spawnDepth: 1 }))
   }
   return { dir, transcript: path.join(dir, `${S}.jsonl`), map: MAP }
 }

.agents/skills/bpmn2agent-cost/scripts/otel-sink.mjs

@@ -0,0 +1,23 @@
+#!/usr/bin/env node
+// Minimal OTLP/HTTP-JSON receiver for bpmn2agent-cost: appends every request body as one {path, body}
+// line, which `cost-report.mjs --otel <out.jsonl>` reads for exact per-request usage. No dependencies.
+// Usage: node otel-sink.mjs <out.jsonl> [port=4318]   (stop it with Ctrl-C / SIGTERM)
+import fs from 'node:fs'
+import http from 'node:http'
+const [out, port = '4318'] = process.argv.slice(2)
+if (!out) { console.error('usage: otel-sink.mjs <out.jsonl> [port]'); process.exit(2) }
+const server = http.createServer((req, res) => {
+  const chunks = []
+  req.on('data', (c) => chunks.push(c))
+  req.on('end', () => {
+    const text = Buffer.concat(chunks).toString('utf8')
+    let body = null
+    try { body = JSON.parse(text) } catch { body = { _raw: text.slice(0, 2000) } }
+    fs.appendFileSync(out, JSON.stringify({ path: req.url, body }) + '\n')
+    res.writeHead(200, { 'content-type': 'application/json' })
+    res.end('{}')
+  })
+})
+server.listen(Number(port), '127.0.0.1', () => console.log(`otel sink on 127.0.0.1:${port} -> ${out}`))
+process.on('SIGTERM', () => server.close(() => process.exit(0)))
+process.on('SIGINT', () => server.close(() => process.exit(0)))

.agents/skills/bpmn2agent-cost/scripts/read-usage.mjs

@@ -7,10 +7,14 @@
 // Layout it understands (internal Claude Code format, see docs/plans/laufkosten/spike.md):
 //   <dir>/<sessionId>.jsonl                     main thread, also holds the one `cost-state` entry
 //   <dir>/<sessionId>/subagents/agent-<id>.jsonl (+ .meta.json with agentType and description)
+//   <dir>/<sessionId>/subagents/workflows/<workflowRun>/agent-<id>.jsonl   agents of the Workflow tool;
+//       their meta.json also carries `workflowPhase`. Any depth below subagents/ is scanned.
 //
 // One API request spans several lines (one per content block). Only the last one carries the final
-// output_tokens, so each requestId keeps the line with the highest output_tokens. A request whose
-// final line has no stop_reason is incomplete: it is counted and reported, never silently summed.
+// output_tokens, so each requestId keeps the line with the highest output_tokens. Input and cache
+// tokens are exact on every line. A request whose best line has no stop_reason is `final: false`: its
+// output_tokens is only a streaming snapshot (the agents of the Workflow tool never get the final
+// value written back). It is kept, flagged, and cost-report decides how to complete it.
 // A line that looks like a request but lacks requestId/usage/model is a format change: fail loudly.
 
 import fs from 'node:fs'
@@ -57,6 +61,18 @@ function toUsage(u, where) {
   }
 }
 
+/** Every agent-<id>.jsonl below dir, at any depth (Workflow tool agents sit in workflows/<run>/). */
+function agentTranscripts(dir) {
+  if (!fs.existsSync(dir)) return []
+  const out = []
+  for (const e of fs.readdirSync(dir, { withFileTypes: true }).sort((a, b) => a.name.localeCompare(b.name))) {
+    const p = path.join(dir, e.name)
+    if (e.isDirectory()) out.push(...agentTranscripts(p))
+    else if (/^agent-.*\.jsonl$/.test(e.name)) out.push(p)
+  }
+  return out
+}
+
 function readThread(file, meta, requests, stats) {
   for (const { line, entry } of jsonLines(file)) {
     if (entry.type !== 'assistant') continue
@@ -68,6 +84,8 @@ function readThread(file, meta, requests, stats) {
       throw new Error(`${where}: assistant entry without requestId/usage/model (transcript format changed?)`)
     }
     const usage = toUsage(msg.usage, where)
+    const chars = (stats.chars ??= new Map())
+    chars.set(entry.requestId, (chars.get(entry.requestId) ?? 0) + JSON.stringify(msg.content ?? '').length)
     const cur = requests.get(entry.requestId)
     if (!cur || usage.output >= cur.usage.output) {
       requests.set(entry.requestId, {
@@ -77,6 +95,7 @@ function readThread(file, meta, requests, stats) {
         agentType: meta.agentType,
         description: meta.description,
         parentAgentId: meta.parentAgentId,
+        phase: meta.phase ?? null,
         attributionSkill: entry.attributionSkill ?? null,
         attributionMcpServer: entry.attributionMcpServer ?? null,
         attributionMcpTool: entry.attributionMcpTool ?? null,
@@ -106,24 +125,23 @@ export function readSession(transcript) {
 
   const subDir = path.join(path.dirname(transcript), sessionId, 'subagents')
   let files = 1
-  if (fs.existsSync(subDir)) {
-    for (const f of fs.readdirSync(subDir).filter((n) => /^agent-.*\.jsonl$/.test(n)).sort()) {
-      const agentId = f.slice('agent-'.length, -'.jsonl'.length)
-      const metaFile = path.join(subDir, `agent-${agentId}.meta.json`)
-      const m = fs.existsSync(metaFile) ? JSON.parse(fs.readFileSync(metaFile, 'utf8')) : {}
-      readThread(path.join(subDir, f), {
-        source: 'subagent', agentId, agentType: m.agentType ?? null, description: m.description ?? null,
-        parentAgentId: m.parentAgentId ?? null,
-      }, requests, stats)
-      files++
-    }
+  for (const file of agentTranscripts(subDir)) {
+    const f = path.basename(file)
+    const agentId = f.slice('agent-'.length, -'.jsonl'.length)
+    const metaFile = path.join(path.dirname(file), `agent-${agentId}.meta.json`)
+    const m = fs.existsSync(metaFile) ? JSON.parse(fs.readFileSync(metaFile, 'utf8')) : {}
+    readThread(file, {
+      source: 'subagent', agentId, agentType: m.agentType ?? null, description: m.description ?? null,
+      parentAgentId: m.parentAgentId ?? null, phase: m.workflowPhase ?? null,
+    }, requests, stats)
+    files++
   }
-  const all = [...requests.values()]
+  const all = [...requests.values()].map((r) => ({ ...r, final: !!r.stopReason, chars: stats.chars?.get(r.requestId) ?? 0 }))
   return {
     sessionId,
     version,
-    requests: all.filter((r) => r.stopReason),
-    incomplete: all.filter((r) => !r.stopReason).length,
+    requests: all,
+    incomplete: all.filter((r) => !r.final).length,
     costState: costState && {
       totalCostUSD: costState.totalCostUSD,
       hasUnknownModelCost: !!costState.hasUnknownModelCost,
@@ -134,13 +152,37 @@ export function readSession(transcript) {
   }
 }
 
+/**
+ * Reads the `api_request` log events of an OTLP/JSON telemetry dump (one {path, body} per line, as the
+ * pilot sink writes it) into Map<request_id, exact per-request usage and cost>. Only events of
+ * `sessionId` count. `cost_usd` is Claude Code's own estimate, so it is kept as a cross-check.
+ */
+export function readOtel(file, sessionId) {
+  const out = new Map()
+  for (const { entry } of jsonLines(file)) {
+    for (const rl of entry.body?.resourceLogs ?? []) for (const sl of rl.scopeLogs ?? []) for (const lr of sl.logRecords ?? []) {
+      const a = Object.fromEntries((lr.attributes ?? []).map((x) => [x.key, Object.values(x.value)[0]]))
+      if (a['event.name'] !== 'api_request' || !a.request_id || (sessionId && a['session.id'] !== sessionId)) continue
+      out.set(a.request_id, {
+        requestId: a.request_id, model: a.model, querySource: a.query_source ?? null,
+        usage: {
+          input: Number(a.input_tokens), output: Number(a.output_tokens), cacheRead: Number(a.cache_read_tokens),
+          cacheWriteTotal: Number(a.cache_creation_tokens), webSearch: 0,
+        },
+        costUsd: Number(a.cost_usd),
+      })
+    }
+  }
+  return out
+}
+
 /** Reads ledger.jsonl lines written by the generated cost-ledger hook. */
 export function readLedger(file) {
-  const requests = new Map() // keyed by requestId: a line written twice counts once
+  const requests = new Map() // keyed by requestId: the latest line wins (a request may be re-written once final)
   const costStates = new Map()
   for (const { entry } of jsonLines(file)) {
     if (entry.kind === 'cost-state') costStates.set(entry.sessionId, entry)
-    else if (entry.kind === 'request') requests.set(`${entry.sessionId}/${entry.requestId}`, entry)
+    else if (entry.kind === 'request') requests.set(`${entry.sessionId}/${entry.requestId}`, { ...entry, final: entry.final ?? true }) // the first hook version wrote only final rows
   }
   return { requests: [...requests.values()], costStates }
 }

.agents/skills/bpmn2agent-cost/scripts/test.mjs

@@ -5,7 +5,7 @@ import os from 'node:os'
 import path from 'node:path'
 import test from 'node:test'
 import { fileURLToPath } from 'node:url'
-import { buildReport, attribute } from './cost-report.mjs'
+import { buildReport, attribute, completeRequests, elementIdOf } from './cost-report.mjs'
 import { HAUS_COST, MAP, OPUS_COST, makeFixture } from './fixture.mjs'
 import { costOf, loadPrices, readSession, sumUsage } from './read-usage.mjs'
 
@@ -48,6 +48,71 @@ test('attribution order: description id, skill, lane, main, unassigned', () => {
   assert.deepEqual(by.req_a3, { type: 'unassigned', id: 'nicht zugeordnet' })
 })
 
+test('agents of the Workflow tool (subagents/workflows/<run>/) are read, with their phase', () => {
+  const s = readSession(makeFixture(tmp()).transcript)
+  const a3 = s.requests.find((r) => r.requestId === 'req_a3')
+  assert.equal(a3.source, 'subagent')
+  assert.equal(a3.phase, 'Eingang')
+  assert.equal(s.files, 4) // main + three agents, one of them nested
+})
+
+test('a composite dispatch label resolves to the inner element, an unknown one to nothing', () => {
+  assert.equal(elementIdOf('K2 Story prüfen', MAP), 'K2')
+  assert.equal(elementIdOf('Call_X/C1 Einordnen', MAP), 'C1')
+  assert.equal(elementIdOf('Call_X/ZZ nope', MAP), null)
+  assert.equal(elementIdOf(null, MAP), null)
+})
+
+test('phase: the map wins, else the agent\'s own workflowPhase, else none', () => {
+  const s = readSession(makeFixture(tmp()).transcript)
+  const rep = buildReport({ ...s, map: MAP, prices })
+  const by = Object.fromEntries(rep.byPhase.map((p) => [p.name, p.requests]))
+  assert.deepEqual(by, { 'Prüfung': 1, 'Eingang': 2, '(ohne Phase)': 2 }) // K2 by map; C1 by map + a3 by meta; main + QA lane agent
+})
+
+test('workflow agents with snapshot-only output: total stays exact, the missing output is an estimate', () => {
+  const s = readSession(makeFixture(tmp(), { snapshotWorkflowAgents: true }).transcript)
+  assert.equal(s.incomplete, 1)
+  const rep = buildReport({ ...s, map: MAP, prices })
+  assert.equal(rep.ok, true, JSON.stringify(rep.checks))
+  near(rep.totals.sessionUsd, 5 * OPUS_COST + HAUS_COST, 'exact total from cost-state')
+  assert.equal(rep.estimate.requests, 1)
+  near(rep.estimate.outputTokens, 2000 - 3, 'the exact gap')
+  near(rep.totals.estimatedUsd, (1997 * 20) / 1e6, 'only the missing output is estimated')
+  assert.match(rep.checks.map((c) => c.text).join('\n'), /sind geschätzt/)
+})
+
+test('without cost-state an incomplete request stays a flagged lower bound', () => {
+  const s = readSession(makeFixture(tmp(), { snapshotWorkflowAgents: true }).transcript)
+  const rep = buildReport({ ...s, costState: null, map: MAP, prices })
+  assert.equal(rep.estimate, null)
+  assert.match(rep.checks.map((c) => c.text).join('\n'), /nur eine Untergrenze/)
+})
+
+test('no estimate when tokens do not match (resumed session): lower bound instead of a wrong split', () => {
+  const s = readSession(makeFixture(tmp(), { snapshotWorkflowAgents: true }).transcript)
+  const requests = s.requests.filter((r) => r.requestId !== 'req_a1') // an earlier process is missing
+  const rep = buildReport({ ...s, requests, map: MAP, prices })
+  assert.equal(rep.estimate, null)
+  assert.match(rep.checks.map((c) => c.text).join('\n'), /Untergrenze/)
+})
+
+test('telemetry api_request events make it exact: output per request and the helper call', () => {
+  const s = readSession(makeFixture(tmp(), { snapshotWorkflowAgents: true }).transcript)
+  const otel = new Map([
+    ['req_a3', { requestId: 'req_a3', model: 'claude-opus-5-5', querySource: 'repl', usage: { input: 1000, output: 2000, cacheRead: 100000, cacheWriteTotal: 15000 }, costUsd: OPUS_COST }],
+    ['aux1', { requestId: 'aux1', model: 'claude-haiku-4-5-20251001', querySource: 'web_search_tool', usage: { input: 10000, output: 1000, cacheRead: 0, cacheWriteTotal: 0 }, costUsd: HAUS_COST }],
+  ])
+  const rep = buildReport({ ...s, otel, map: MAP, prices })
+  assert.equal(rep.ok, true, JSON.stringify(rep.checks))
+  assert.equal(rep.estimate, null)
+  near(rep.totals.sessionUsd, 5 * OPUS_COST + HAUS_COST, 'total')
+  near(rep.totals.auxiliaryUsd, 0, 'nothing left only in cost-state')
+  const aux = rep.byUnit.find((u) => u.type === 'auxiliary')
+  near(aux.costUsd, HAUS_COST, 'helper call priced as OpenTelemetry reports it')
+  assert.equal(completeRequests({ requests: s.requests, costState: null, otel, prices }).info.otel.matched, 1)
+})
+
 test('a skill shared by several elements is attributed to the skill, not guessed', () => {
   const r = { source: 'main', attributionSkill: 'story-check', description: null, agentType: null }
   assert.deepEqual(attribute(r, MAP), { type: 'skill', id: 'story-check' })
@@ -80,7 +145,7 @@ test('a missing subagent transcript (resumed session) is a warning with the reas
   // cost-state of the full session against only the main thread: tokens missing, not exceeding
   const rep = buildReport({ ...full, requests: s.requests, map: MAP, prices })
   assert.equal(rep.ok, true)
-  assert.match(rep.checks.map((c) => c.text).join('\n'), /fortgesetzt oder geleert/)
+  assert.match(rep.checks.map((c) => c.text).join('\n'), /fortgesetzte bzw\. geleerte Session/)
 })
 
 test('cost-state behind the transcript (resumed session) warns and the transcript sum counts', () => {