Build a Policy-Aware Repair Controller

Introduction

30 Second Summary

A quick fix can make one problem disappear while quietly changing the rules that defined success. Without a trustworthy referee, a confident result can hide a more dangerous mistake.

In this project, you will build a terminal-based repair controller that safely evaluates deterministic fixes plus Claude Haiku 4.5 fixes for a broken HTTP retry policy. Each proposal must survive protected contract tests in a temporary sandbox before it can change your workspace.

What You'll Build

When you run the finished controller, you will see unsafe proposals rejected without touching trusted files before a policy-aware Claude repair earns acceptance with independent test evidence.

By the end of this project, you'll have:

• A visible failing contract that proves 408, 409, 429, and 5xx responses are missing retries while permanent-error safety still passes.
• A zero-cost deterministic run that displays one protected-file rejection plus one contract rejection while workspace/retry_policy.py remains unchanged.
• A policy-aware Claude repair whose audit.jsonl entry records the request ID, both model explanations, and separate passing contract evidence.
• Secret Mission: Detect a repeated patch hash and stop the controller early when another attempt cannot make progress.

Are there any prerequisites?

You'll need a Mac with internet access for the included Python setup. A Claude Console account with an API key is only needed for the paid live repair in Step 5.

Before We Start

Before any hands-on work, lock in what this controller repairs and the safeguards that make an accepted repair trustworthy. Every model output remains an untrusted proposal until protected contracts pass and the controller records its policy reasoning.

Set Up the Isolated Python Project

Every later safety check depends on the same Python behavior each time it runs. A virtual environment gives the controller a reproducible toolchain before any repair logic arrives.

You'll confirm Python 3.10 or newer in macOS Terminal. The environment pins pytest 9.1.1 plus the Anthropic Python SDK 1.3.0.

In this step, get ready to:

• Confirm that your Mac has a compatible Python version.
• Create the policy-aware-repair folder with a pinned dependency file.
• Activate an isolated environment with verified packages.

Check your Python version

The pinned packages require Python 3.10 or newer. This check determines whether your current interpreter can support them.

• Press Command-Space bar to open Spotlight.
• Type Terminal in the search field.
• Press Return to open Terminal.
• Check your installed Python version by running this command:

python3 --version

What does this command check?

The python3 command prints the version of the Python interpreter available in your current shell.

Your result falls into one of three states.

• Choose the tab that matches your Terminal output.

✔️ I see a supported version

Your interpreter meets the package requirement.

• Continue to the project setup below.

ⓧ I see an older version

Your current interpreter is below Python 3.10. Install the current macOS release before creating the environment.

• Open the official Python macOS downloads page.
• Select Python 3.14.7.
• Choose Download macOS installer.
• Open the downloaded installer.
• Complete the guided installation.
• Return to Terminal.
• Run the version check above again.

ⓧ Command not found

Your shell cannot find a Python interpreter. Install Python 3.14.7 before continuing.

• Open the official Python macOS downloads page.
• Select Python 3.14.7.
• Choose Download macOS installer.
• Open the downloaded installer.
• Complete the guided installation.
• Return to Terminal.
• Run the version check above again.

Still seeing the previous result?

Your existing Terminal session may still use the path it loaded before the installation.

• Quit Terminal.
• Reopen Terminal with the Spotlight steps above.
• Run the version check above again.

Help me diagnose why my Mac still cannot use the newly installed Python version.

Create the isolated project

The policy-aware-repair folder keeps every project artifact in one known location. Its .venv directory keeps the pinned packages separate from your system installation.

• Create the project folder on your Desktop and move into it by running these commands:

cd ~/Desktop
mkdir policy-aware-repair
ls
cd policy-aware-repair

What do these commands do?

• The first cd selects your Desktop as the starting location.
• The mkdir command creates the policy-aware-repair folder.
• The ls command lists the Desktop contents so you can confirm the folder exists.
• The final cd selects the new folder for the remaining commands.

You'll see policy-aware-repair in the listing before Terminal moves into that folder.

Folder command failed?

Check that each folder name uses the exact spelling policy-aware-repair.

Help me fix the Terminal commands that create my policy-aware-repair folder on the Desktop.

• Open a new requirements.txt file with the terminal editor by running this command:

nano requirements.txt

What does this command do?

The nano command opens requirements.txt for editing inside Terminal.

• Paste this exact dependency list into requirements.txt:

anthropic==1.3.0
pytest==9.1.1

Why pin these versions?

Each == pin tells pip to install one exact package release. Every learner gets the same controller dependencies.

• Press Ctrl+O to write requirements.txt to disk.
• Press Return to confirm the filename.

Nano confirms that the file was written.

• Press Ctrl+X to return to Terminal.
• Confirm that the dependency file exists by running this command:

ls

What does this command confirm?

The ls command shows the files saved inside policy-aware-repair.

You'll see requirements.txt in the output.

✔️ Awesome, I've got everything!

Your pinned dependency file is saved inside policy-aware-repair.

ⓧ I'd like to double check the full code

Compare your complete requirements.txt file with this reference:

anthropic==1.3.0
pytest==9.1.1

• Create the virtual environment and activate it by running these commands:

python3 -m venv .venv
source .venv/bin/activate

How does the environment work?

• The venv module creates an isolated Python installation inside .venv.
• The source command switches the current shell to that environment.

That's your clean boundary in place. Your Terminal prompt now begins with (.venv).

Missing the environment marker?

Check that Terminal is still inside the policy-aware-repair folder. Confirm that the activation command uses .venv/bin/activate.

Help me activate the .venv environment in my policy-aware-repair folder on macOS.

Install and verify dependencies

The active environment now has its own package location. Installing from requirements.txt reproduces the exact dependency versions required by the controller.

• Install the pinned dependencies by running this command:

python -m pip install -r requirements.txt

What does this command install?

The active Python interpreter runs pip against the package pins in requirements.txt.

Terminal reports that the packages installed successfully before returning to the prompt.

Dependency installation failed?

Confirm that your prompt begins with (.venv). Check that requirements.txt appears in the current folder.

Help me diagnose a failed requirements installation inside my policy-aware-repair virtual environment.

Before you run the final check, which three version results do you expect to see?

• Verify the interpreter and both installed packages by running this command:

python --version && python -m pytest --version && python -c "import anthropic; print(anthropic.__version__)"

What does the final check prove?

• The first python check confirms the interpreter selected by the active environment.
• The pytest check confirms the contract runner is available.
• The anthropic.__version__ expression confirms the SDK can be imported.

You'll see Python 3.10 or newer. You'll also see pytest 9.1.1 plus Anthropic SDK 1.3.0.

Your isolated toolchain is ready. The controller can now rely on a known interpreter plus reproducible package versions.

Your project environment is pinned and verified. Next up, you'll make the broken retry behavior visible with independent contracts.

Expose the Retry Policy Bug

Your isolated Python environment is ready. You can now give pytest a concrete retry rule to judge.

The starting policy returns False for every HTTP status. This exposes the missing transient-retry behavior before a repair controller can propose changes.

In this step, get ready to:

• Create a known-bad retry policy in the baseline folder and editable workspace.
• Define independent contracts for transient failures, permanent errors, and successful responses.
• Run the contract suite and isolate the intended failure.

Create the known-bad retry policy

The baseline preserves a known starting point for every repair demonstration. The workspace holds the copy that future repair proposals can replace.

• Return to the editor you used for requirements.txt earlier.
• Create a baseline folder inside policy-aware-repair using the editor's file sidebar.

You should now see baseline beside requirements.txt in the file sidebar.

• Create rushed_retry_policy.py inside the baseline folder using the file sidebar.
• Fill baseline/rushed_retry_policy.py by pasting this implementation:

"""Known-bad policy used to reset each repair demonstration."""


def should_retry(status_code: int) -> bool:
    """Return whether a failed HTTP request should be attempted again."""
    return False

What does this policy do?

• The module docstring identifies this file as the known-bad reset state.
• The should_retry() function receives one status code.
• The unconditional return False blocks every retry.

• Save baseline/rushed_retry_policy.py.
• Confirm the file ends with return False.

Does the baseline file look different?

Check that rushed_retry_policy.py sits inside baseline. Make sure the function body stays indented beneath should_retry().

Help me compare my baseline retry policy with the expected Python file.

✔️ Awesome, I've got everything!

Great. Double check that you saved baseline/rushed_retry_policy.py.

ⓧ I'd like to double check the full code

Compare your complete baseline/rushed_retry_policy.py file with this reference:

"""Known-bad policy used to reset each repair demonstration."""


def should_retry(status_code: int) -> bool:
    """Return whether a failed HTTP request should be attempted again."""
    return False

The editable workspace starts with the same broken behavior as the baseline. This gives the later controller a real target while preserving the reset copy.

