エージェントに任せた仕事の「やり直し率」を測る — 委譲境界を勘ではなく数字で引く

#!/usr/bin/env python3
"""rework_rate.py — エージェント成果のやり直し率をgit履歴から算出する。"""
import subprocess, json, math
from collections import defaultdict
 
WINDOW_HOURS = 72        # この時間内の人手修正をやり直しとみなす
WINDOW_COMMITS = 5       # かつ、同一ファイルを触る次のNコミット以内
 
def git(*args):
    out = subprocess.run(["git", *args], capture_output=True, text=True, check=True)
    return out.stdout
 
def load_commits():
    """新しい順にコミットのメタデータとファイル一覧を読む。"""
    fmt = "%H%x1f%ct%x1f%B%x1e"
    raw = git("log", "--no-merges", f"--pretty=format:{fmt}", "--name-only")
    commits = []
    for block in raw.split("\x1e"):
        block = block.strip("\n")
        if not block:
            continue
        head, *file_lines = block.split("\n")
        sha, ts, body = head.split("\x1f")
        files = {f for f in file_lines if f.strip()}
        cat, agent = None, False
        for line in body.splitlines():
            if line.startswith("Antigravity-Category:"):
                cat = line.split(":", 1)[1].strip()
            if line.startswith("Antigravity-Agent:"):
                agent = True
        commits.append({"sha": sha, "ts": int(ts), "files": files,
                        "category": cat, "agent": agent, "body": body})
    return commits
 
def is_rework(body):
    low = body.lower()
    return any(k in low for k in ("fix", "revert", "rework", "hotfix"))
 
def compute(commits):
    ordered = list(reversed(commits))   # 古い順に走査したい
    stats = defaultdict(lambda: {"total": 0, "reworked": 0})
    for i, c in enumerate(ordered):
        if not c["agent"] or not c["category"]:
            continue
        stats[c["category"]]["total"] += 1
        deadline = c["ts"] + WINDOW_HOURS * 3600
        touched_after, reworked = 0, False
        for later in ordered[i + 1:]:
            if later["ts"] > deadline:
                break
            if not (later["files"] & c["files"]):
                continue
            touched_after += 1
            if not later["agent"] and is_rework(later["body"]):
                reworked = True
                break
            if touched_after >= WINDOW_COMMITS:
                break
        if reworked:
            stats[c["category"]]["reworked"] += 1
    return stats
 
def wilson(reworked, total, z=1.96):
    """Wilson信頼区間の下限・上限。少数標本での過信を防ぐ。"""
    if total == 0:
        return (0.0, 0.0, 0.0)
    p = reworked / total
    denom = 1 + z * z / total
    center = (p + z * z / (2 * total)) / denom
    half = (z * math.sqrt(p * (1 - p) / total + z * z / (4 * total * total))) / denom
    return (p, max(0.0, center - half), min(1.0, center + half))
 
if __name__ == "__main__":
    stats = compute(load_commits())
    rows = []
    for cat, s in sorted(stats.items()):
        p, lo, hi = wilson(s["reworked"], s["total"])
        rows.append({"category": cat, "n": s["total"],
                     "rework_rate": round(p, 3),
                     "ci_low": round(lo, 3), "ci_high": round(hi, 3)})
    print(json.dumps(rows, ensure_ascii=False, indent=2))

[
  { "category": "boilerplate",  "n": 41, "rework_rate": 0.049, "ci_low": 0.013, "ci_high": 0.163 },
  { "category": "feature-impl", "n": 33, "rework_rate": 0.394, "ci_low": 0.245, "ci_high": 0.564 },
  { "category": "bugfix",       "n": 18, "rework_rate": 0.278, "ci_low": 0.126, "ci_high": 0.509 },
  { "category": "refactor",     "n": 12, "rework_rate": 0.500, "ci_low": 0.251, "ci_high": 0.749 }
]