#!/usr/bin/env python3 """review_reflex.py - is the reviewer agent you defined ever actually called? Reads Claude Code session logs (~/.claude/projects/**/*.jsonl) and counts, per session, who edited files and whether any review step ran. Prints counts only: no prompts, no file contents, no paths from inside the transcripts. What it counts - tool calls by name, from assistant tool_use blocks only (not from the system-prompt blobs the log also contains, which is why `grep Agent` lies) - Agent calls by subagent_type, and which of those types look like a reviewer - Edit/Write/NotebookEdit/MultiEdit calls made by the main session, and the same calls made inside sub-agent transcripts, attributed to the sub-agent's type through the agentId the parent log records when it launches one - read-only `codex exec` runs in Bash (a second model used as a reviewer) - Skill calls whose name matches the reviewer pattern (e.g. code-review) - a session is "producing" if it made at least one file mutation, and "reviewed" if it did at least one of the three review things above - agent definition files: which ones look like reviewers, whether they can Write/Edit, whether they contain verdict vocabulary, and how often each one was called in the logs Usage review_reflex.py # ~/.claude/projects, all time review_reflex.py --since 2026-10-01 # sessions with a message on/after this date review_reflex.py --defs ~/.claude/agents --defs ./.claude/agents --defs ~/.claude/plugins/cache review_reflex.py --reviewer 'critic|review|verif|audit|gate|judge' review_reflex.py --per-project # one line per project directory (names hashed) review_reflex.py --json out.json # machine-readable copy of the counts Standard library only. Python 3.8+. Exit 0 always; this is a measurement, not a gate. """ import argparse, collections, hashlib, json, os, re, sys MUTATORS = {"Edit", "Write", "NotebookEdit", "MultiEdit"} DEFAULT_REVIEWER = r"critic|review|verif|audit|gate|judge|checker|qa\b" VERDICT_RE = re.compile(r"\b(PASS|FAIL|verdict|approve|reject|Y/N|blocker|findings|severity)\b", re.I) CODEX_RO_RE = re.compile(r"codex\s+exec\b.*(--sandbox\s+read-only|-s\s+read-only)", re.S) CODEX_ANY_RE = re.compile(r"codex\s+exec\b") def iter_json(path): with open(path, encoding="utf-8", errors="replace") as f: for line in f: if not line.startswith("{"): continue try: yield json.loads(line) except ValueError: continue def tool_uses(rec): """Yield tool_use blocks from an assistant record. Nothing else in the log counts.""" if rec.get("type") != "assistant": return content = (rec.get("message") or {}).get("content") if not isinstance(content, list): return for b in content: if isinstance(b, dict) and b.get("type") == "tool_use": yield b def scan_transcript(path, reviewer_re): """Counts for one transcript (main session or sub-agent).""" c = {"tools": collections.Counter(), "agents": collections.Counter(), "mutations": 0, "codex_ro": 0, "codex_any": 0, "review_skills": 0, "agent_ids": {}, "first": None, "last": None, "user_msgs": 0} pending = {} # tool_use_id -> subagent_type, resolved when the launch result arrives  for rec in iter_json(path): ts = rec.get("timestamp") if ts and rec.get("type") in ("assistant", "user"): c["first"] = ts if c["first"] is None or ts < c["first"] else c["first"] c["last"] = ts if c["last"] is None or ts > c["last"] else c["last"] if rec.get("type") == "user" and not rec.get("isSidechain") and isinstance((rec.get("message") or {}).get("content"), str): c["user_msgs"] += 1 tur = rec.get("toolUseResult") if isinstance(tur, dict) and tur.get("agentId"): # link launch result -> the Agent call it answers  content = (rec.get("message") or {}).get("content") if isinstance(content, list): for b in content: if isinstance(b, dict) and b.get("type") == "tool_result" and b.get("tool_use_id") in pending: c["agent_ids"][tur["agentId"]] = pending[b["tool_use_id"]] for b in tool_uses(rec): name = b.get("name") or "?" c["tools"][name] += 1 inp = b.get("input") if isinstance(b.get("input"), dict) else {} if name in MUTATORS: c["mutations"] += 1 elif name in ("Agent", "Task"): st = str(inp.get("subagent_type") or "general-purpose") c["agents"][st] += 1 pending[b.get("id")] = st elif name == "Bash": cmd = str(inp.get("command") or "") if CODEX_ANY_RE.search(cmd): c["codex_any"] += 1 if CODEX_RO_RE.search(cmd): c["codex_ro"] += 1 elif name == "Skill": if reviewer_re.search(str(inp.get("skill") or "")): c["review_skills"] += 1 return c def find_sessions(root): """Yield (project_dir, session_path, [subagent paths]) for every main transcript under root.""" for proj in sorted(os.listdir(root)): pdir = os.path.join(root, proj) if not os.path.isdir(pdir) or proj.startswith("."): continue for name in sorted(os.listdir(pdir)): if not name.endswith(".jsonl"): continue sid = name[:-6] subs = [] sdir = os.path.join(pdir, sid, "subagents") if os.path.isdir(sdir): for dp, _, fn in os.walk(sdir): subs += [os.path.join(dp, f) for f in fn if f.startswith("agent-") and f.endswith(".jsonl")] yield proj, os.path.join(pdir, name), subs def parse_frontmatter(text): m = re.match(r"^---\s*\n(.*?)\n---", text, re.S) fm = {} if not m: return fm for line in m.group(1).splitlines(): if ":" in line and not line.startswith((" ", "\t")): k, v = line.split(":", 1) fm[k.strip()] = v.split("#", 1)[0].strip() return fm def scan_definitions(dirs, reviewer_re): out = [] seen = set() for d in dirs: d = os.path.expanduser(d) if not os.path.isdir(d): continue for dp, dn, fn in os.walk(d): dn[:] = [x for x in dn if x not in ("node_modules", "dist", "tests", "test")] if os.path.basename(dp) != "agents": continue for f in fn: if not f.endswith(".md"): continue p = os.path.join(dp, f) try: text = open(p, encoding="utf-8", errors="replace").read() except OSError: continue fm = parse_frontmatter(text) name = fm.get("name") or f[:-3] if name in seen: continue seen.add(name) tools = fm.get("tools", "") dis = fm.get("disallowedTools", "") can_write = (not tools or tools in ("*", "All tools") or re.search(r"\b(Write|Edit)\b", tools)) and not re.search(r"\b(Write|Edit)\b", dis) is_rev = bool(reviewer_re.search(name)) # name only; descriptions over-match ("verification crawler")  desc_hit = bool(reviewer_re.search(fm.get("description", ""))) and not is_rev out.append({"name": name, "reviewer": is_rev, "desc_hit": desc_hit, "can_write": bool(can_write), "verdict_vocab": bool(VERDICT_RE.search(text)), "where": os.path.basename(os.path.dirname(dp)) or d}) return out def main(): ap = argparse.ArgumentParser() ap.add_argument("roots", nargs="*", default=[os.path.expanduser("~/.claude/projects")]) ap.add_argument("--defs", action="append", help="directory tree holding agents/*.md definitions (repeatable)") ap.add_argument("--reviewer", default=DEFAULT_REVIEWER, help="regex that marks a subagent_type/skill/definition as a reviewer") ap.add_argument("--since", help="keep sessions whose last message is on/after YYYY-MM-DD") ap.add_argument("--per-project", action="store_true") ap.add_argument("--json") a = ap.parse_args() reviewer_re = re.compile(a.reviewer, re.I) tot = collections.Counter() tools = collections.Counter() agents = collections.Counter() sub_mut_by_type = collections.Counter() per_proj = collections.defaultdict(collections.Counter) n_bytes = 0 for root in a.roots: root = os.path.expanduser(root) for proj, spath, subs in find_sessions(root): m = scan_transcript(spath, reviewer_re) if a.since and (m["last"] or "")[:10] < a.since: continue n_bytes += os.path.getsize(spath) + sum(os.path.getsize(s) for s in subs) tot["sessions"] += 1 tot["subagent_files"] += len(subs) tools.update(m["tools"]) agents.update(mSource：https://dev.to/morganvale/is-your-reviewer-agent-ever-actually-called-counting-it-from-session-logs-1c5d
My favorites