• Create a workspace folder inside policy-aware-repair using the editor's file sidebar.

You should now see separate baseline and workspace folders in the file sidebar.

• Create retry_policy.py inside the workspace folder using the file sidebar.
• Fill workspace/retry_policy.py by pasting the same known-bad implementation:

"""Known-bad policy used to reset each repair demonstration."""


def should_retry(status_code: int) -> bool:
    """Return whether a failed HTTP request should be attempted again."""
    return False

Why keep two copies?

The baseline gives each demonstration a repeatable broken state. The workspace is the only implementation that a successful repair eventually replaces.

• Save workspace/retry_policy.py.
• Confirm the file defines should_retry(status_code: int) with an unconditional return False.

Is the workspace copy missing?

Check that the file path is exactly workspace/retry_policy.py. A file placed beside the workspace folder cannot be imported from the expected location.

Help me fix the location or contents of my workspace retry policy.

✔️ Awesome, I've got everything!

Your baseline and workspace now share the same deliberately broken policy.

ⓧ I'd like to double check the full code

Compare your complete workspace/retry_policy.py file with this reference:

"""Known-bad policy used to reset each repair demonstration."""


def should_retry(status_code: int) -> bool:
    """Return whether a failed HTTP request should be attempted again."""
    return False

Define the independent contract suite

A contract test captures behavior that every acceptable repair must preserve. Keeping the suite outside workspace prevents a candidate from rewriting its own success criteria.

The suite separates transient failures from permanent client errors. It also protects successful responses from accidental retries.

• Create a contracts folder inside policy-aware-repair using the editor's file sidebar.

You should see contracts beside workspace. This location keeps the tests outside the candidate-editable folder.

• Create test_retry_policy.py inside the contracts folder using the file sidebar.
• Add the transient-retry contract to contracts/test_retry_policy.py by pasting this first chunk:

from workspace.retry_policy import should_retry


def test_retries_transient_failures() -> None:
    transient_statuses = (408, 409, 429, 500, 503)
    assert all(should_retry(status) for status in transient_statuses)

What does the transient contract prove?

• The import loads should_retry() from the editable workspace.
• The transient_statuses tuple covers 408, 409, 429, 500, and 503.
• The all() assertion requires every listed transient status to trigger a retry.

• Save contracts/test_retry_policy.py.
• Confirm the file now contains test_retries_transient_failures().

Is the transient test indented incorrectly?

Keep transient_statuses and the assertion indented beneath the test function. Keep the import at the far left of the file.

Help me correct the transient contract's Python structure.

• Add the permanent-error contract below test_retries_transient_failures() by pasting this chunk:

def test_does_not_retry_permanent_errors() -> None:
    permanent_statuses = (400, 401, 403, 404, 422)
    assert all(not should_retry(status) for status in permanent_statuses)

What does the permanent-error contract protect?

• The permanent_statuses tuple covers 400, 401, 403, 404, and 422.
• The not should_retry(status) expression requires every permanent error to remain unretried.

• Save contracts/test_retry_policy.py.
• Confirm test_does_not_retry_permanent_errors() appears below the transient contract.

Does the permanent-error test look incomplete?

Check that not appears inside the all() expression. Without it, the test asks permanent errors to retry.

Help me verify the permanent-error contract.

• Add the successful-response contract below test_does_not_retry_permanent_errors() by pasting this final chunk:

def test_does_not_retry_successes() -> None:
    successful_statuses = (200, 201, 204)
    assert all(not should_retry(status) for status in successful_statuses)

Why test successful responses?

A repair could become too broad and retry every response. This contract requires 200, 201, and 204 to remain unretried.

• Save contracts/test_retry_policy.py.
• Confirm the file contains three separate test functions.

Is one contract missing from the file?

Check for one function covering transient statuses. Check for another covering permanent errors. Check for a third covering successful responses.

Help me compare my complete retry policy contract suite with the expected file.

✔️ Awesome, I've got everything!

Your independent suite now checks both sides of the retry policy plus successful responses.

ⓧ I'd like to double check the full code

Compare your complete contracts/test_retry_policy.py file with this reference:

from workspace.retry_policy import should_retry


def test_retries_transient_failures() -> None:
    transient_statuses = (408, 409, 429, 500, 503)
    assert all(should_retry(status) for status in transient_statuses)


def test_does_not_retry_permanent_errors() -> None:
    permanent_statuses = (400, 401, 403, 404, 422)
    assert all(not should_retry(status) for status in permanent_statuses)


def test_does_not_retry_successes() -> None:
    successful_statuses = (200, 201, 204)
    assert all(not should_retry(status) for status in successful_statuses)

Run the full contract suite

The suite now asks three separate policy questions in one run. A prediction gives you a hypothesis before the test result reveals the exact gap.

Before you run the suite, predict which contract the always-false function will violate.

• Run the independent contract suite in the activated Terminal by using this command:

python -m pytest -q contracts/test_retry_policy.py

How do you read this result?

• The test_retries_transient_failures contract fails because every call returns False.
• The test_does_not_retry_permanent_errors contract passes because permanent errors remain unretried.
• The test_does_not_retry_successes contract passes because successful responses remain unretried.
• One failing contract plus two passing contracts isolates the missing transient-retry behavior.

• Read the failed assertion under test_retries_transient_failures.
• Trace the listed transient statuses through the unconditional return False.
• Confirm the terminal summary reports one failed test plus two passed tests.

Seeing an import or collection problem?

Confirm that the activated Terminal is still inside policy-aware-repair. Check that the implementation sits at workspace/retry_policy.py.

Help me diagnose why pytest cannot run my retry policy contracts.

You've pinned down the bug: transient retries fail while permanent-error safety stays intact. Next, you'll build the sandbox that tests proposed repairs without risking workspace/retry_policy.py.

Build the Sandboxed Contract Runner

Your independent contracts now expose the missing transient retries. Candidate code still needs containment before it can be trusted with your project files.

A temporary sandbox gives each proposal a disposable workspace. pytest evaluates the copied candidate against the trusted contract inside that boundary.

In this step, get ready to:

• Define the controller's editable boundary.
• Run candidate code inside a temporary sandbox.
• Expose separate results for both policy duties.

Define the repair boundary

An allowlist names the files a proposal may change. Protected paths keep the policy file plus trusted contracts outside that boundary.

• Create `.repair-policy.json` inside the `policy-aware-repair` folder by copying this configuration:

{
  "editable_paths": [
    "workspace/retry_policy.py"
  ],
  "protected_paths": [
    ".repair-policy.json",
    "contracts/test_retry_policy.py",
    "repair_controller.py",
    "requirements.txt"
  ],
  "max_attempts": 2
}

What does this policy control?

• `editable_paths` limits accepted proposals to `workspace/retry_policy.py`.
• `protected_paths` identifies files that proposals must never replace.
• `max_attempts` caps one controller run at two proposals.

• Save `.repair-policy.json`.
• Confirm that `.repair-policy.json` shows one editable path plus four protected paths in your editor.

Seeing an invalid policy file?

Check the commas after each list item. Confirm that every property name uses double quotes.

Ask for help checking the policy structure without changing its paths: Help me find the JSON syntax problem in my repair policy.

The controller needs standard-library tools for paths plus temporary directories. It also needs immutable records for proposals plus test results.

• Create `repair_controller.py` inside the `policy-aware-repair` folder by adding these imports:

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Protocol

What do these imports provide?

• `tempfile` creates the disposable proving ground.
• `subprocess` launches each independent contract check.
• `Path` handles files in the real project plus the sandbox.
• `dataclass` creates fixed-shape proposal plus result records.

• Save `repair_controller.py`.
• Confirm that your editor reports no import syntax errors.

Seeing an import syntax error?

Check that every import occupies its own line. Keep `Path` plus `PurePosixPath` on the same import line.

Ask for help comparing the import group: Help me check these controller imports.

• Add the project paths plus contract labels below `from typing import Protocol`:

ROOT = Path(__file__).resolve().parent
POLICY_PATH = ROOT / ".repair-policy.json"
BASELINE_PATH = ROOT / "baseline/rushed_retry_policy.py"
TARGET_RELATIVE_PATH = "workspace/retry_policy.py"
TARGET_PATH = ROOT / TARGET_RELATIVE_PATH
CONTRACT_PATH = ROOT / "contracts/test_retry_policy.py"
AUDIT_PATH = ROOT / "audit.jsonl"

CONTRACT_NODES = {
    "transient_retry": (
        "contracts/test_retry_policy.py::test_retries_transient_failures"
    ),
    "permanent_error_safety": (
        "contracts/test_retry_policy.py::test_does_not_retry_permanent_errors"
    ),
    "full_suite": "contracts/test_retry_policy.py",
}

What do these paths represent?

