209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
"""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,
|
|
)
|
|
)
|