Edit Strategies & the Test-Driven Repair Loop

# Colab cell — pure Python, no key needed. Builds a tiny sandbox repo so
# the edit tool runs against a real file.
import pathlib, shutil

REPO = pathlib.Path("sandbox/repo").resolve()   # absolute so containment checks work
REPO.mkdir(parents=True, exist_ok=True)
(REPO / "pricing.py").write_text(
    "def calculate_discount(price, pct):\n"
    "    return price * (pct / 100)\n")

def apply_edit(rel: str, old: str, new: str) -> str:
    """Replace an exact block. Returns a clear error string on any mismatch."""
    target = (REPO / rel).resolve()
    if REPO not in target.parents:
        return "error: path escapes the repo"
    if not target.is_file():
        return f"error: {rel} does not exist"

    text = target.read_text()
    count = text.count(old)
    if count == 0:
        # The single most common failure: block drifted. Tell the model to re-read.
        return ("error: old block not found. Re-read the file and copy the "
                "exact current text (including indentation) before editing.")
    if count > 1:
        return (f"error: old block appears {count} times; include more "
                "surrounding context to make it unique.")

    backup = target.with_suffix(target.suffix + ".bak")
    shutil.copy(target, backup)              # cheap rollback point
    target.write_text(text.replace(old, new, 1))
    return f"applied: 1 replacement in {rel}"


# a valid edit, then the two recoverable failure branches:
print(apply_edit("pricing.py",
                 "    return price * (pct / 100)\n",
                 "    return min(price, price * (pct / 100))\n"))
print(apply_edit("pricing.py", "return refund(order)\n", "x"))   # not found
print(apply_edit("pricing.py", "\n", "x"))                       # many matches

# Colab cell — run the previous cell first (it defines REPO). Pure Python.
import hashlib

LAST_READ_HASH: dict[str, str] = {}   # populated by read_file on each read

def file_hash(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()

def apply_edit_guarded(rel: str, old: str, new: str) -> str:
    target = (REPO / rel).resolve()
    if REPO not in target.parents:
        return "error: path escapes the repo"
    if not target.is_file():
        return f"error: {rel} does not exist"

    text = target.read_text()
    expected = LAST_READ_HASH.get(rel)
    if expected is not None and file_hash(text) != expected:
        # The file changed since the model last read it — don't guess why.
        return ("error: file changed on disk since it was last read. "
                "Re-read " + rel + " before editing again.")

    count = text.count(old)
    if count == 0:
        return "error: old block not found. Re-read the file before editing."
    if count > 1:
        return f"error: old block appears {count} times; add more context."

    target.write_text(text.replace(old, new, 1))
    LAST_READ_HASH.pop(rel, None)   # invalidate — the next read repopulates it
    return f"applied: 1 replacement in {rel}"


# reset the file to a known state (the earlier cell's demo mutated it):
(REPO / "pricing.py").write_text(
    "def calculate_discount(price, pct):\n    return price * (pct / 100)\n")

# A) recorded hash doesn't match what's on disk -> rejected, sent to re-read:
LAST_READ_HASH["pricing.py"] = "stale-hash-that-wont-match"
print("A:", apply_edit_guarded("pricing.py",
      "    return price * (pct / 100)\n", "    return 0\n"))

# B) recorded hash matches current content -> applies cleanly:
LAST_READ_HASH["pricing.py"] = file_hash((REPO / "pricing.py").read_text())
print("B:", apply_edit_guarded("pricing.py",
      "    return price * (pct / 100)\n",
      "    return min(price, price * (pct / 100))\n"))

# Colab cell — run the first cell first (defines REPO). Installs pytest and
# drops a passing test into the sandbox so the runner has something to run.
!pip install -q pytest

import subprocess, sys, pathlib

REPO = pathlib.Path("sandbox/repo").resolve()
(REPO / "test_pricing.py").write_text(
    "from pricing import calculate_discount\n"
    "def test_discount():\n"
    "    assert calculate_discount(100, 10) == 10\n")

def run_tests(target: str = "") -> dict:
    """Run pytest in the sandbox; return pass/fail + trimmed output."""
    cmd = [sys.executable, "-m", "pytest", "-q", "--no-header"]   # this interpreter
    if target:
        cmd.append(target)
    proc = subprocess.run(
        cmd, cwd=REPO, capture_output=True, text=True, timeout=600,
    )
    output = proc.stdout + proc.stderr
    # Trim so a huge traceback dump doesn't blow the context budget.
    tail = "\n".join(output.splitlines()[-60:])
    return {
        "passed": proc.returncode == 0,
        "returncode": proc.returncode,
        "output_tail": tail,
    }

result = run_tests()
print("passed:", result["passed"])
print(result["output_tail"])

# Colab cell — run once. Set your key in the 🔑 panel (name it
# ANTHROPIC_API_KEY) or paste it. This is the assembled repair loop: it needs
# read_file/apply_edit/run_tests wired into REPAIR_TOOLS (schemas) and a
# run_repair_tool dispatcher, plus a sandboxed repo — read and adapt.
!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 anthropic

client = anthropic.Anthropic()
MAX_ATTEMPTS = 5

def repair(plan: dict, issue_text: str) -> dict:
    system = (
        "Fix the bug per the plan. FIRST write a test that reproduces the "
        "issue and fails (red). Then edit source with apply_edit until that "
        "test AND all existing tests pass (green). Read failing output before "
        "each edit. Tools: read_file, apply_edit, run_tests."
    )
    messages = [{"role": "user",
                 "content": f"Issue:\n{issue_text}\n\nPlan:\n{plan}"}]

    for attempt in range(1, MAX_ATTEMPTS + 1):
        resp = client.messages.create(
            model="claude-sonnet-5", max_tokens=4096,
            system=system, tools=REPAIR_TOOLS, messages=messages,
            output_config={"effort": "high"},   # hard repair steps earn depth
        )
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason != "tool_use":
            # stop_reason == "end_turn": the model THINKS it's done —
            # verify independently, never take its word.
            final = run_tests()
            if final["passed"]:
                return {"status": "success", "attempts": attempt}
            messages.append({"role": "user", "content":
                f"You stopped but tests still fail:\n{final['output_tail']}\n"
                "Keep fixing."})
            continue

        results = []
        for block in resp.content:
            if block.type == "tool_use":
                results.append(run_repair_tool(block))   # dispatches to the 3 tools
        messages.append({"role": "user", "content": results})

    # Bounded: give up cleanly rather than loop forever.
    return {"status": "exhausted", "attempts": MAX_ATTEMPTS,
            "last_tests": run_tests()}