• `ROOT` anchors every path to the `policy-aware-repair` folder.
• `TARGET_RELATIVE_PATH` identifies the only editable implementation.
• `CONTRACT_NODES` separates transient behavior from permanent-error safety.
• `full_suite` checks every contract together before acceptance.

• Save `repair_controller.py`.
• Confirm that each path matches an existing artifact from the previous step.

Does a path look unresolved?

Compare the spelling with the files in your editor sidebar. Pay close attention to `contracts/test_retry_policy.py` plus `workspace/retry_policy.py`.

Ask for help checking the path constants: Help me compare the controller paths with my project layout.

• Confirm that the trusted suite still reports its original failure by running:

python -m pytest -q contracts/test_retry_policy.py

You should still see only `test_retries_transient_failures` fail. This confirms that the new policy plus controller foundations have not changed the trusted contract behavior.

Seeing a different contract result?

Confirm that `workspace/retry_policy.py` still returns `False`. Check that `contracts/test_retry_policy.py` still contains all three original tests.

Ask for help isolating the changed artifact: Help me find why my original contract result changed.

• Add the controller records plus provider interface below `CONTRACT_NODES`:

@dataclass(frozen=True)
class Proposal:
    files: dict[str, str]
    transient_retry_reason: str
    permanent_error_safety_reason: str
    request_id: str | None = None


@dataclass(frozen=True)
class CheckResult:
    passed: bool
    output: str


class Provider(Protocol):
    name: str

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        ...

What do these types enforce?

• `Proposal` carries proposed file contents plus separate policy explanations.
• `CheckResult` keeps a pass decision beside captured test output.
• `Provider` defines the interface that later repair candidates follow.
• `frozen=True` prevents proposal records from changing during evaluation.

• Save `repair_controller.py`.
• Confirm that `Proposal`, `CheckResult`, plus `Provider` appear as top-level definitions.

Seeing an indentation warning?

Keep each class at the left edge. Indent the `propose()` parameters beneath the method definition.

Ask for help checking the type definitions: Help me fix the controller type indentation.

• Add policy loading plus path validation below `Provider`:

def load_policy() -> dict:
    return json.loads(POLICY_PATH.read_text(encoding="utf-8"))


def reset_workspace() -> None:
    TARGET_PATH.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(BASELINE_PATH, TARGET_PATH)


def validate_paths(files: dict[str, str], policy: dict) -> str | None:
    editable = set(policy["editable_paths"])
    protected = set(policy["protected_paths"])

    for name in files:
        path = PurePosixPath(name)
        if path.is_absolute() or ".." in path.parts:
            return f"unsafe_path: {name}"
        if name in protected:
            return f"protected_file: {name}"
        if name not in editable:
            return f"outside_editable_allowlist: {name}"

    if TARGET_RELATIVE_PATH not in files:
        return f"missing_required_target: {TARGET_RELATIVE_PATH}"
    return None

How does path validation protect the project?

• `load_policy()` reads the trusted boundary from `.repair-policy.json`.
• `PurePosixPath` exposes unsafe absolute paths plus parent-directory traversal.
• `validate_paths()` rejects protected files before candidate code runs.
• The final check requires every proposal to include `workspace/retry_policy.py`.

• Save `repair_controller.py`.
• Confirm that `validate_paths()` returns `None` only after every path check passes.

Seeing an undefined path name?

Check that the path constants sit above these functions. Confirm that `PurePosixPath` appears in the import group.

Ask for help tracing the missing name: Help me find the undefined controller path.

• Run the trusted contract suite again to confirm that the editable implementation remains untouched:

python -m pytest -q contracts/test_retry_policy.py

You should see the same single transient-retry failure. The path policy exists without altering candidate code or the independent contracts.

Did the contract result change?

Inspect `reset_workspace()` for accidental execution outside a function. Its two body lines must remain indented beneath the definition.

Ask for help checking the function boundary: Help me verify that reset_workspace is only defined.

Build the disposable contract runner

Each contract runs as a separate subprocess inside the temporary directory. Captured output becomes evidence that the controller can record later.

• Add the single-contract subprocess function below `validate_paths()`:

def run_one_contract(sandbox: Path, node_id: str) -> CheckResult:
    completed = subprocess.run(
        [sys.executable, "-m", "pytest", "-q", node_id],
        cwd=sandbox,
        capture_output=True,
        text=True,
        check=False,
    )
    output = (completed.stdout + completed.stderr).strip()
    return CheckResult(passed=completed.returncode == 0, output=output)

What does the subprocess capture?

• `sys.executable` uses the Python interpreter from the activated virtual environment.
• `cwd=sandbox` keeps test execution inside the disposable directory.
• `capture_output=True` collects the result without mixing it into controller output.
• The subprocess return code becomes the contract's pass decision.

• Save `repair_controller.py`.
• Confirm that `run_one_contract()` returns one `CheckResult`.

Seeing a subprocess syntax error?

Check the closing parenthesis after `check=False`. Keep the command arguments inside one Python list.

Ask for help comparing the function: Help me check run_one_contract.

• Add the fresh-sandbox function below `run_one_contract()`:

def run_contracts(candidate_source: str) -> dict[str, CheckResult]:
    with tempfile.TemporaryDirectory(prefix="policy-repair-") as temporary:
        sandbox = Path(temporary)
        workspace = sandbox / "workspace"
        contracts = sandbox / "contracts"
        workspace.mkdir()
        contracts.mkdir()
        (workspace / "__init__.py").write_text("", encoding="utf-8")
        (workspace / "retry_policy.py").write_text(
            candidate_source, encoding="utf-8"
        )
        shutil.copy2(CONTRACT_PATH, contracts / "test_retry_policy.py")

        return {
            label: run_one_contract(sandbox, node_id)
            for label, node_id in CONTRACT_NODES.items()
        }

How is the sandbox assembled?

• `TemporaryDirectory` creates a fresh directory for one candidate evaluation.
• The candidate source becomes the sandbox copy of `workspace/retry_policy.py`.
• The contract is copied from the trusted project path.
• Leaving the context removes the temporary directory plus its contents.

• Save `repair_controller.py`.
• Confirm that `run_contracts()` copies `CONTRACT_PATH` instead of reading proposed test content.

Does the sandbox miss the workspace package?

Confirm that the function creates `workspace/__init__.py`. Check that `workspace.mkdir()` runs before the file writes.

Ask for help checking the sandbox tree: Help me inspect the temporary sandbox setup.

• Run the original contract suite once more to confirm that sandbox support has not changed the real workspace:

python -m pytest -q contracts/test_retry_policy.py

You should still see the transient contract fail plus the other two contracts pass. The new runner functions are present without executing candidate code in the real workspace.

Seeing new failures in the real suite?

Check that every sandbox file write begins with `workspace /` or `contracts /`. Confirm that no new call to `run_contracts()` sits at the top level.

Ask for help tracing the unexpected execution: Help me find why sandbox code changed the real suite.

Verify sandboxed results and preserve utilities

The raw test output needs a JSON-friendly shape before the command line can print it. This creates a readable snapshot for each policy duty.

• Add result serialization below `run_contracts()`:

def serialize_checks(checks: dict[str, CheckResult]) -> dict:
    return {
        label: {"passed": result.passed, "output": result.output[-3000:]}
        for label, result in checks.items()
    }

Why serialize the checks?

`serialize_checks()` converts each `CheckResult` into plain JSON-compatible values. The output limit keeps future audit records bounded.

• Save `repair_controller.py`.
• Confirm that each serialized label contains `passed` plus `output`.

Seeing a dictionary syntax error?

Check the braces around the nested result dictionary. Keep the comprehension beneath the opening return brace.

Ask for help checking the comprehension: Help me fix serialize_checks.

• Add the check command plus command-line entry point below `serialize_checks()`:

def print_current_checks() -> None:
    checks = run_contracts(TARGET_PATH.read_text(encoding="utf-8"))
    print(json.dumps(serialize_checks(checks), indent=2))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Safely evaluate deterministic or Claude-generated repairs."
    )
    parser.add_argument("--reset", action="store_true")
    parser.add_argument("--check-current", action="store_true")
    args = parser.parse_args()

    if args.reset:
        reset_workspace()
    if args.check_current:
        print_current_checks()
        return


if __name__ == "__main__":
    main()

What does the check command prove?

• `print_current_checks()` reads the real candidate source without editing it.
• `run_contracts()` evaluates that source inside a fresh sandbox.
• `json.dumps()` prints separate evidence for each contract label.
• `--reset` can restore the known-bad baseline for later demonstrations.

• Save `repair_controller.py`.
• Confirm that `main()` sits above the final `if __name__` block.

Does the check flag do nothing?

Confirm that `main()` is indented beneath the final condition. Check that `print_current_checks()` appears inside the `args.check_current` branch.

Ask for help tracing the flag: Help me debug the check-current entry point.

