"""Summarize measured trace usage; never infer token counts from text length.""" import argparse import hashlib import json import sqlite3 from collections import Counter, defaultdict from pathlib import Path from statistics import mean def enrich(row, data, traces): """Read actual evidence separately; the fixture never feeds the agent.""" session_root = traces / row["trace_id"] outcome = session_root / "run_outcome.json" if outcome.exists(): value = json.loads(outcome.read_text("utf-8")) row["ai_verdict"] = value.get("test_result", {}).get("status") row["ai_verdict_reason"] = value.get("test_result", {}).get("reason") row["outcome_code"] = value.get("code") recordings = session_root / "evidence/recordings/recordings.json" row["recordings"] = [] if recordings.exists(): for source in json.loads(recordings.read_text("utf-8")).get("sources", []): row["recordings"].append( { "scope": source.get("scope"), "status": source.get("status"), "area": source.get("area"), "segments": [ { **segment, "exists": (session_root / segment["file"]).is_file(), } for segment in source.get("segments", []) ], } ) proof = session_root / "artifacts/plugins" / row["plugin"] / "proof.png" row["proof_sha256"] = ( hashlib.sha256(proof.read_bytes()).hexdigest() if proof.is_file() else None ) start = data["session"]["start_time"] finish = next( (e["at"] / 1000 for e in data["fixture"]["events"] if e["event"] == "finish"), None, ) row["fixture_finish_seconds"] = ( round(finish - start, 3) if finish is not None else None ) row["evidence_ready_seconds"] = ( round(max(finish, proof.stat().st_mtime) - start, 3) if finish is not None and proof.is_file() and row["actual_pass"] else None ) row["post_evidence_seconds"] = ( round(row["seconds"] - row["evidence_ready_seconds"], 3) if row["evidence_ready_seconds"] is not None else None ) with sqlite3.connect( f"file:{(traces / 'data_engine.db').as_posix()}?mode=ro", uri=True ) as db: errors = [] observed_plugins = set() for status, payload in db.execute( "SELECT status,payload FROM traces WHERE session_id=? AND type='tool' ORDER BY timestamp", (row["trace_id"],), ): value = json.loads(payload or "{}") if value.get("plugin_id"): observed_plugins.add(value["plugin_id"]) if value.get("plugin_id") and status != "success": # Keep the diagnostic first line, not private window inventory. message = str(value.get("error", "")).splitlines() errors.append( { "plugin": value["plugin_id"], "tool": value.get("tool_name"), "message": message[0] if message else "", } ) row["tool_errors"] = errors row["observed_plugins"] = sorted(observed_plugins) def summarize(root, traces=None): rows = [] for path in sorted(root.glob("*-*/result.json")): data = json.loads(path.read_text("utf-8")) usage = data["llm_usage"] calls = data["plugin_calls"] session = data["session"] requests = data.get("requests", []) usage_by_node = defaultdict( lambda: {"calls": 0, "input": 0, "cached": 0, "output": 0} ) models_from_usage = Counter() for item in usage: source_parts = item.get("source", "").split(":") model_index = 2 if source_parts[0] == "lens" else 1 model = ( source_parts[model_index] if len(source_parts) > model_index else "unknown" ) models_from_usage[model] += 1 node = usage_by_node[item.get("node", "unknown")] node["calls"] += 1 node["input"] += item.get("prompt_tokens", 0) node["cached"] += item.get("cached_tokens", 0) node["output"] += item.get("completion_tokens", 0) models = Counter( p.get("model", "unknown") for p in requests if p.get("phase") == "completed" ) input_tokens = sum(u.get("prompt_tokens", 0) for u in usage) cached = sum(u.get("cached_tokens", 0) for u in usage) output_tokens = sum(u.get("completion_tokens", 0) for u in usage) duration = (session.get("end_time") or session["start_time"]) - session[ "start_time" ] row = { "plugin": data["plugin"], "round": data["repetition"], "trace_id": data["trace_id"], "actual_pass": data["actual_pass"], "execution_status": session["status"], "input_tokens": input_tokens, "cached_tokens": cached, "uncached_input": input_tokens - cached, "output_tokens": output_tokens, "total_tokens": input_tokens + output_tokens, "cache_rate": cached / input_tokens if input_tokens else None, "llm_calls": len(usage), "models": dict(models), "models_from_usage": dict(models_from_usage), "usage_by_node": dict(usage_by_node), "request_phases": dict(Counter(p.get("phase") for p in requests)), "request_seconds_by_role": { role: round( sum( p.get("elapsed_seconds", 0) for p in requests if p.get("role") == role ), 3, ) for role in sorted({p.get("role", "unknown") for p in requests}) }, "fallback_requests": sum(bool(p.get("is_fallback")) for p in requests), "retried_requests": sum((p.get("attempt") or 1) > 1 for p in requests), "executor_text_only_responses": sum( p.get("role") == "FlashRunner" and p.get("phase") == "completed" and not p.get("tools") for p in requests ), "seconds": round(duration, 3), "plugin_calls": len(calls), "plugin_seconds": round(sum(c.get("duration") or 0 for c in calls), 3), "plugin_errors": sum(c["status"] != "success" for c in calls), "scroll_calls": sum(c["name"] == "scroll" for c in calls), "fixture": data["fixture"], } if traces is not None: enrich(row, data, traces) rows.append(row) aggregates = {} for plugin in sorted({r["plugin"] for r in rows}): group = [r for r in rows if r["plugin"] == plugin] aggregate = {"runs": len(group), "passed": sum(r["actual_pass"] for r in group)} for key in ( "input_tokens", "cached_tokens", "uncached_input", "output_tokens", "total_tokens", "seconds", "llm_calls", "plugin_calls", "plugin_seconds", "plugin_errors", "scroll_calls", ): aggregate[f"mean_{key}"] = round(mean(r[key] for r in group), 3) aggregate["cache_rate"] = sum(r["cached_tokens"] for r in group) / max( 1, sum(r["input_tokens"] for r in group) ) aggregates[plugin] = aggregate report = {"runs": rows, "aggregate": aggregates} (root / "summary.json").write_text( json.dumps(report, ensure_ascii=False, indent=2), "utf-8" ) return report if __name__ == "__main__": parser = argparse.ArgumentParser(__doc__) parser.add_argument("directory", type=Path) parser.add_argument("--traces", type=Path) options = parser.parse_args() print( json.dumps( summarize(options.directory, options.traces)["aggregate"], ensure_ascii=False, indent=2, ) )