"""レビュー→修正ループ実験。レビュー役と修正役は毎回新しい会話（履歴なし、ツールなし）。 使い方: python run_loop.py [--smoke] """ import difflib, json, os, re, shutil, subprocess, sys, tempfile, time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path HERE = Path(__file__).parent PY = os.environ.get("PYTHON", sys.executable) # pytest入りのpython MODELS = {"sonnet": "claude-sonnet-5-5", "haiku": "claude-haiku-4-5-20251001"} MODULES = ["invoice", "rate_limiter", "ranges", "access"] MAX_FIX = 6 REVIEW_OPEN = """You are a code reviewer. Review the Python module below carefully and list every problem you find. Order them by importance. If you find no problems, return an empty list. Return ONLY a JSON object in a ```json block, in this shape: {{"findings": [{{"title": "...", "severity": "high|medium|low", "detail": "..."}}]}} ```python {code} ```""" REVIEW_EVIDENCE = """You are a code reviewer. The module's docstring is its specification. Report ONLY defects where the code violates the specification, and for each one give a concrete input and the expected-vs-actual result. Do not report style, naming, performance, hardening, hypothetical concerns, or things the specification does not require. An empty list is a perfectly good answer. Return ONLY a JSON object in a ```json block, in this shape: {{"findings": [{{"title": "...", "severity": "high|medium|low", "detail": "input, expected, actual"}}]}} ```python {code} ```""" FIX = """Below is a Python module and a list of review findings. Fix the findings. Keep the public function and class names and signatures unchanged. Return the COMPLETE updated module in a single ```python block, with nothing else after it. Findings: {findings} ```python {code} ```""" REVIEWS = {"open": REVIEW_OPEN, "evidence": REVIEW_EVIDENCE} def call(model, prompt): with tempfile.TemporaryDirectory() as cwd: t = time.time() p = subprocess.run( ["claude", "-p", "--model", MODELS[model], "--tools", "", "--strict-mcp-config", "--setting-sources", "", "--no-session-persistence", "--disable-slash-commands", "--output-format", "json"], input=prompt, capture_output=True, text=True, cwd=cwd, timeout=600) d = json.loads(p.stdout) return d.get("result", ""), d.get("total_cost_usd", 0.0), time.time() - t def block(text, lang): m = re.findall(rf"```{lang}\s*\n(.*?)```", text, flags=re.S) return m[-1] if m else None def run_tests(module, code): with tempfile.TemporaryDirectory() as d: Path(d, f"{module}.py").write_text(code) shutil.copy(HERE / "tests" / f"test_{module}.py", d) p = subprocess.run([PY, "-m", "pytest", "-q", "-p", "no:cacheprovider", f"test_{module}.py"], cwd=d, capture_output=True, text=True, timeout=120) out = p.stdout passed = int((re.search(r"(\d+) passed", out) or [0, 0])[1]) failed = int((re.search(r"(\d+) failed", out) or [0, 0])[1]) errors = int((re.search(r"(\d+) error", out) or [0, 0])[1]) return passed, failed + errors def changed_lines(a, b): n = 0 for l in difflib.unified_diff(a.splitlines(), b.splitlines(), lineterm="", n=0): if (l.startswith("+") or l.startswith("-")) and not l.startswith(("+++", "---")): n += 1 return n def one_run(module, model, prompt_kind, rep): code = (HERE / "modules" / f"{module}.py").read_text() code0 = code rows = [] p, f = run_tests(module, code) base = dict(module=module, model=model, prompt=prompt_kind, rep=rep) for r in range(MAX_FIX + 1): text, cost_r, sec_r = call(model, REVIEWS[prompt_kind].format(code=code)) raw = block(text, "json") try: review = json.loads(raw) if not isinstance(review, dict) or not isinstance(review.get("findings"), list): raise ValueError("findings must be a list") findings = review["findings"] if any(not isinstance(x, dict) or any(not isinstance(x.get(k), str) for k in ("title", "severity", "detail")) for x in findings): raise ValueError("findings must contain title/severity/detail strings") except (ValueError, TypeError): findings = None row = dict(base, round=r, tests_passed=p, tests_failed=f, code_lines=len(code.splitlines()), # Historical field: net snapshot diff, not cumulative edits. cum_changed=changed_lines(code0, code), review_cost=cost_r, parse_ok=findings is not None, findings=findings or [], review_text=text if findings is None else "") if findings is None or not findings or r == MAX_FIX: row["stopped"] = "parse_error" if findings is None else ("no_findings" if not findings else "round_cap") rows.append(row) break ftxt = "\n".join(f"{i+1}. [{x.get('severity')}] {x.get('title')}: {x.get('detail')}" for i, x in enumerate(findings)) text2, cost_f, sec_f = call(model, FIX.format(findings=ftxt, code=code)) new = block(text2, "python") row["fix_cost"] = cost_f row["fix_ok"] = new is not None if new: row["step_changed"] = changed_lines(code, new) code = new p, f = run_tests(module, code) rows.append(row) row["code_after"] = code return rows def main(): out = Path(sys.argv[1]) smoke = "--smoke" in sys.argv reps = 1 if smoke else int(os.environ.get("REPS", 3)) jobs = [(m, mod, pk, rep) for mod in (MODULES[:1] if smoke else MODULES) for m in (["haiku"] if smoke else MODELS) for pk in REVIEWS for rep in range(reps)] done = set() if out.exists(): for l in out.read_text().splitlines(): r = json.loads(l) done.add((r["model"], r["module"], r["prompt"], r["rep"])) jobs = [j for j in jobs if j not in done] print(len(jobs), "runs", flush=True) with ThreadPoolExecutor(int(os.environ.get("WORKERS", 8))) as ex: futs = {ex.submit(one_run, mod, m, pk, rep): (m, mod, pk, rep) for m, mod, pk, rep in jobs} for fu in as_completed(futs): try: rows = fu.result() except Exception as e: print("FAIL", futs[fu], repr(e), flush=True) continue with out.open("a") as fh: for r in rows: fh.write(json.dumps(r, ensure_ascii=False) + "\n") print("done", futs[fu], len(rows), "rounds", flush=True) if __name__ == "__main__": main()