Before you run the controller, which policy result do you expect to differ from the other focused check?

• Evaluate the current implementation inside the sandbox by running:

python repair_controller.py --check-current

You should see `transient_retry` set to `false`. You should see `permanent_error_safety` set to `true` plus `full_suite` set to `false`.

Seeing a file or import error?

Confirm that the activated terminal is inside `policy-aware-repair`. Check that `contracts/test_retry_policy.py` plus `workspace/retry_policy.py` still exist.

Ask for help using the captured error: Help me debug my sandbox check.

The output is ready for later audit records. Two small utilities complete the controller foundation without changing the current check.

• Insert JSONL audit writing between `serialize_checks()` plus `print_current_checks()`:

def append_audit(event: dict) -> None:
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **event,
    }
    with AUDIT_PATH.open("a", encoding="utf-8") as audit_file:
        audit_file.write(json.dumps(record, sort_keys=True) + "\n")

How will audit records be stored?

`append_audit()` adds a UTC timestamp to each event. It writes one sorted JSON object per line to `audit.jsonl`.

• Save `repair_controller.py`.
• Confirm that the write ends with `\n` so each future event occupies one line.

Seeing a string syntax error?

Check the quotes around `utf-8`. Confirm that the newline remains inside its own quoted string.

Ask for help checking the audit function: Help me fix append_audit.

• Add atomic accepted-file replacement below `append_audit()`:

def apply_files(files: dict[str, str]) -> None:
    for relative_name, content in files.items():
        destination = ROOT / relative_name
        temporary = destination.with_suffix(destination.suffix + ".accepted")
        temporary.write_text(content, encoding="utf-8")
        temporary.replace(destination)

Why use a temporary accepted file?

`apply_files()` writes accepted content beside its destination first. `replace()` performs the final swap after the full content is ready.

• Save `repair_controller.py`.
• Confirm that `temporary.replace(destination)` remains inside the loop.

Seeing an undefined destination?

Check that all four lines beneath the loop use one indentation level. Confirm that `ROOT` remains defined near the top of the file.

Ask for help checking the atomic replacement: Help me fix apply_files.

Before the final check, do you expect the two new utility functions to change the sandboxed policy results?

• Run the complete current-state check again:

python repair_controller.py --check-current

You should again see `transient_retry` as `false`. You should see `permanent_error_safety` as `true` plus `full_suite` as `false`.

• Return to `workspace/retry_policy.py` in your editor.
• Confirm that `should_retry()` still ends with `return False`.

That is the key safety result: the controller evaluated the broken source in a disposable sandbox while leaving the real workspace unchanged.

Did the workspace file change?

Restore `workspace/retry_policy.py` from `baseline/rushed_retry_policy.py`. Check that no call to `apply_files()` exists outside a function.

Ask for help tracing the write: Help me find what changed my real workspace.

✔️ Awesome, I've got everything!

Great. Your controller now reports independent sandbox results while `workspace/retry_policy.py` remains the known-bad implementation.

ⓧ I'd like to double check the full code

Compare your complete `repair_controller.py` with this cumulative version:

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Protocol

ROOT = Path(__file__).resolve().parent
POLICY_PATH = ROOT / ".repair-policy.json"
BASELINE_PATH = ROOT / "baseline/rushed_retry_policy.py"
TARGET_RELATIVE_PATH = "workspace/retry_policy.py"
TARGET_PATH = ROOT / TARGET_RELATIVE_PATH
CONTRACT_PATH = ROOT / "contracts/test_retry_policy.py"
AUDIT_PATH = ROOT / "audit.jsonl"

CONTRACT_NODES = {
    "transient_retry": (
        "contracts/test_retry_policy.py::test_retries_transient_failures"
    ),
    "permanent_error_safety": (
        "contracts/test_retry_policy.py::test_does_not_retry_permanent_errors"
    ),
    "full_suite": "contracts/test_retry_policy.py",
}


@dataclass(frozen=True)
class Proposal:
    files: dict[str, str]
    transient_retry_reason: str
    permanent_error_safety_reason: str
    request_id: str | None = None


@dataclass(frozen=True)
class CheckResult:
    passed: bool
    output: str


class Provider(Protocol):
    name: str

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        ...


def load_policy() -> dict:
    return json.loads(POLICY_PATH.read_text(encoding="utf-8"))


def reset_workspace() -> None:
    TARGET_PATH.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(BASELINE_PATH, TARGET_PATH)


def validate_paths(files: dict[str, str], policy: dict) -> str | None:
    editable = set(policy["editable_paths"])
    protected = set(policy["protected_paths"])

    for name in files:
        path = PurePosixPath(name)
        if path.is_absolute() or ".." in path.parts:
            return f"unsafe_path: {name}"
        if name in protected:
            return f"protected_file: {name}"
        if name not in editable:
            return f"outside_editable_allowlist: {name}"

    if TARGET_RELATIVE_PATH not in files:
        return f"missing_required_target: {TARGET_RELATIVE_PATH}"
    return None


def run_one_contract(sandbox: Path, node_id: str) -> CheckResult:
    completed = subprocess.run(
        [sys.executable, "-m", "pytest", "-q", node_id],
        cwd=sandbox,
        capture_output=True,
        text=True,
        check=False,
    )
    output = (completed.stdout + completed.stderr).strip()
    return CheckResult(passed=completed.returncode == 0, output=output)


def run_contracts(candidate_source: str) -> dict[str, CheckResult]:
    with tempfile.TemporaryDirectory(prefix="policy-repair-") as temporary:
        sandbox = Path(temporary)
        workspace = sandbox / "workspace"
        contracts = sandbox / "contracts"
        workspace.mkdir()
        contracts.mkdir()
        (workspace / "__init__.py").write_text("", encoding="utf-8")
        (workspace / "retry_policy.py").write_text(
            candidate_source, encoding="utf-8"
        )
        shutil.copy2(CONTRACT_PATH, contracts / "test_retry_policy.py")

        return {
            label: run_one_contract(sandbox, node_id)
            for label, node_id in CONTRACT_NODES.items()
        }


def serialize_checks(checks: dict[str, CheckResult]) -> dict:
    return {
        label: {"passed": result.passed, "output": result.output[-3000:]}
        for label, result in checks.items()
    }


def append_audit(event: dict) -> None:
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **event,
    }
    with AUDIT_PATH.open("a", encoding="utf-8") as audit_file:
        audit_file.write(json.dumps(record, sort_keys=True) + "\n")


def apply_files(files: dict[str, str]) -> None:
    for relative_name, content in files.items():
        destination = ROOT / relative_name
        temporary = destination.with_suffix(destination.suffix + ".accepted")
        temporary.write_text(content, encoding="utf-8")
        temporary.replace(destination)


def print_current_checks() -> None:
    checks = run_contracts(TARGET_PATH.read_text(encoding="utf-8"))
    print(json.dumps(serialize_checks(checks), indent=2))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Safely evaluate deterministic or Claude-generated repairs."
    )
    parser.add_argument("--reset", action="store_true")
    parser.add_argument("--check-current", action="store_true")
    args = parser.parse_args()

    if args.reset:
        reset_workspace()
    if args.check_current:
        print_current_checks()
        return


if __name__ == "__main__":
    main()

Your controller now has a safe proving ground plus a machine-readable result. Next, you will send rushed proposals through these guardrails and record why each one is rejected.

Reject the Rushed Candidate Safely

Your sandbox now evaluates candidate code without touching the real workspace. That isolation gives each proposal a safe proving ground.

A rushed fixer can still try to rewrite a protected contract. It can also submit an allowed patch that retries permanent errors. This step makes the controller reject both risks before any change reaches workspace/retry_policy.py.

In this step, get ready to:

• Add a deterministic provider that submits two deliberately unsafe proposals.
• Build a bounded controller loop that validates paths before sandbox execution.
• Inspect the audit trail after both proposals are rejected.

Add the deterministic rushed provider

A deterministic test double produces the same proposals every time. It lets you prove the guardrails work without making a paid model request.

• In repair_controller.py, place the following class immediately below the existing Provider protocol:

class DeterministicRushedProvider:
    name = "deterministic-rushed"

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        del source, contract, feedback
        unsafe_source = '''"""Unsafe rushed retry policy."""


def should_retry(status_code: int) -> bool:
    return status_code >= 400
'''
        return Proposal(
            files={TARGET_RELATIVE_PATH: unsafe_source},
            transient_retry_reason="Every error status retries, including transient ones.",
            permanent_error_safety_reason=(
                "The implementation is short, so the permanent-error risk was ignored."
            ),
        )

What does this provider propose?

• The unsafe_source implementation retries every status code of 400 or higher.
• The returned Proposal changes only the editable target.
• The permanent-error explanation admits that the proposal ignores the safety risk.

