Brummel/Skills
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(postmortem): aggregate Workflow run logs per stage

Browse Source 
Sessions that ran Workflow-substrate pipelines read as near-zero
subagent spend: analyze_subagents() globs one level only, while
Workflow runs log under workflows/wf_*.json (per-agent summary) and
subagents/workflows/<runId>/agent-*.jsonl (raw usage). New
analyze_workflows() joins both — raw transcripts authoritative
(dedup by requestId, as everywhere), summary supplies stage labels
and models, label-prefix grouping (impl:3 -> impl), untracked
transcripts isolated under '?'. Report gains workflows[] and
workflow_totals; totals now include workflow spend.

Live check against a real 17-agent workflow run: raw 1.30M vs
reported 1.22M (the live counter's known ~7-11% under-tracking).

closes #28
This commit is contained in:
Claude
2026-07-17 15:23:22 +02:00
parent 22abd37458
commit 7a08259db7
3 changed files with 236 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
postmortem/SKILL.md
+19 -5
View File
@@ -1,6 +1,6 @@
---
name: postmortem
description: Use after a working session (or a `/boss` run) to grade how well the toolchain and the dispatched agents actually performed, and how many tokens the run spent. Reads the session's own Claude Code transcript + subagent logs, aggregates tokens/tool-health deterministically via a helper script, and writes a scored retrospective. Utility skill, user-invoked or orchestrator-invoked at run close; never a pipeline gate.
description: Use after a working session (or a `/boss` run) to grade how well the toolchain and the dispatched agents actually performed, and how many tokens the run spent. Reads the session's own Claude Code transcript + subagent logs + Workflow run logs, aggregates tokens/tool-health deterministically via a helper script, and writes a scored retrospective. Utility skill, user-invoked or orchestrator-invoked at run close; never a pipeline gate.
---
# postmortem — grade the run
@@ -10,9 +10,12 @@ description: Use after a working session (or a `/boss` run) to grade how well th
## Overview
Every session leaves a detailed flight recorder behind — the
JSONL transcript under `~/.claude/projects/<slug>/<session>.jsonl`
plus one sidechain log per dispatched subagent under
`<session>/subagents/`. Nobody reads it. So the same expensive
JSONL transcript under `~/.claude/projects/<slug>/<session>.jsonl`,
one sidechain log per dispatched subagent under
`<session>/subagents/`, and for every Workflow run a per-agent
summary in `<session>/workflows/wf_*.json` with raw per-agent
transcripts under `<session>/subagents/workflows/<runId>/`. Nobody
reads it. So the same expensive
habits repeat: a subagent fleet that bounces `BLOCKED`, a tool
loop with a 30% error rate, a cache-cold run that re-sends the full
input on every turn. This skill turns that recorder into a graded
@@ -55,7 +58,12 @@ Read before interpreting, so the report never overclaims:
timestamps, every `tool_use` with its input, every `tool_result`
with the authoritative `is_error` flag, skill invocations,
slash commands, and a full separate transcript + `agentType` /
`description` for each subagent.
`description` for each subagent. Workflow runs additionally carry a
per-agent summary (stage label, model, a live token counter) in
`workflows/wf_*.json`; the counter under-tracks vs. the deduped raw
transcripts (~11% in the corpus behind issue #28), so the script
treats the raw transcripts as authoritative and reports the counter
only as `reported_total_tokens` for cross-checking.
- **Absent:** `costUSD` is always `null`; `durationMs` is always
`null`. There is no metered cost in the log, so this skill does
not compute one — it reports **raw token counts** (kept split by
@@ -111,6 +119,12 @@ behind it is an opinion; drop it.
`NEEDS_CONTEXT` / `unknown`), `total_tokens`, and `requests`. A
fleet that burned real tokens to return `BLOCKED` is the headline
finding. Compare each agent's token spend to what it delivered.
For Workflow runs, grade per stage instead: `workflow_totals` and
each run's `stages` table (label-prefix grouping, e.g. all
`impl:N` under `impl`) show where the run's spend concentrated;
`agents_untracked > 0` flags retried/dropped attempts. Schema-bound
workflow stages carry no `terminal_status` — the workflow script's
own aggregation already judged them.
### Step 3 — Write the report
postmortem/scripts/postmortem.py
+93 -3
View File
@@ -2,8 +2,10 @@
"""Deterministic aggregator for a single Claude Code session.
Reads the main transcript JSONL for one session plus its subagent
sidechain logs, and emits a compact JSON summary on stdout that the
post-mortem SKILL interprets into a written report.
sidechain logs and its Workflow-substrate run logs (workflows/wf_*.json
+ subagents/workflows/<runId>/agent-*.jsonl), and emits a compact JSON
summary on stdout that the post-mortem SKILL interprets into a written
report.
Why a script and not skill prose: transcripts run to >1 MB and a
session can spawn dozens of subagent logs. Token accounting has two
@@ -266,6 +268,77 @@ def subagent_terminal_status(path):
return "unknown"
def analyze_workflows(session_dir):
"""Aggregate Workflow-substrate runs: workflows/wf_*.json (per-agent
summary: stage label, model, live token counter) joined with the raw
per-agent transcripts under subagents/workflows/<runId>/agent-*.jsonl.
The raw transcripts are authoritative for token counts (deduped by
requestId, same as every other transcript here); the wf_*.json
'tokens' counters under-track vs. the deduped raw usage (~11% in the
206-run corpus that motivated this path, issue #28) and are reported
only as a cross-check. Stage = the label up to the first ':'
('impl:3' -> 'impl'), so loop stages aggregate across tasks/rounds."""
runs = []
for wf_path in sorted(glob.glob(os.path.join(session_dir, "workflows",
"wf_*.json"))):
try:
with open(wf_path) as fh:
data = json.load(fh)
except (json.JSONDecodeError, OSError):
continue
run_id = data.get("runId") or os.path.basename(wf_path)[:-len(".json")]
raw_dir = os.path.join(session_dir, "subagents", "workflows", run_id)
meta_by_agent = {}
for entry in data.get("workflowProgress") or []:
if entry.get("type") != "workflow_agent" or not entry.get("agentId"):
continue
meta_by_agent[entry["agentId"]] = entry
run_tokens = empty_tokens()
stages = {}
agent_count = 0
untracked = 0
for jl in sorted(glob.glob(os.path.join(raw_dir, "agent-*.jsonl"))):
aid = os.path.basename(jl)[len("agent-"):-len(".jsonl")]
agg = aggregate_transcript(jl)
entry = meta_by_agent.get(aid)
if entry is None:
# Transcript with no progress entry (retried/dropped
# attempt): keep it counted, grouped under '?', never
# silently folded into a named stage.
untracked += 1
label = (entry or {}).get("label") or "?"
stage = label.split(":", 1)[0]
st = stages.setdefault(stage, {"count": 0, "tokens": empty_tokens(),
"models": {}})
st["count"] += 1
agent_count += 1
for k in run_tokens:
run_tokens[k] += agg["tokens"][k]
st["tokens"][k] += agg["tokens"][k]
fallback_model = (entry or {}).get("model")
for m, n in agg["models"].items():
key = m or fallback_model
st["models"][key] = st["models"].get(key, 0) + n
for st in stages.values():
st["total_tokens"] = token_total(st["tokens"])
runs.append({
"run_id": run_id,
"name": data.get("workflowName"),
"status": data.get("status"),
"agents": agent_count,
"agents_untracked": untracked,
"reported_total_tokens": data.get("totalTokens"),
"tokens": run_tokens,
"total_tokens": token_total(run_tokens),
"stages": stages,
})
return runs
def analyze_subagents(session_dir):
out = []
sub_dir = os.path.join(session_dir, "subagents")
@@ -335,17 +408,27 @@ def main():
main_meta = analyze_main(main_path)
subagents = []
workflows = []
if session_dir and os.path.isdir(session_dir):
subagents = analyze_subagents(session_dir)
workflows = analyze_workflows(session_dir)
sub_tokens = empty_tokens()
for s in subagents:
for k in sub_tokens:
sub_tokens[k] += s["tokens"][k]
wf_tokens = empty_tokens()
wf_agents = 0
for w in workflows:
wf_agents += w["agents"]
for k in wf_tokens:
wf_tokens[k] += w["tokens"][k]
total_tokens = empty_tokens()
for k in total_tokens:
total_tokens[k] = main_agg["tokens"][k] + sub_tokens[k]
total_tokens[k] = (main_agg["tokens"][k] + sub_tokens[k]
+ wf_tokens[k])
warnings = []
is_active = False
@@ -396,6 +479,13 @@ def main():
"tokens": sub_tokens,
"total_tokens": token_total(sub_tokens),
},
"workflows": workflows,
"workflow_totals": {
"runs": len(workflows),
"agents": wf_agents,
"tokens": wf_tokens,
"total_tokens": token_total(wf_tokens),
},
"totals": {
"tokens": total_tokens,
"total_tokens": token_total(total_tokens),
postmortem/tests/test_workflow_aggregation.py
+124
View File
@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""Executable spec for postmortem.analyze_workflows.
Run directly: `python3 postmortem/tests/test_workflow_aggregation.py`.
Exits 0 on pass, 1 on failure. No pytest dependency — same plain-assert
convention as test_terminal_status.py.
Pins issue #28: sessions that ran Workflow-substrate pipelines
(workflows/wf_*.json + subagents/workflows/<runId>/agent-*.jsonl) must
not read as zero subagent spend. The raw transcripts are authoritative
(dedup by requestId); the summary supplies stage labels; a transcript
missing from workflowProgress groups under '?' instead of being folded
into a named stage or dropped.
"""
import json
import os
import sys
import tempfile
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
from postmortem import analyze_workflows # noqa: E402
def _write(path, obj_lines):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as fh:
for obj in obj_lines:
fh.write(json.dumps(obj) + "\n")
def _assistant_line(request_id, in_tok, out_tok, model="claude-sonnet-5"):
return {
"type": "assistant",
"requestId": request_id,
"message": {
"model": model,
"usage": {"input_tokens": in_tok, "output_tokens": out_tok,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0},
},
}
def main():
with tempfile.TemporaryDirectory() as session_dir:
run_id = "wf_test-1"
wf = {
"runId": run_id,
"workflowName": "spec-fixture",
"status": "completed",
"totalTokens": 999, # live counter: deliberately wrong
"defaultModel": "claude-sonnet-5",
"workflowProgress": [
{"type": "workflow_phase", "index": 1, "title": "P"},
{"type": "workflow_agent", "agentId": "a1",
"label": "impl:1", "model": "claude-sonnet-5",
"state": "done", "tokens": 1},
{"type": "workflow_agent", "agentId": "a2",
"label": "impl:2", "model": "claude-sonnet-5",
"state": "done", "tokens": 1},
{"type": "workflow_agent", "agentId": "a4",
"label": "qual:1", "model": "claude-opus-4-8",
"state": "error", "tokens": 0}, # no transcript on disk
],
}
os.makedirs(os.path.join(session_dir, "workflows"))
with open(os.path.join(session_dir, "workflows",
run_id + ".json"), "w") as fh:
json.dump(wf, fh)
raw = os.path.join(session_dir, "subagents", "workflows", run_id)
# a1: streamed repeat of the SAME requestId — must dedup to
# max-per-field (input 10, output 7), not sum to 20/12.
_write(os.path.join(raw, "agent-a1.jsonl"),
[_assistant_line("req-1", 10, 5),
_assistant_line("req-1", 10, 7)])
# a2: two distinct requests — must sum.
_write(os.path.join(raw, "agent-a2.jsonl"),
[_assistant_line("req-2", 3, 2),
_assistant_line("req-3", 4, 1)])
# a3: transcript with NO workflowProgress entry — counts under '?'.
_write(os.path.join(raw, "agent-a3.jsonl"),
[_assistant_line("req-4", 100, 50)])
runs = analyze_workflows(session_dir)
assert len(runs) == 1, runs
run = runs[0]
assert run["run_id"] == run_id
assert run["name"] == "spec-fixture"
assert run["reported_total_tokens"] == 999
# Three transcripts on disk, one untracked.
assert run["agents"] == 3, run
assert run["agents_untracked"] == 1, run
stages = run["stages"]
assert set(stages) == {"impl", "?"}, stages
impl = stages["impl"]
assert impl["count"] == 2, impl
# a1 deduped (10 in / 7 out) + a2 summed (7 in / 3 out)
assert impl["tokens"]["input"] == 17, impl
assert impl["tokens"]["output"] == 10, impl
assert impl["total_tokens"] == 27, impl
assert impl["models"] == {"claude-sonnet-5": 3}, impl
unk = stages["?"]
assert unk["count"] == 1, unk
assert unk["tokens"]["input"] == 100 and unk["tokens"]["output"] == 50, unk
# Run totals = authoritative raw aggregation, not the live counter.
assert run["tokens"]["input"] == 117, run
assert run["tokens"]["output"] == 60, run
assert run["total_tokens"] == 177, run
print("ok: analyze_workflows aggregates raw transcripts per stage, "
"dedups by requestId, and isolates untracked agents under '?'")
return 0
if __name__ == "__main__":
sys.exit(main())