PR Etiquette, HITL Gating & Evaluating a Coding Agent

# Reference implementation — needs 'pip install PyGithub', a GITHUB_TOKEN,
# a real repo/branch, and the audit() logger from Module 7. It opens a real
# draft PR, so it's here to read and adapt, not to run in Colab.
from github import Github        # PyGithub
import os

def open_pr_if_approved(repo_full: str, branch: str, base: str,
                        title: str, body: str, diff: str,
                        cost_usd: float) -> str:
    # 1) Surface everything the human needs to approve in seconds.
    print("=== PROPOSED PR ===")
    print(title)
    print(body)
    print(f"cost to produce: $" + f"{cost_usd:.3f}")
    print(diff[:4000])            # trimmed diff preview
    decision = input("open this draft PR? [y/N] ").strip().lower()

    audit({"event": "pr_decision", "repo": repo_full, "branch": branch,
           "approved": decision == "y", "cost_usd": cost_usd})
    if decision != "y":
        return "rejected by human — nothing pushed"

    # 2) Only on explicit yes do we touch GitHub, and only as a DRAFT.
    gh = Github(os.environ["GITHUB_TOKEN"])
    repo = gh.get_repo(repo_full)
    pr = repo.create_pull(title=title, body=body, head=branch,
                          base=base, draft=True)
    audit({"event": "pr_opened", "url": pr.html_url})
    return pr.html_url

# Colab cell — run once. Set your key in the 🔑 panel (name it
# ANTHROPIC_API_KEY) or just paste it when prompted.
!pip install -q anthropic

import os
try:
    from google.colab import userdata
    os.environ["ANTHROPIC_API_KEY"] = userdata.get("ANTHROPIC_API_KEY")
except Exception:
    from getpass import getpass
    os.environ.setdefault("ANTHROPIC_API_KEY", getpass("Anthropic API key: "))

import json

import anthropic

reviewer_client = anthropic.Anthropic()   # if the fixer is an OpenAI model,
                                          # this IS the cross-family reviewer

REVIEW_SCHEMA = {
    "type": "object",
    "properties": {
        "touches_tests": {"type": "boolean"},
        "scope_concern": {"type": "boolean"},
        "verdict": {"type": "string", "enum": ["pass", "flag"]},
    },
    "required": ["touches_tests", "scope_concern", "verdict"],
    "additionalProperties": False,
}

def independent_review(diff: str, issue_text: str) -> dict:
    """A DIFFERENT model/context reviews the diff before a human does.
    Never let the model that wrote the fix be the only reviewer of it."""
    resp = reviewer_client.messages.create(
        model="claude-sonnet-5", max_tokens=1024,
        system=(
            "You are reviewing a code change, not the engineer who wrote it. "
            "Check: does the diff plausibly fix the described issue? Does it "
            "touch any file under tests/ or named test_*/*_test? Is the diff "
            "larger than the issue seems to warrant?"
        ),
        messages=[{"role": "user", "content":
            f"Issue:\n{issue_text}\n\nDiff:\n{diff}"}],
        # Structured output: the verdict is machine-routed, so don't parse prose.
        output_config={"format": {"type": "json_schema", "schema": REVIEW_SCHEMA}},
    )
    text = next(b.text for b in resp.content if b.type == "text")
    return json.loads(text)


# a diff that weakens a test instead of fixing code -> should be flagged:
gamed_diff = (
    "--- a/tests/test_pricing.py\n+++ b/tests/test_pricing.py\n"
    "@@\n-    assert result <= 100\n+    assert result <= 200\n")
print(independent_review(gamed_diff,
                         "calculate_discount returns discounts over 100%"))

# Colab cell — pure Python, no key needed; run it as-is.
from dataclasses import dataclass, field

@dataclass
class IssueResult:
    issue_id: str
    outcome: str          # one of the taxonomy categories
    cost_usd: float
    seconds: float

def score_issue(run) -> str:
    if run.status == "exhausted":
        return "exhausted_retries"
    before, after = run.tests_before, run.tests_after
    if not run.wrote_reproducing_test:
        return "fix_without_test"
    if after.existing_failed:                 # broke something that passed
        return "regression_introduced"
    if not after.repro_test_passed:
        return "wrong_location"               # never actually fixed it
    return "full_success"

def report(results: list[IssueResult]) -> dict:
    n = len(results)
    from statistics import median
    succ = sum(r.outcome == "full_success" for r in results)
    by_outcome: dict[str, int] = {}
    for r in results:
        by_outcome[r.outcome] = by_outcome.get(r.outcome, 0) + 1
    return {
        "n_issues": n,
        "success_rate": round(succ / n, 3),
        "taxonomy": by_outcome,
        "median_cost_usd": round(median(r.cost_usd for r in results), 3),
        "median_seconds": round(median(r.seconds for r in results), 1),
    }


# fake runs, one per taxonomy category, to see the report populate:
from types import SimpleNamespace as NS
def make_run(status, wrote_test, existing_failed, repro_passed):
    return NS(status=status, wrote_reproducing_test=wrote_test,
              tests_before=NS(),
              tests_after=NS(existing_failed=existing_failed,
                             repro_test_passed=repro_passed))

runs = [
    make_run("done", True, False, True),         # full_success
    make_run("done", False, False, True),        # fix_without_test
    make_run("done", True, True, True),          # regression_introduced
    make_run("done", True, False, False),        # wrong_location
    make_run("exhausted", False, False, False),  # exhausted_retries
]
results = [IssueResult(f"issue-{i}", score_issue(r), cost_usd=0.05 * (i + 1),
                       seconds=10 * (i + 1)) for i, r in enumerate(runs)]
print(report(results))