• Inside propose(), insert the following branch immediately before the existing return Proposal( line:

        if attempt == 1:
            return Proposal(
                files={
                    TARGET_RELATIVE_PATH: unsafe_source,
                    "contracts/test_retry_policy.py": (
                        "def test_contracts_were_weakened():\n    assert True\n"
                    ),
                },
                transient_retry_reason="All failed requests are retried.",
                permanent_error_safety_reason=(
                    "The tests were changed so permanent errors no longer block acceptance."
                ),
            )

Why are there two proposals?

Attempt one includes the protected contracts/test_retry_policy.py file. Attempt two stays inside the editable allowlist.

The second proposal still violates the semantic contract because it retries permanent client errors. The two attempts exercise different layers of the controller.

• Save repair_controller.py.
• Confirm the existing sandbox check still runs by using:

python repair_controller.py --check-current

What should this check prove?

You should still see transient_retry and full_suite reported as false. You should see permanent_error_safety reported as true.

The new provider has only been defined. The real workspace remains unchanged.

Seeing a Python syntax error?

Check that the first-attempt branch sits inside propose() with eight spaces of indentation. Make sure the existing second Proposal remains below that branch.

Ask for help with the class structure: Help me compare my DeterministicRushedProvider with the required class structure.

Build the bounded controller loop

The controller needs a strict order of operations. It asks for a proposal first. It validates file paths before running any candidate code in the sandbox.

• In repair_controller.py, add the start of run_controller() immediately above print_current_checks() by copying this code:

def run_controller(provider: Provider) -> bool:
    policy = load_policy()
    max_attempts = int(policy["max_attempts"])
    source = TARGET_PATH.read_text(encoding="utf-8")
    contract = CONTRACT_PATH.read_text(encoding="utf-8")
    feedback = "No previous attempt has been evaluated."

    for attempt in range(1, max_attempts + 1):
        try:
            proposal = provider.propose(source, contract, feedback, attempt)
        except Exception as error:
            feedback = f"provider_error: {type(error).__name__}: {error}"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "reason": feedback,
                }
            )
            continue

How does the attempt bound work?

• The policy supplies max_attempts, so the provider cannot choose how long the loop runs.
• The controller keeps the original source and immutable contract available for every proposal.
• A provider exception becomes an attempt_rejected audit event.

• Continue run_controller() by adding this path-validation block after the provider error handler:

        path_error = validate_paths(proposal.files, policy)
        if path_error is not None:
            feedback = path_error
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": path_error,
                    "proposed_paths": sorted(proposal.files),
                }
            )
            continue

Why validate paths first?

The path check runs before run_contracts(). A proposal that targets a protected file is rejected before its candidate source reaches the sandbox.

The audit record preserves the rejected paths. This makes the trust-boundary decision reviewable.

• Save repair_controller.py.
• Check that the controller still loads by running:

python repair_controller.py --check-current

What confirms the file still loads?

You should receive the same three JSON check results from the current broken policy. A syntax error would stop the file before those results were printed.

• Add the policy-reasoning check after the path-validation block:

        if not proposal.transient_retry_reason.strip() or not (
            proposal.permanent_error_safety_reason.strip()
        ):
            feedback = "missing_policy_reasoning"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": feedback,
                }
            )
            continue

Why require policy reasoning?

A passing patch must explain both policy duties. Empty reasoning causes rejection before acceptance can be considered.

This evidence requirement supports later audit review. It does not replace the independent tests.

• Add the sandbox test and contract-rejection block immediately below the reasoning check:

        checks = run_contracts(proposal.files[TARGET_RELATIVE_PATH])
        check_log = serialize_checks(checks)
        if not all(result.passed for result in checks.values()):
            feedback = "contract_failure\n" + "\n".join(
                f"[{label}]\n{result.output}"
                for label, result in checks.items()
                if not result.passed
            )
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": "contract_failure",
                    "checks": check_log,
                }
            )
            continue

How are semantic failures rejected?

The candidate source runs against every entry in CONTRACT_NODES. Any failed result records contract_failure with the serialized test evidence.

The continue statement keeps the failed candidate away from apply_files().

• Save repair_controller.py.
• Verify the current workspace still passes through the sandbox checker by running:

python repair_controller.py --check-current

What remains unchanged?

You should still see the broken policy results. Defining the rejection loop does not apply a candidate.

• Add the accepted-candidate branch immediately after the contract-rejection block:

        apply_files(proposal.files)
        append_audit(
            {
                "event": "accepted",
                "provider": provider.name,
                "attempt": attempt,
                "request_id": proposal.request_id,
                "acceptance_reason": (
                    "Accepted only after the immutable contract suite passed and "
                    "the proposal supplied explicit reasoning for both policy duties."
                ),
                "checks": check_log,
            }
        )
        print("Accepted repair. Review the final audit record for policy evidence.")
        return True

When can files be applied?

The call to apply_files() appears after path validation, reasoning validation, and all contract checks. Only a fully accepted proposal reaches the atomic replacement function.

The deterministic proposals never reach this branch. Their failures remain contained.

• Inside the accepted audit dictionary, insert the following policy_evidence entry immediately before "checks": check_log:

                "policy_evidence": {
                    "transient_retry": {
                        "requirement": "Retry 408, 409, 429, and 5xx responses.",
                        "candidate_reason": proposal.transient_retry_reason,
                        "contract": CONTRACT_NODES["transient_retry"],
                        "result": "passed",
                    },
                    "permanent_error_safety": {
                        "requirement": (
                            "Do not retry permanent 400, 401, 403, 404, or 422 errors."
                        ),
                        "candidate_reason": proposal.permanent_error_safety_reason,
                        "contract": CONTRACT_NODES["permanent_error_safety"],
                        "result": "passed",
                    },
                },

What does acceptance evidence contain?

Each policy duty records its requirement. It also records the candidate's reason and the independent contract that passed.

This structure keeps transient retry behavior separate from permanent-error safety. An accepted record explains both sides of the policy.

• Save repair_controller.py.
• Confirm the current policy remains untouched by running:

python repair_controller.py --check-current

What should the JSON still show?

You should see false for transient_retry and full_suite. You should see true for permanent_error_safety.

Run both rejected attempts

The loop still needs a final record for the case where every permitted attempt fails. This outcome distinguishes bounded rejection from an accepted repair.

• Complete run_controller() by adding this block after the attempt loop:

    append_audit(
        {
            "event": "attempts_exhausted",
            "provider": provider.name,
            "max_attempts": max_attempts,
        }
    )
    print(f"No repair accepted after {max_attempts} attempts.")
    return False

Why record exhaustion separately?

Each failed proposal already has its own rejection event. The final attempts_exhausted event records that the policy limit ended the run.

The controller returns false because no candidate earned acceptance. The CLI converts that result into a failing exit status.

• Save repair_controller.py.

✔️ Awesome, I've got everything!

Your controller now defines the deterministic provider and bounded repair loop. Make sure repair_controller.py is saved before running the designed rejection.

ⓧ I'd like to double check the full code

Compare your cumulative repair_controller.py with this version:

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Protocol

ROOT = Path(__file__).resolve().parent
POLICY_PATH = ROOT / ".repair-policy.json"
BASELINE_PATH = ROOT / "baseline/rushed_retry_policy.py"
TARGET_RELATIVE_PATH = "workspace/retry_policy.py"
TARGET_PATH = ROOT / TARGET_RELATIVE_PATH
CONTRACT_PATH = ROOT / "contracts/test_retry_policy.py"
AUDIT_PATH = ROOT / "audit.jsonl"

CONTRACT_NODES = {
    "transient_retry": (
        "contracts/test_retry_policy.py::test_retries_transient_failures"
    ),
    "permanent_error_safety": (
        "contracts/test_retry_policy.py::test_does_not_retry_permanent_errors"
    ),
    "full_suite": "contracts/test_retry_policy.py",
}


@dataclass(frozen=True)
class Proposal:
    files: dict[str, str]
    transient_retry_reason: str
    permanent_error_safety_reason: str
    request_id: str | None = None


@dataclass(frozen=True)
class CheckResult:
    passed: bool
    output: str


class Provider(Protocol):
    name: str

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        ...


class DeterministicRushedProvider:
    name = "deterministic-rushed"

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        del source, contract, feedback
        unsafe_source = '''"""Unsafe rushed retry policy."""


def should_retry(status_code: int) -> bool:
    return status_code >= 400
'''
        if attempt == 1:
            return Proposal(
                files={
                    TARGET_RELATIVE_PATH: unsafe_source,
                    "contracts/test_retry_policy.py": (
                        "def test_contracts_were_weakened():\n    assert True\n"
                    ),
                },
                transient_retry_reason="All failed requests are retried.",
                permanent_error_safety_reason=(
                    "The tests were changed so permanent errors no longer block acceptance."
                ),
            )
        return Proposal(
            files={TARGET_RELATIVE_PATH: unsafe_source},
            transient_retry_reason="Every error status retries, including transient ones.",
            permanent_error_safety_reason=(
                "The implementation is short, so the permanent-error risk was ignored."
            ),
        )


