Skills › Compound Engineering › ce-code-review › run-log.py
run-log.py
280 lines · skills/ce-code-review/scripts/run-log.py at 3449230
1#!/usr/bin/env python3
2"""Record what a ce-code-review run cost, stage by stage.
4`event` appends one JSON line per stage boundary to `<run-dir>/stages.jsonl`
5as the run progresses, so a killed run still leaves parseable partial data.
6`summarize` folds those lines, the run directory's artifact bytes, and any
7peer usage file into a `cost` object merged into `<run-dir>/metadata.json`
8without touching the fields already there. Neither subcommand ever fails
9the run: an error leaves `cost.status` as `unavailable` with the reason.
10"""
12from __future__ import annotations
14import argparse
15import json
16import os
17import sys
18from datetime import datetime, timezone
19from pathlib import Path
21STAGES_FILE = "stages.jsonl"
22METADATA_FILE = "metadata.json"
23# Stages whose end marks the receipt being written; a run with one of these
24# ended and no dangling start is complete.
25RECEIPT_STAGES = {"report", "receipt"}
26# The job directory holds detached-runner state, not review output.
27EXCLUDED_DIRS = {"jobs"}
28# Stages that produce candidates; merge and validate only filter them, so
29# their counts stay per stage and never enter the total.
30PRODUCING_STAGES = {"review", "peer", "dispatch"}
33def now_iso() -> str:
34 return datetime.now(timezone.utc).isoformat()
37def parse_iso(value: str) -> datetime | None:
38 try:
39 return datetime.fromisoformat(value)
40 except (TypeError, ValueError):
41 return None
44def host_from_env() -> str:
45 """The same attestation the cross-model reference uses, host only."""
46 env = os.environ
47 seen = [
48 host
49 for host, present in (
50 ("claude", env.get("CLAUDECODE") == "1"),
51 ("codex", any(env.get(k) for k in ("CODEX_SANDBOX", "CODEX_SANDBOX_NETWORK_DISABLED", "CODEX_SESSION_ID", "CODEX_THREAD_ID", "CODEX_CI"))),
52 ("grok", env.get("GROK_AGENT") == "1" or bool(env.get("GROK_SESSION_ID"))),
53 )
54 if present
55 ]
56 if env.get("OMPCODE"):
57 return "omp"
58 if len(seen) == 1:
59 return seen[0]
60 if seen:
61 return "unknown"
62 if env.get("CURSOR_AGENT") or env.get("CURSOR_CONVERSATION_ID"):
63 return "cursor"
64 if env.get("OPENCODE_TERMINAL"):
65 return "opencode"
66 return "unknown"
69def parse_facts(items: list[str] | None) -> dict[str, object]:
70 facts: dict[str, object] = {}
71 for item in items or []:
72 key, sep, raw = item.partition("=")
73 if not sep or not key:
74 continue
75 try:
76 facts[key] = json.loads(raw)
77 except ValueError:
78 facts[key] = raw
79 return facts
82def append_line(run_dir: Path, record: dict[str, object]) -> None:
83 run_dir.mkdir(parents=True, exist_ok=True)
84 line = json.dumps(record, sort_keys=True) + "\n"
85 with (run_dir / STAGES_FILE).open("a", encoding="utf-8") as handle:
86 handle.write(line)
87 handle.flush()
90def cmd_event(args: argparse.Namespace) -> int:
91 run_dir = Path(args.run_dir)
92 if not args.end and not args.start:
93 print("run-log event: nothing to record; pass --end and/or --start", file=sys.stderr)
94 return 0
95 ts = now_iso()
96 extras: dict[str, object] = {}
97 if args.reviewers is not None:
98 extras["reviewers"] = args.reviewers
99 if args.candidates is not None:
100 extras["candidates"] = args.candidates
101 if args.tokens is not None:
102 extras["tokens"] = args.tokens
103 if args.note:
104 extras["note"] = args.note
105 facts = parse_facts(args.fact)
106 if facts:
107 extras["facts"] = facts
108 written = []
109 if args.end:
110 record = {"ts": ts, "stage": args.end, "phase": "end", **extras}
111 append_line(run_dir, record)
112 written.append(record)
113 if args.start:
114 record = {"ts": ts, "stage": args.start, "phase": "start"}
115 # Facts and counts describe the stage that just ended; a lone --start
116 # carries them so the scope stage can record its helper facts.
117 if not args.end:
118 record.update(extras)
119 append_line(run_dir, record)
120 written.append(record)
121 print(json.dumps({"written": written}, sort_keys=True))
122 return 0
125def read_events(run_dir: Path) -> tuple[list[dict[str, object]], int]:
126 path = run_dir / STAGES_FILE
127 if not path.is_file():
128 raise FileNotFoundError(str(path))
129 events: list[dict[str, object]] = []
130 truncated = 0
131 with path.open("r", encoding="utf-8") as handle:
132 for raw in handle:
133 line = raw.strip()
134 if not line:
135 continue
136 try:
137 parsed = json.loads(line)
138 except ValueError:
139 truncated += 1
140 continue
141 if isinstance(parsed, dict) and "stage" in parsed and "phase" in parsed:
142 events.append(parsed)
143 else:
144 truncated += 1
145 return events, truncated
148def artifact_bytes(run_dir: Path) -> int:
149 total = 0
150 for root, dirs, files in os.walk(run_dir):
151 dirs[:] = [d for d in dirs if d not in EXCLUDED_DIRS]
152 for name in files:
153 try:
154 total += (Path(root) / name).stat().st_size
155 except OSError:
156 continue
157 return total
160def peer_usage(run_dir: Path) -> dict[str, object] | None:
161 for path in sorted(run_dir.glob("adversarial-*-usage.json")):
162 try:
163 data = json.loads(path.read_text(encoding="utf-8"))
164 except (OSError, ValueError):
165 continue
166 if isinstance(data, dict):
167 provider = path.name[len("adversarial-"):-len("-usage.json")]
168 return {"provider": provider, **data}
169 return None
172def build_cost(run_dir: Path) -> dict[str, object]:
173 events, truncated = read_events(run_dir)
174 stages: list[dict[str, object]] = []
175 order: list[str] = []
176 open_starts: dict[str, dict[str, object]] = {}
177 by_name: dict[str, dict[str, object]] = {}
178 scope_facts: dict[str, object] = {}
179 for event in events:
180 name = str(event["stage"])
181 if name not in by_name:
182 by_name[name] = {"stage": name}
183 order.append(name)
184 entry = by_name[name]
185 if event["phase"] == "start":
186 open_starts[name] = event
187 entry["started_at"] = event.get("ts")
188 elif event["phase"] == "end":
189 entry["ended_at"] = event.get("ts")
190 start = open_starts.pop(name, None)
191 if start is not None:
192 t0, t1 = parse_iso(str(start.get("ts"))), parse_iso(str(event.get("ts")))
193 if t0 and t1:
194 entry["elapsed_seconds"] = round((t1 - t0).total_seconds(), 3)
195 for key in ("reviewers", "candidates", "tokens", "note"):
196 if key in event:
197 entry[key] = event[key]
198 facts = event.get("facts")
199 if isinstance(facts, dict):
200 entry.setdefault("facts", {}).update(facts)
201 if name == "scope":
202 scope_facts.update(facts)
203 stages = [by_name[name] for name in order]
204 dangling = sorted(open_starts)
205 receipt_done = any(s.get("ended_at") for s in stages if s["stage"] in RECEIPT_STAGES)
206 status = "complete" if receipt_done and not dangling else "partial"
207 totals = {
208 "elapsed_seconds": round(sum(float(s.get("elapsed_seconds", 0) or 0) for s in stages), 3),
209 "reviewers": sum(int(s.get("reviewers", 0) or 0) for s in stages),
210 "candidates": sum(int(s.get("candidates", 0) or 0) for s in stages if s["stage"] in PRODUCING_STAGES),
211 "tokens": sum(int(s.get("tokens", 0) or 0) for s in stages if "tokens" in s),
212 "artifact_bytes": artifact_bytes(run_dir),
213 }
214 if not any("tokens" in s for s in stages):
215 del totals["tokens"]
216 cost: dict[str, object] = {
217 "status": status,
218 "host": host_from_env(),
219 "stages": stages,
220 "totals": totals,
221 "truncated_events": truncated,
222 }
223 if dangling:
224 cost["dangling_stages"] = dangling
225 if scope_facts:
226 cost["scope"] = scope_facts
227 peer = peer_usage(run_dir)
228 if peer is not None:
229 cost["peer"] = peer
230 return cost
233def cmd_summarize(args: argparse.Namespace) -> int:
234 run_dir = Path(args.run_dir)
235 try:
236 cost = build_cost(run_dir)
237 except FileNotFoundError as exc:
238 cost = {"status": "unavailable", "reason": f"no stage log: {exc}"}
239 except OSError as exc:
240 cost = {"status": "unavailable", "reason": str(exc)}
241 metadata_path = run_dir / METADATA_FILE
242 metadata: dict[str, object] = {}
243 if metadata_path.is_file():
244 try:
245 loaded = json.loads(metadata_path.read_text(encoding="utf-8"))
246 if isinstance(loaded, dict):
247 metadata = loaded
248 except (OSError, ValueError):
249 cost.setdefault("notes", []).append("metadata.json was unreadable and was rewritten")
250 metadata["cost"] = cost
251 run_dir.mkdir(parents=True, exist_ok=True)
252 tmp = metadata_path.with_suffix(".json.tmp")
253 tmp.write_text(json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8")
254 os.replace(tmp, metadata_path)
255 print(json.dumps({"status": cost.get("status"), "metadata": str(metadata_path)}, sort_keys=True))
256 return 0
259def main() -> int:
260 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
261 sub = parser.add_subparsers(dest="command", required=True)
262 ev = sub.add_parser("event", help="record a stage boundary")
263 ev.add_argument("--run-dir", required=True)
264 ev.add_argument("--end", help="stage that just ended")
265 ev.add_argument("--start", help="stage that starts now")
266 ev.add_argument("--reviewers", type=int)
267 ev.add_argument("--candidates", type=int)
268 ev.add_argument("--tokens", type=int)
269 ev.add_argument("--note")
270 ev.add_argument("--fact", action="append", help="key=value (value parsed as JSON when possible)")
271 ev.set_defaults(func=cmd_event)
272 sm = sub.add_parser("summarize", help="fold the stage log into metadata.json")
273 sm.add_argument("--run-dir", required=True)
274 sm.set_defaults(func=cmd_summarize)
275 args = parser.parse_args()
276 return args.func(args)
279if __name__ == "__main__":
280 raise SystemExit(main())