def load_policy() -> dict:
    return json.loads(POLICY_PATH.read_text(encoding="utf-8"))


def reset_workspace() -> None:
    TARGET_PATH.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(BASELINE_PATH, TARGET_PATH)


def validate_paths(files: dict[str, str], policy: dict) -> str | None:
    editable = set(policy["editable_paths"])
    protected = set(policy["protected_paths"])

    for name in files:
        path = PurePosixPath(name)
        if path.is_absolute() or ".." in path.parts:
            return f"unsafe_path: {name}"
        if name in protected:
            return f"protected_file: {name}"
        if name not in editable:
            return f"outside_editable_allowlist: {name}"

    if TARGET_RELATIVE_PATH not in files:
        return f"missing_required_target: {TARGET_RELATIVE_PATH}"
    return None


def run_one_contract(sandbox: Path, node_id: str) -> CheckResult:
    completed = subprocess.run(
        [sys.executable, "-m", "pytest", "-q", node_id],
        cwd=sandbox,
        capture_output=True,
        text=True,
        check=False,
    )
    output = (completed.stdout + completed.stderr).strip()
    return CheckResult(passed=completed.returncode == 0, output=output)


def run_contracts(candidate_source: str) -> dict[str, CheckResult]:
    with tempfile.TemporaryDirectory(prefix="policy-repair-") as temporary:
        sandbox = Path(temporary)
        workspace = sandbox / "workspace"
        contracts = sandbox / "contracts"
        workspace.mkdir()
        contracts.mkdir()
        (workspace / "__init__.py").write_text("", encoding="utf-8")
        (workspace / "retry_policy.py").write_text(
            candidate_source, encoding="utf-8"
        )
        shutil.copy2(CONTRACT_PATH, contracts / "test_retry_policy.py")

        return {
            label: run_one_contract(sandbox, node_id)
            for label, node_id in CONTRACT_NODES.items()
        }


def serialize_checks(checks: dict[str, CheckResult]) -> dict:
    return {
        label: {"passed": result.passed, "output": result.output[-3000:]}
        for label, result in checks.items()
    }


def append_audit(event: dict) -> None:
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **event,
    }
    with AUDIT_PATH.open("a", encoding="utf-8") as audit_file:
        audit_file.write(json.dumps(record, sort_keys=True) + "\n")


def apply_files(files: dict[str, str]) -> None:
    for relative_name, content in files.items():
        destination = ROOT / relative_name
        temporary = destination.with_suffix(destination.suffix + ".accepted")
        temporary.write_text(content, encoding="utf-8")
        temporary.replace(destination)


def run_controller(provider: Provider) -> bool:
    policy = load_policy()
    max_attempts = int(policy["max_attempts"])
    source = TARGET_PATH.read_text(encoding="utf-8")
    contract = CONTRACT_PATH.read_text(encoding="utf-8")
    feedback = "No previous attempt has been evaluated."

    for attempt in range(1, max_attempts + 1):
        try:
            proposal = provider.propose(source, contract, feedback, attempt)
        except Exception as error:
            feedback = f"provider_error: {type(error).__name__}: {error}"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "reason": feedback,
                }
            )
            continue

        path_error = validate_paths(proposal.files, policy)
        if path_error is not None:
            feedback = path_error
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": path_error,
                    "proposed_paths": sorted(proposal.files),
                }
            )
            continue

        if not proposal.transient_retry_reason.strip() or not (
            proposal.permanent_error_safety_reason.strip()
        ):
            feedback = "missing_policy_reasoning"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": feedback,
                }
            )
            continue

        checks = run_contracts(proposal.files[TARGET_RELATIVE_PATH])
        check_log = serialize_checks(checks)
        if not all(result.passed for result in checks.values()):
            feedback = "contract_failure\n" + "\n".join(
                f"[{label}]\n{result.output}"
                for label, result in checks.items()
                if not result.passed
            )
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": "contract_failure",
                    "checks": check_log,
                }
            )
            continue

        apply_files(proposal.files)
        append_audit(
            {
                "event": "accepted",
                "provider": provider.name,
                "attempt": attempt,
                "request_id": proposal.request_id,
                "acceptance_reason": (
                    "Accepted only after the immutable contract suite passed and "
                    "the proposal supplied explicit reasoning for both policy duties."
                ),
                "policy_evidence": {
                    "transient_retry": {
                        "requirement": "Retry 408, 409, 429, and 5xx responses.",
                        "candidate_reason": proposal.transient_retry_reason,
                        "contract": CONTRACT_NODES["transient_retry"],
                        "result": "passed",
                    },
                    "permanent_error_safety": {
                        "requirement": (
                            "Do not retry permanent 400, 401, 403, 404, or 422 errors."
                        ),
                        "candidate_reason": proposal.permanent_error_safety_reason,
                        "contract": CONTRACT_NODES["permanent_error_safety"],
                        "result": "passed",
                    },
                },
                "checks": check_log,
            }
        )
        print("Accepted repair. Review the final audit record for policy evidence.")
        return True

    append_audit(
        {
            "event": "attempts_exhausted",
            "provider": provider.name,
            "max_attempts": max_attempts,
        }
    )
    print(f"No repair accepted after {max_attempts} attempts.")
    return False


def print_current_checks() -> None:
    checks = run_contracts(TARGET_PATH.read_text(encoding="utf-8"))
    print(json.dumps(serialize_checks(checks), indent=2))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Safely evaluate deterministic or Claude-generated repairs."
    )
    parser.add_argument(
        "--provider",
        choices=("deterministic", "claude"),
        default="deterministic",
    )
    parser.add_argument("--reset", action="store_true")
    parser.add_argument("--check-current", action="store_true")
    args = parser.parse_args()

    if args.reset:
        reset_workspace()
    if args.check_current:
        print_current_checks()
        return

    provider: Provider
    if args.provider == "claude":
        provider = ClaudeProvider()
    else:
        provider = DeterministicRushedProvider()

    success = run_controller(provider)
    raise SystemExit(0 if success else 1)


if __name__ == "__main__":
    main()

Before you run the provider, how many proposals do you think the policy permits before the controller stops?

• Run the deterministic repair attempt from the activated environment by using:

python repair_controller.py --provider deterministic --reset

What happened during the run?

You should see No repair accepted after 2 attempts. The command exits unsuccessfully because rejection is the intended result.

The first proposal is blocked for targeting a protected file. The second reaches the sandbox but fails the permanent-error contract.

• Print the audit trail in Terminal by running:

cat audit.jsonl

What should the audit show?

• Find an attempt_rejected event whose reason contains protected_file.
• Find a second attempt_rejected event whose reason is contract_failure.
• Find a final attempts_exhausted event with max_attempts set to 2.

• Print the real workspace policy by running:

cat workspace/retry_policy.py

What proves the workspace stayed safe?

You should see the known-bad baseline with return False. Neither rejected proposal changed the real file.

That is the hard part done. Your controller now rejects unsafe edits while preserving evidence for every decision.

Missing the expected audit events?

Check that .repair-policy.json still protects contracts/test_retry_policy.py. Confirm that max_attempts is 2.

Ask for help reading the records: Help me diagnose why my deterministic repair audit does not show protected_file, contract_failure, and attempts_exhausted.

Your deterministic candidate has proved that the guardrails hold. Next, you will let Claude propose a repair that can earn acceptance through separate policy evidence.

Accept a Policy-Aware Claude Repair

Your deterministic run proved that the controller blocks protected-file edits. It also proved that an unsafe allowed-file patch still fails the independent contracts.

Now you will connect the controller to Claude through the Anthropic API. The controller will accept Claude's proposal only after separate tests prove both sides of the retry policy.

In this step, get ready to:

• Create a personal development API key for the live repair.
• Add structured Claude proposals to the existing repair controller.
• Run the live repair and inspect its policy evidence.

Create and export your Claude API key

An API key authorizes the SDK to send your repair request. The key belongs in the current Terminal environment so it stays outside every project file.

This next run uses paid API tokens. Claude Haiku 4.5 costs $1 per million input tokens plus $5 per million output tokens.

You stay in control because the request starts only when you run the Claude provider. You can keep the deterministic controller without making this paid call.

• Open your browser and navigate to platform.claude.com.
• Sign in to your Claude Console account.
• Click Settings.
• Click API keys.
• Click Create key.
• Enter policy-aware-repair as the key name.
• Choose an expiration for the development key.
• Link the key to your account.

Protect the one-time key reveal

The full key appears only once after creation. Store it in a password manager before closing the reveal screen.

Keep the key out of repair_controller.py, audit.jsonl, and every other project file.

• Complete the key form to generate your personal development key.
• Copy the key before closing the reveal screen.
• Store the copied key in your password manager.
• Switch back to the activated Terminal session from earlier.
• Export the key for the current Terminal session by replacing your-api-key-here in this command:

export ANTHROPIC_API_KEY="your-api-key-here"

What does this command do?

The command stores your key in ANTHROPIC_API_KEY for the current Terminal session. The Anthropic SDK reads that environment variable automatically.

Closing this Terminal session removes the exported value. Your project files never receive the credential.

Key unavailable in the live run?

Make sure you run the export command in the same Terminal session that runs the controller. A newly opened session does not inherit the earlier value.

Check that you replaced the placeholder with the complete key. Keep the key private while asking for help.

Help me diagnose why the Anthropic SDK cannot read my API key without asking me to share the key.

Connect the controller to Claude

The provider needs machine-checkable output because free-form model text can omit a file or policy explanation. A JSON Schema requires Claude to return proposed files plus separate reasoning for transient retries and permanent-error safety.

Why disable SDK retries?

The Anthropic SDK normally retries selected connection and HTTP failures. Setting max_retries=0 keeps the controller's own two-attempt limit visible in the audit trail.

This setting governs requests to Claude. The candidate code still repairs the separate HTTP retry policy in workspace/retry_policy.py.

• Update repair_controller.py by matching it to the completed Claude-enabled file in the comparison tab below.

✔️ I've added the Claude integration

Your controller now imports the Anthropic SDK. It also defines MODEL, OUTPUT_CONFIG, and ClaudeProvider.

The command-line provider choices now include claude.

ⓧ I'd like to double check the full code

Compare your repair_controller.py with this complete version:

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
from typing import Protocol

import anthropic

ROOT = Path(__file__).resolve().parent
POLICY_PATH = ROOT / ".repair-policy.json"
BASELINE_PATH = ROOT / "baseline/rushed_retry_policy.py"
TARGET_RELATIVE_PATH = "workspace/retry_policy.py"
TARGET_PATH = ROOT / TARGET_RELATIVE_PATH
CONTRACT_PATH = ROOT / "contracts/test_retry_policy.py"
AUDIT_PATH = ROOT / "audit.jsonl"
MODEL = "claude-haiku-4-5-20251001"

CONTRACT_NODES = {
    "transient_retry": (
        "contracts/test_retry_policy.py::test_retries_transient_failures"
    ),
    "permanent_error_safety": (
        "contracts/test_retry_policy.py::test_does_not_retry_permanent_errors"
    ),
    "full_suite": "contracts/test_retry_policy.py",
}

OUTPUT_CONFIG = {
    "format": {
        "type": "json_schema",
        "schema": {
            "type": "object",
            "properties": {
                "files": {
                    "type": "array",
                    "minItems": 1,
                    "items": {
                        "type": "object",
                        "properties": {
                            "path": {"type": "string"},
                            "content": {"type": "string"},
                        },
                        "required": ["path", "content"],
                        "additionalProperties": False,
                    },
                },
                "transient_retry_reason": {"type": "string"},
                "permanent_error_safety_reason": {"type": "string"},
            },
            "required": [
                "files",
                "transient_retry_reason",
                "permanent_error_safety_reason",
            ],
            "additionalProperties": False,
        },
    }
}


@dataclass(frozen=True)
class Proposal:
    files: dict[str, str]
    transient_retry_reason: str
    permanent_error_safety_reason: str
    request_id: str | None = None


@dataclass(frozen=True)
class CheckResult:
    passed: bool
    output: str


class Provider(Protocol):
    name: str

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        ...


class DeterministicRushedProvider:
    name = "deterministic-rushed"

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        del source, contract, feedback
        unsafe_source = '''"""Unsafe rushed retry policy."""


def should_retry(status_code: int) -> bool:
    return status_code >= 400
'''
        if attempt == 1:
            return Proposal(
                files={
                    TARGET_RELATIVE_PATH: unsafe_source,
                    "contracts/test_retry_policy.py": (
                        "def test_contracts_were_weakened():\n    assert True\n"
                    ),
                },
                transient_retry_reason="All failed requests are retried.",
                permanent_error_safety_reason=(
                    "The tests were changed so permanent errors no longer block acceptance."
                ),
            )
        return Proposal(
            files={TARGET_RELATIVE_PATH: unsafe_source},
            transient_retry_reason="Every error status retries, including transient ones.",
            permanent_error_safety_reason=(
                "The implementation is short, so the permanent-error risk was ignored."
            ),
        )


class ClaudeProvider:
    name = "claude-haiku-4.5"

    def __init__(self) -> None:
        self.client = anthropic.Anthropic(max_retries=0, timeout=45.0)

    def propose(
        self,
        source: str,
        contract: str,
        feedback: str,
        attempt: int,
    ) -> Proposal:
        prompt = f"""
Repair the Python source below. Treat the contract suite as immutable.
Return only the files that need changing. The accepted implementation must retry
transient HTTP failures 408, 409, 429, and 5xx responses, while never retrying
permanent 4xx errors or successful responses.

Attempt: {attempt}
Previous controller feedback:
{feedback}

Current source at {TARGET_RELATIVE_PATH}:
```python
{source}
```

Immutable contract suite:
```python
{contract}
```

Explain separately why your implementation satisfies the transient-retry
requirement and the permanent-error safety contract.
""".strip()

        response = self.client.messages.create(
            model=MODEL,
            max_tokens=2048,
            system=(
                "You repair one small Python module. Never modify tests, policy files, "
                "controller code, dependency files, or any path outside the editable target."
            ),
            messages=[{"role": "user", "content": prompt}],
            output_config=OUTPUT_CONFIG,
        )
        text = next(
            block.text for block in response.content if block.type == "text"
        )
        payload = json.loads(text)
        files: dict[str, str] = {}
        for item in payload["files"]:
            path = item["path"]
            if path in files:
                raise ValueError(f"Duplicate path in proposal: {path}")
            files[path] = item["content"]

        return Proposal(
            files=files,
            transient_retry_reason=payload["transient_retry_reason"],
            permanent_error_safety_reason=payload[
                "permanent_error_safety_reason"
            ],
            request_id=response._request_id,
        )


def load_policy() -> dict:
    return json.loads(POLICY_PATH.read_text(encoding="utf-8"))


def reset_workspace() -> None:
    TARGET_PATH.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(BASELINE_PATH, TARGET_PATH)


def validate_paths(files: dict[str, str], policy: dict) -> str | None:
    editable = set(policy["editable_paths"])
    protected = set(policy["protected_paths"])

    for name in files:
        path = PurePosixPath(name)
        if path.is_absolute() or ".." in path.parts:
            return f"unsafe_path: {name}"
        if name in protected:
            return f"protected_file: {name}"
        if name not in editable:
            return f"outside_editable_allowlist: {name}"

    if TARGET_RELATIVE_PATH not in files:
        return f"missing_required_target: {TARGET_RELATIVE_PATH}"
    return None


def run_one_contract(sandbox: Path, node_id: str) -> CheckResult:
    completed = subprocess.run(
        [sys.executable, "-m", "pytest", "-q", node_id],
        cwd=sandbox,
        capture_output=True,
        text=True,
        check=False,
    )
    output = (completed.stdout + completed.stderr).strip()
    return CheckResult(passed=completed.returncode == 0, output=output)


def run_contracts(candidate_source: str) -> dict[str, CheckResult]:
    with tempfile.TemporaryDirectory(prefix="policy-repair-") as temporary:
        sandbox = Path(temporary)
        workspace = sandbox / "workspace"
        contracts = sandbox / "contracts"
        workspace.mkdir()
        contracts.mkdir()
        (workspace / "__init__.py").write_text("", encoding="utf-8")
        (workspace / "retry_policy.py").write_text(
            candidate_source, encoding="utf-8"
        )
        shutil.copy2(CONTRACT_PATH, contracts / "test_retry_policy.py")

        return {
            label: run_one_contract(sandbox, node_id)
            for label, node_id in CONTRACT_NODES.items()
        }


def serialize_checks(checks: dict[str, CheckResult]) -> dict:
    return {
        label: {"passed": result.passed, "output": result.output[-3000:]}
        for label, result in checks.items()
    }


def append_audit(event: dict) -> None:
    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        **event,
    }
    with AUDIT_PATH.open("a", encoding="utf-8") as audit_file:
        audit_file.write(json.dumps(record, sort_keys=True) + "\n")


def apply_files(files: dict[str, str]) -> None:
    for relative_name, content in files.items():
        destination = ROOT / relative_name
        temporary = destination.with_suffix(destination.suffix + ".accepted")
        temporary.write_text(content, encoding="utf-8")
        temporary.replace(destination)


def run_controller(provider: Provider) -> bool:
    policy = load_policy()
    max_attempts = int(policy["max_attempts"])
    source = TARGET_PATH.read_text(encoding="utf-8")
    contract = CONTRACT_PATH.read_text(encoding="utf-8")
    feedback = "No previous attempt has been evaluated."

    for attempt in range(1, max_attempts + 1):
        try:
            proposal = provider.propose(source, contract, feedback, attempt)
        except Exception as error:
            feedback = f"provider_error: {type(error).__name__}: {error}"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "reason": feedback,
                }
            )
            continue

        path_error = validate_paths(proposal.files, policy)
        if path_error is not None:
            feedback = path_error
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": path_error,
                    "proposed_paths": sorted(proposal.files),
                }
            )
            continue

        if not proposal.transient_retry_reason.strip() or not (
            proposal.permanent_error_safety_reason.strip()
        ):
            feedback = "missing_policy_reasoning"
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": feedback,
                }
            )
            continue

        checks = run_contracts(proposal.files[TARGET_RELATIVE_PATH])
        check_log = serialize_checks(checks)
        if not all(result.passed for result in checks.values()):
            feedback = "contract_failure\n" + "\n".join(
                f"[{label}]\n{result.output}"
                for label, result in checks.items()
                if not result.passed
            )
            append_audit(
                {
                    "event": "attempt_rejected",
                    "provider": provider.name,
                    "attempt": attempt,
                    "request_id": proposal.request_id,
                    "reason": "contract_failure",
                    "checks": check_log,
                }
            )
            continue

        apply_files(proposal.files)
        append_audit(
            {
                "event": "accepted",
                "provider": provider.name,
                "attempt": attempt,
                "request_id": proposal.request_id,
                "acceptance_reason": (
                    "Accepted only after the immutable contract suite passed and "
                    "the proposal supplied explicit reasoning for both policy duties."
                ),
                "policy_evidence": {
                    "transient_retry": {
                        "requirement": "Retry 408, 409, 429, and 5xx responses.",
                        "candidate_reason": proposal.transient_retry_reason,
                        "contract": CONTRACT_NODES["transient_retry"],
                        "result": "passed",
                    },
                    "permanent_error_safety": {
                        "requirement": (
                            "Do not retry permanent 400, 401, 403, 404, or 422 errors."
                        ),
                        "candidate_reason": proposal.permanent_error_safety_reason,
                        "contract": CONTRACT_NODES["permanent_error_safety"],
                        "result": "passed",
                    },
                },
                "checks": check_log,
            }
        )
        print("Accepted repair. Review the final audit record for policy evidence.")
        return True

    append_audit(
        {
            "event": "attempts_exhausted",
            "provider": provider.name,
            "max_attempts": max_attempts,
        }
    )
    print(f"No repair accepted after {max_attempts} attempts.")
    return False


def print_current_checks() -> None:
    checks = run_contracts(TARGET_PATH.read_text(encoding="utf-8"))
    print(json.dumps(serialize_checks(checks), indent=2))


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Safely evaluate deterministic or Claude-generated repairs."
    )
    parser.add_argument(
        "--provider",
        choices=("deterministic", "claude"),
        default="deterministic",
    )
    parser.add_argument("--reset", action="store_true")
    parser.add_argument("--check-current", action="store_true")
    args = parser.parse_args()

    if args.reset:
        reset_workspace()
    if args.check_current:
        print_current_checks()
        return

    provider: Provider
    if args.provider == "claude":
        provider = ClaudeProvider()
    else:
        provider = DeterministicRushedProvider()

    success = run_controller(provider)
    raise SystemExit(0 if success else 1)


if __name__ == "__main__":
    main()

How the Claude integration works

• The MODEL constant pins the request to claude-haiku-4-5-20251001.
• The OUTPUT_CONFIG schema requires at least one proposed file plus both policy explanations.
• The ClaudeProvider sends the current source and immutable contract to Claude.
• The duplicate-path check rejects ambiguous proposals before they reach the controller.
• The response request ID follows the proposal into audit.jsonl.

• Save repair_controller.py.

Run the live repair and inspect evidence

The controller resets the workspace to the known-bad baseline before asking Claude for a proposal. The existing allowlist and independent contracts still decide whether that proposal is accepted.

Before you run this, do you expect the Claude proposal to pass both policy contracts on its first attempt?

• Run the live repair from the activated Terminal session with this command:

python repair_controller.py --provider claude --reset

What happens during this run?

The controller sends the broken source and trusted contract to Claude. Structured output returns proposed files plus two separate policy explanations.

The proposal passes through path validation before entering a temporary sandbox. Atomic replacement updates the real workspace only after every contract passes.

You should see Accepted repair. Review the final audit record for policy evidence. in Terminal. This confirms that the immutable suite accepted the proposed implementation.

Live repair rejected?

An authentication failure usually means the current Terminal session cannot read ANTHROPIC_API_KEY. Export the key again in that same session without placing it in a file.

A provider failure can also come from account access or billing. Read the latest rejection reason in audit.jsonl before running another paid request.

Help me interpret my Claude provider rejection without asking me to share my API key.

• Print the final record from audit.jsonl by running:

tail -n 1 audit.jsonl

What should the final record prove?

• The event field is accepted.
• The request_id field identifies the Claude API request.
• The transient_retry evidence has a passed result plus Claude's explanation.
• The permanent_error_safety evidence has a passed result plus a separate explanation.
• The recorded contracts show which independent test proved each policy claim.

The accepted workspace/retry_policy.py now retries 408, 409, 429, and server failures. Permanent client errors and successful responses remain safe from retries.

That is the full safety case working: Claude supplied the repair, while your controller supplied the proof required to trust it.

Secret mission

Stop When Repairs Make No Progress

A two-attempt limit still wastes work when a provider submits the same rejected patch twice. Add a SHA-256 proposal digest that identifies repeated work before another sandbox run. Then prove the controller stops with a precise no-progress audit event.

Clean Up Your Resources

Clean Up Your Resources

Only live Claude API requests cost money because the controller has no background service. Choose whether to keep your resources, pause API access, or delete everything.

Resources you used:

• A personal development API key in Claude Console.
• The ANTHROPIC_API_KEY variable in your current macOS Terminal session.
• The policy-aware-repair folder with all local project artifacts.

Keep everything running

No action is needed. Choose this if you want to keep testing the repair controller or build more providers.

• Keep the policy-aware-repair folder for future experiments.
• Keep the personal development key active for future live repairs.
• Run the controller only when you are ready to make another paid Claude API request.

Pause - I'll come back to this later

Clear the current shell variable to prevent an accidental live request. Your project folder stays ready for your return.

• Return to the Terminal session from earlier.
• Clear the exported API key from that shell by running this command:

unset ANTHROPIC_API_KEY

What does this command do?

The unset command removes ANTHROPIC_API_KEY from the current Terminal session. It leaves your project files unchanged.

The personal development key stays active in Claude Console until you revoke it.

Delete - I don't want to use this again

Remove every project resource if you want to start fresh. Deleting the folder is permanent, but you can rebuild the project from this guide.

• Sign in to Claude Console.
• Open Settings.
• Select API keys.
• Find the personal development key you created for this project.
• Revoke the personal development key.

• Return to the Terminal session from earlier.
• Remove the exported API key from that shell by running this command:

unset ANTHROPIC_API_KEY

What does this command do?

The unset command clears the credential from the current Terminal session. Revoking the Console key prevents that credential from authorizing future API requests.

Your project files contain no copy of the API key.

• Use the macOS file browser to locate the policy-aware-repair folder.
• Move the policy-aware-repair folder to Trash.
• Permanently remove the policy-aware-repair folder from Trash.
• Confirm that policy-aware-repair no longer appears in Trash.

Nice Work!

Nice Work!

You pulled it off! You built a policy-aware repair controller that treats every model-generated patch as an untrusted proposal.

You've learned how to:

• Built a deliberately failing retry policy that makes missing transient retries visible. An independent pytest contract proves the required retry behavior. Separate contracts protect permanent errors. Another contract protects successful responses.
• Evaluated every proposal inside a temporary sandbox. A protected-file allowlist rejects attempts to rewrite trusted contracts. Bounded attempts prevent an endless repair loop. Atomic replacement keeps rejected patches out of workspace/retry_policy.py. A JSONL audit trail records every decision.
• Used Claude Haiku 4.5 structured output to receive a machine-checkable repair proposal. The controller accepted the live patch only after independent checks passed. The final audit evidence records the Claude request ID. It records the transient-retry rationale. It also records the permanent-error safety rationale.
• Completed the Secret Mission by adding SHA-256 no-progress detection. A repeated proposal digest now produces a no_progress event before sandbox execution. The controller stops immediately without changing workspace/retry_policy.py.

Ready to quiz yourself?