Recommend a Code Repair for Review

Day 06 handed you architecture-selection.json, untouched-test results, and deterministic selection semantics. Today you will implement a bounded local coding-agent runner that samples repairs in parallel, verifies each in a separate temporary Git copy, writes selected code only to a review directory, emits a machine-readable receipt, and fails closed if the target repo changes. Boundary is a release recommendation, not production execution or automatic merge.

CodeMonkeys explores serial and parallel test-time compute for real software issues (paper v2, official repository at inspected commit). Its selection pipeline uses generated-test voting, retains a top-three set, then uses a model selection trajectory. This local reimplementation instead selects only from sealed-suite passes; it is not a CodeMonkeys reproduction. Generated-test voting is a later-course extension. Diagram is a conceptual reconstruction, not a source-figure transcription.

The program below requires Python 3 on macOS or Linux and no packages. --model-cmd accepts a command that runs inside each copied workspace; it receives TTC_PROMPT, TTC_CANDIDATE, and TTC_INDEX environment variables and must write the configured artifact. --demo skips a real model and creates three deterministic candidates. A timed-out command terminates its POSIX process group before the temporary workspace is removed.

Temporary directories provide filesystem separation from the target, not a hostile-code security sandbox. They do not block network, process, credential, or kernel access. Back-route to Day 02: agent loop and sandbox, Day 03: generated-code validation, Day 04: falsification and security, and Day 10: recovery evidence before running untrusted code. This local harness teaches compute allocation and candidate selection only.

#!/usr/bin/env python3
"""Bounded, review-only test-time compute runner using Python stdlib."""

from __future__ import annotations

import argparse
import concurrent.futures
import dataclasses
import hashlib
import json
import os
from pathlib import Path
import signal
import shlex
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Sequence


@dataclasses.dataclass
class Step:
    exit_code: int | None
    output_bytes: int
    status: str


@dataclasses.dataclass
class Candidate:
    index: int
    artifact: bytes
    artifact_sha256: str | None
    artifact_bytes: int
    generation: Step
    public_test: Step | None
    hidden_test: Step | None
    status: str
    passed: bool

    def receipt(self) -> dict[str, object]:
        def encode(step: Step | None) -> dict[str, object] | None:
            return dataclasses.asdict(step) if step else None

        return {
            "index": self.index,
            "artifact_bytes": self.artifact_bytes,
            "artifact_sha256": self.artifact_sha256,
            "generation": encode(self.generation),
            "public_test": encode(self.public_test),
            "hidden_test": encode(self.hidden_test),
            "status": self.status,
            "passed": self.passed,
        }


def digest_tree(root: Path) -> str:
    digest = hashlib.sha256()
    for path in sorted(
        p for p in root.rglob("*")
        if p.is_file() and ".git" not in p.parts and "__pycache__" not in p.parts
    ):
        digest.update(path.relative_to(root).as_posix().encode())
        digest.update(b"\0")
        digest.update(path.read_bytes())
        digest.update(b"\0")
    return digest.hexdigest()


def file_hashes(root: Path) -> dict[str, str]:
    return {
        path.relative_to(root).as_posix(): hashlib.sha256(path.read_bytes()).hexdigest()
        for path in root.rglob("*")
        if path.is_file() and ".git" not in path.parts
    }


def remaining(deadline: float, per_step: float) -> float:
    seconds = min(per_step, deadline - time.monotonic())
    if seconds <= 0:
        raise TimeoutError("global wall-time budget exhausted")
    return seconds


def process_group_exists(process_group: int) -> bool:
    try:
        os.killpg(process_group, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return False
    return True


def terminate_process_group(process_group: int) -> bool:
    if not process_group_exists(process_group):
        return True
    try:
        os.killpg(process_group, signal.SIGTERM)
    except ProcessLookupError:
        return True
    except PermissionError:
        return False
    stop_deadline = time.monotonic() + 1
    while process_group_exists(process_group) and time.monotonic() < stop_deadline:
        time.sleep(0.01)
    if process_group_exists(process_group):
        try:
            os.killpg(process_group, signal.SIGKILL)
        except ProcessLookupError:
            return True
        except PermissionError:
            return False
    stop_deadline = time.monotonic() + 1
    while process_group_exists(process_group) and time.monotonic() < stop_deadline:
        time.sleep(0.01)
    return not process_group_exists(process_group)


def run_step(
    command: Sequence[str],
    cwd: Path,
    timeout: float,
    max_output_bytes: int,
    env: dict[str, str] | None = None,
) -> Step:
    run_env = os.environ.copy()
    if env:
        run_env.update(env)
    run_env["PYTHONDONTWRITEBYTECODE"] = "1"
    process = subprocess.Popen(
        list(command),
        cwd=cwd,
        env=run_env,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        start_new_session=True,
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(process.pid, signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            pass
        try:
            stdout, stderr = process.communicate(timeout=1)
        except subprocess.TimeoutExpired:
            try:
                os.killpg(process.pid, signal.SIGKILL)
            except (ProcessLookupError, PermissionError):
                pass
            stdout, stderr = process.communicate()
        group_stopped = terminate_process_group(process.pid)
        status = "timeout" if group_stopped else "process_leak"
        return Step(None, len(stdout) + len(stderr), status)

    size = len(stdout) + len(stderr)
    if process_group_exists(process.pid):
        group_stopped = terminate_process_group(process.pid)
        status = "background_process" if group_stopped else "process_leak"
        return Step(process.returncode, size, status)
    if size > max_output_bytes:
        return Step(process.returncode, size, "output_limit")
    status = "pass" if process.returncode == 0 else "fail"
    return Step(process.returncode, size, status)


def demo_source(index: int) -> bytes:
    candidates = [
        """def discounted_cents(price_cents: int, discount_bps: int) -> int:
    if price_cents < 0 or not 0 <= discount_bps <= 10_000:
        raise ValueError("invalid price or discount")
    return round(price_cents * (10_000 - discount_bps) / 10_000)
""",
        """def discounted_cents(price_cents: int, discount_bps: int) -> int:
    if price_cents < 0 or not 0 <= discount_bps <= 10_000:
        raise ValueError("invalid price or discount")
    return price_cents * (10_000 - discount_bps) // 10_000
""",
        """def discounted_cents(price_cents: int, discount_bps: int) -> int:
    return 900
""",
    ]
    if index >= len(candidates):
        raise ValueError("demo supports at most three samples")
    return candidates[index].encode()


def evaluate(
    index: int,
    target: Path,
    temp_root: Path,
    args: argparse.Namespace,
    deadline: float,
) -> Candidate:
    workspace = temp_root / f"candidate-{index}"
    shutil.copytree(target, workspace, ignore=shutil.ignore_patterns(".git", args.hidden_test))
    baseline_files = file_hashes(workspace)
    artifact_path = (workspace / args.artifact).resolve()
    if workspace.resolve() not in artifact_path.parents:
        return Candidate(index, b"", None, 0, Step(None, 0, "invalid_artifact_path"), None, None, "out_of_scope_change", False)

    if args.demo:
        artifact_path.parent.mkdir(parents=True, exist_ok=True)
        artifact_path.write_bytes(demo_source(index))
        generation = Step(0, 0, "pass")
    else:
        command = shlex.split(args.model_cmd.format(index=index))
        if not command:
            return Candidate(index, b"", None, 0, Step(None, 0, "empty_command"), None, None, "generation_failed", False)
        env = os.environ.copy()
        env.update(
            {
                "TTC_PROMPT": args.prompt,
                "TTC_CANDIDATE": str(artifact_path),
                "TTC_INDEX": str(index),
            }
        )
        try:
            generation = run_step(
                command,
                workspace,
                remaining(deadline, args.command_timeout),
                args.max_output_bytes,
                env,
            )
        except TimeoutError:
            generation = Step(None, 0, "global_timeout")

    after_files = file_hashes(workspace)
    changed = {
        name
        for name in set(baseline_files) | set(after_files)
        if baseline_files.get(name) != after_files.get(name)
    }
    if not changed.issubset(set(args.allow_path)):
        return Candidate(index, b"", None, 0, generation, None, None, "out_of_scope_change", False)

    if generation.status != "pass" or not artifact_path.is_file():
        status = "artifact_missing" if generation.status == "pass" else generation.status
        return Candidate(index, b"", None, 0, generation, None, None, status, False)

    artifact_size = artifact_path.stat().st_size
    if artifact_size > args.max_candidate_bytes:
        return Candidate(index, b"", None, artifact_size, generation, None, None, "candidate_byte_limit", False)
    artifact = artifact_path.read_bytes()
    artifact_hash = hashlib.sha256(artifact).hexdigest()

    try:
        public = run_step(
            [sys.executable, args.public_test],
            workspace,
            remaining(deadline, args.command_timeout),
            args.max_output_bytes,
        )
    except TimeoutError:
        public = Step(None, 0, "global_timeout")
    if public.status != "pass":
        return Candidate(index, artifact, artifact_hash, len(artifact), generation, public, None, "public_test_failed", False)

    try:
        hidden = run_step(
            [sys.executable, str(target / args.hidden_test), str(artifact_path)],
            workspace,
            remaining(deadline, args.command_timeout),
            args.max_output_bytes,
        )
    except TimeoutError:
        hidden = Step(None, 0, "global_timeout")
    passed = hidden.status == "pass" and time.monotonic() <= deadline
    status = "passed" if passed else "hidden_test_failed"
    return Candidate(index, artifact, artifact_hash, len(artifact), generation, public, hidden, status, passed)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--target", required=True)
    parser.add_argument("--review-output", required=True)
    parser.add_argument("--architecture-selection", required=True)
    parser.add_argument("--artifact", default="pricing.py")
    parser.add_argument("--public-test", default="public_test.py")
    parser.add_argument("--hidden-test", default="sealed_test.py")
    parser.add_argument("--target-commit", required=True)
    parser.add_argument("--allow-path", action="append", default=["pricing.py"])
    parser.add_argument("--prompt", default="Repair pricing.py without editing tests")
    parser.add_argument("--model-cmd", default="")
    parser.add_argument("--demo", action="store_true")
    parser.add_argument("--samples", type=int, default=3)
    parser.add_argument("--workers", type=int, default=3)
    parser.add_argument("--budget-units", type=int, default=3)
    parser.add_argument("--command-timeout", type=float, default=5.0)
    parser.add_argument("--max-wall-seconds", type=float, default=20.0)
    parser.add_argument("--max-candidate-bytes", type=int, default=8192)
    parser.add_argument("--max-total-candidate-bytes", type=int, default=24576)
    parser.add_argument("--max-output-bytes", type=int, default=8192)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    target = Path(args.target).resolve(strict=True)
    review = Path(args.review_output).resolve()
    architecture_path = Path(args.architecture_selection).resolve(strict=True)
    architecture = json.loads(architecture_path.read_text(encoding="utf-8"))
    if architecture.get("schema") != "architecture-selection/v1":
        raise SystemExit("unsupported architecture-selection schema")
    if architecture.get("release_decision") != "eligible_for_local_capstone":
        raise SystemExit("architecture is not eligible for local capstone")
    if architecture.get("selection_split") != "development_plus_validation":
        raise SystemExit("architecture must be selected on development plus validation")
    if architecture.get("test_opened_once") is not True:
        raise SystemExit("untouched test evidence is required before local capstone")
    architecture_id = architecture.get("winner")
    if architecture_id != "width-3":
        raise SystemExit("this capstone implements only the selected width-3 architecture")
    run_profile = architecture.get("run_profile")
    if not isinstance(run_profile, dict):
        raise SystemExit("architecture-selection run_profile is missing")
    requested_profile = {"samples": args.samples, "workers": args.workers, "budget_units": args.budget_units}
    for key, requested in requested_profile.items():
        ceiling = run_profile.get(key)
        if not isinstance(ceiling, int) or ceiling < 1 or requested > ceiling:
            raise SystemExit(f"CLI {key} exceeds architecture-selection run_profile")
    if not target.is_dir():
        raise SystemExit("target must be one existing directory")
    commit_run = subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=target, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False
    )
    target_commit = commit_run.stdout.decode().strip()
    if commit_run.returncode != 0 or target_commit != args.target_commit:
        raise SystemExit("target commit does not match pinned digest")
    clean_run = subprocess.run(
        ["git", "status", "--porcelain=v1"],
        cwd=target,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if clean_run.returncode != 0 or clean_run.stdout.strip():
        raise SystemExit("target must be a clean Git worktree")
    if review == target or target in review.parents:
        raise SystemExit("review output must be outside target")
    if review.exists() and any(review.iterdir()):
        raise SystemExit("review output must be absent or empty")
    if args.samples < 1 or args.workers < 1:
        raise SystemExit("samples and workers must be positive")
    artifact_rel = Path(args.artifact)
    if artifact_rel.is_absolute() or ".." in artifact_rel.parts:
        raise SystemExit("artifact must be one relative allowlisted path")
    if args.samples > args.budget_units:
        raise SystemExit("sample request exceeds budget units")
    if args.demo and args.samples > 3:
        raise SystemExit("demo supports at most three samples")
    if not args.demo and not args.model_cmd:
        raise SystemExit("provide --model-cmd or use --demo")
    for test_name in (args.public_test, args.hidden_test):
        if not (target / test_name).is_file():
            raise SystemExit(f"missing test: {test_name}")

    before = digest_tree(target)
    deadline = time.monotonic() + args.max_wall_seconds
    temp_path = ""
    with tempfile.TemporaryDirectory(prefix="ttc-run-") as temp_name:
        temp_path = temp_name
        temp_root = Path(temp_name)
        baseline_workspace = temp_root / "baseline"
        shutil.copytree(target, baseline_workspace, ignore=shutil.ignore_patterns(".git"))
        baseline_public = run_step(
            [sys.executable, args.public_test],
            baseline_workspace,
            args.command_timeout,
            args.max_output_bytes,
        )
        baseline_sealed = run_step(
            [sys.executable, args.hidden_test, str(baseline_workspace / args.artifact)],
            baseline_workspace,
            args.command_timeout,
            args.max_output_bytes,
        )
        if baseline_public.status != "pass":
            raise SystemExit("baseline public test must pass before candidate generation")
        if baseline_sealed.status != "fail":
            raise SystemExit("sealed test must reproduce the target defect before candidate generation")
        with concurrent.futures.ThreadPoolExecutor(max_workers=args.workers) as executor:
            futures = [
                executor.submit(evaluate, index, target, temp_root, args, deadline)
                for index in range(args.samples)
            ]
            candidates = sorted((future.result() for future in futures), key=lambda item: item.index)

        cumulative = 0
        for candidate in candidates:
            cumulative += candidate.artifact_bytes
            if cumulative > args.max_total_candidate_bytes:
                candidate.passed = False
                candidate.status = "aggregate_byte_limit"

    cleanup_verified = not Path(temp_path).exists()
    after = digest_tree(target)
    post_clean_run = subprocess.run(
        ["git", "status", "--porcelain=v1"],
        cwd=target,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    target_unchanged = before == after and post_clean_run.returncode == 0 and not post_clean_run.stdout.strip()
    passing = [candidate for candidate in candidates if candidate.passed]
    selected = (
        min(passing, key=lambda item: (item.artifact_bytes, item.index))
        if passing and target_unchanged
        else None
    )
    review.mkdir(parents=True, exist_ok=True)
    if selected:
        selected_dir = review / "selected"
        selected_dir.mkdir()
        (selected_dir / args.artifact).write_bytes(selected.artifact)

    receipt = {
        "allow_paths": sorted(set(args.allow_path)),
        "architecture_id": architecture_id,
        "baseline": {
            "public_test": dataclasses.asdict(baseline_public),
            "sealed_test": dataclasses.asdict(baseline_sealed),
        },
        "budget_units": args.budget_units,
        "candidates": [candidate.receipt() for candidate in candidates],
        "cleanup_verified": cleanup_verified,
        "fixture_mode": args.demo,
        "limits": {
            "command_timeout": args.command_timeout,
            "max_candidate_bytes": args.max_candidate_bytes,
            "max_output_bytes": args.max_output_bytes,
            "max_total_candidate_bytes": args.max_total_candidate_bytes,
            "max_wall_seconds": args.max_wall_seconds,
        },
        "sample_count": args.samples,
        "selected_index": selected.index if selected else None,
        "selection": "REVIEW_RECOMMENDATION" if selected else "NO_SELECTION" if target_unchanged else "TARGET_CHANGED",
        "target_commit": target_commit,
        "target_digest_before": before,
        "target_digest_after": after,
        "target_unchanged": target_unchanged,
    }
    encoded = json.dumps(receipt, sort_keys=True, separators=(",", ":"))
    (review / "receipt.json").write_text(encoded + "\n", encoding="utf-8")
    print(encoded)
    return 0 if selected and target_unchanged else 2


if __name__ == "__main__":
    raise SystemExit(main())

Save the exact Python block above as runner.py in an empty working directory. Then run this complete disposable fixture from that directory; it carries Day 06’s selected width-3 ceiling into every capstone run.

runner_source="$PWD/runner.py"
test -f "$runner_source"
fixture=$(mktemp -d /tmp/shipright-ttc-fixture.XXXXXX)
mkdir -p "$fixture/target"
cd "$fixture"
cat > architecture-selection.json <<'JSON'
{"schema":"architecture-selection/v1","winner":"width-3","run_profile":{"samples":3,"workers":3,"budget_units":3},"selection_split":"development_plus_validation","test_opened_once":true,"release_decision":"eligible_for_local_capstone"}
JSON
cat > target/pricing.py <<'PY'
def discounted_cents(price_cents: int, discount_bps: int) -> int:
    return round(price_cents * (10_000 - discount_bps) / 10_000)
PY
cat > target/public_test.py <<'PY'
from pricing import discounted_cents
assert discounted_cents(1_000, 1_000) == 900
PY
cat > target/sealed_test.py <<'PY'
import importlib.util
from pathlib import Path
import sys

candidate = Path(sys.argv[1])
spec = importlib.util.spec_from_file_location("candidate_pricing", candidate)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
ok = module.discounted_cents(3, 5_000) == 1
try:
    module.discounted_cents(-1, 0)
except ValueError:
    rejected = True
else:
    rejected = False
raise SystemExit(0 if ok and rejected else 1)
PY
cat > target/model_adapter.py <<'PY'
import os
from pathlib import Path

candidate = Path(os.environ["TTC_CANDIDATE"])
candidate.write_text("def discounted_cents(price_cents, discount_bps):\n    return 1\n")
(candidate.parent / "public_test.py").write_text("")
PY
cp "$runner_source" ./runner.py
git -C target init -q
git -C target config user.name "Course Fixture"
git -C target config user.email "fixture@example.invalid"
git -C target add pricing.py public_test.py sealed_test.py model_adapter.py
GIT_AUTHOR_DATE=2026-09-04T00:00:00Z GIT_COMMITTER_DATE=2026-09-04T00:00:00Z \
  git -C target commit -q -m "fixture baseline"
target_commit=$(git -C target rev-parse HEAD)
sha256sum target/pricing.py > before.sha256
python3 runner.py \
  --target target \
  --target-commit "$target_commit" \
  --architecture-selection architecture-selection.json \
  --review-output review \
  --demo \
  --samples 3 \
  --workers 3 \
  --budget-units 3
sha256sum -c before.sha256
python3 review/selected/pricing.py

Expected runner JSON is one line. Hashes below correspond to the exact three demo candidates in the script; sha256sum -c then prints target/pricing.py: OK.

{"allow_paths":["pricing.py"],"architecture_id":"width-3","baseline":{"public_test":{"exit_code":0,"output_bytes":0,"status":"pass"},"sealed_test":{"exit_code":1,"output_bytes":0,"status":"fail"}},"budget_units":3,"candidates":[{"artifact_bytes":244,"artifact_sha256":"d70dd0f7121a148b862e78a95bf53ca1f9603f4895642c7fb6479ec70c54f66c","generation":{"exit_code":0,"output_bytes":0,"status":"pass"},"hidden_test":{"exit_code":1,"output_bytes":0,"status":"fail"},"index":0,"passed":false,"public_test":{"exit_code":0,"output_bytes":0,"status":"pass"},"status":"hidden_test_failed"},{"artifact_bytes":238,"artifact_sha256":"1ddacc3ebeaab5dc79c5da6a6da9af6909109c78effeb66331c82e35979139d9","generation":{"exit_code":0,"output_bytes":0,"status":"pass"},"hidden_test":{"exit_code":0,"output_bytes":0,"status":"pass"},"index":1,"passed":true,"public_test":{"exit_code":0,"output_bytes":0,"status":"pass"},"status":"passed"},{"artifact_bytes":81,"artifact_sha256":"488fd9a9f85fb1ecc546b18649b4fa82f6e621848647f72ac5c0934271fb8f01","generation":{"exit_code":0,"output_bytes":0,"status":"pass"},"hidden_test":{"exit_code":1,"output_bytes":0,"status":"fail"},"index":2,"passed":false,"public_test":{"exit_code":0,"output_bytes":0,"status":"pass"},"status":"hidden_test_failed"}],"cleanup_verified":true,"fixture_mode":true,"limits":{"command_timeout":5.0,"max_candidate_bytes":8192,"max_output_bytes":8192,"max_total_candidate_bytes":24576,"max_wall_seconds":20.0},"sample_count":3,"selected_index":1,"selection":"REVIEW_RECOMMENDATION","target_commit":"57f17b00804f4f89a6976a8f68b2dcef5763dfcf","target_digest_after":"e4b8a384f5a06774c1b7140c4fb95c1820047ade5802921b60c322b369e6979d","target_digest_before":"e4b8a384f5a06774c1b7140c4fb95c1820047ade5802921b60c322b369e6979d","target_unchanged":true}

To prove no-pass safety, run a one-candidate fixture. Candidate 0 fails sealed tests, process exits 2, selection is NO_SELECTION, and target hash stays unchanged:

python3 runner.py --target target --target-commit "$target_commit" --architecture-selection architecture-selection.json --review-output review-no-pass --demo --samples 1 --workers 1 --budget-units 1 || test "$?" -eq 2
sha256sum -c before.sha256
test ! -e review-no-pass/selected/pricing.py

Prove allowlist denial separately. Adapter edits pricing.py plus public_test.py; second path is out of scope, so process exits 2 before tests and receipt status is out_of_scope_change:

python3 runner.py \
  --target target \
  --target-commit "$target_commit" \
  --architecture-selection architecture-selection.json \
  --review-output review-denied \
  --samples 1 \
  --workers 1 \
  --budget-units 1 \
  --model-cmd "python3 model_adapter.py" || test "$?" -eq 2
python3 -c 'import json; assert json.load(open("review-denied/receipt.json"))["candidates"][0]["status"] == "out_of_scope_change"'
sha256sum -c before.sha256

For a real model adapter, write model_adapter.py so it reads TTC_PROMPT and writes generated bytes to TTC_CANDIDATE, then replace --demo with:

python3 runner.py \
  --target target \
  --target-commit "$target_commit" \
  --architecture-selection architecture-selection.json \
  --review-output review-real \
  --samples 3 \
  --workers 3 \
  --budget-units 3 \
  --model-cmd "python3 model_adapter.py"

The program never invokes an implicit shell, never writes selected code into target, rejects non-empty review output, and runs baseline plus candidate tests in disposable copies. An operator can still explicitly configure --model-cmd "sh -c ...", so command policy must forbid that form when shell access is outside scope. It checks artifact size before reading, rejects selection when captured output or cumulative candidate bytes exceed acceptance limits, rejects background descendants after terminating their process group, and fails closed if target digest or Git status changes. Because communicate() captures output before applying its acceptance limit, this teaching harness is not a memory-safe streaming sandbox; untrusted workloads still require the sandbox back-routes above. It excludes sealed_test.py from candidate copies and invokes the trusted source file with the candidate workspace as current directory. Fixture mode is capped at three built-in candidates and remains local evidence, not hostile-code isolation.

Finish with scoped cleanup only after saving receipt evidence elsewhere:

cd /
if [[ "$fixture" != /tmp/shipright-ttc-fixture.* || ! -d "$fixture/target/.git" ]]; then
  exit 1
fi
rm -rf -- "$fixture"
test ! -e "$fixture"

Receipt is evidence about one local run, not proof of production safety. Normal path selects candidate 1 because it alone passes both suites. Failure path selects nothing. Recovery requires stronger tests or another bounded run, never weakening hidden assertions. Cleanup proof confirms temporary directories vanished. Matching target hashes prove no mutation in this fixture; any detected digest or Git-status change yields TARGET_CHANGED, exit 2, and no selected artifact, but the runner does not undo an external or hostile mutation.

Operator evidence should add actor, repository commit, issue ID, environment, timestamp, command policy version, model/version, test artifact hashes, and immutable CI run ID. Human review then decides whether to open a pull request. Merge remains outside this runner.

Assessment receipt: given one unknown—no candidate passes sealed tests—the accepted selection is NO_SELECTION; evidence is candidate exits, misconception is choosing the least-bad patch, and remediation is replaying sealed checks without exposing them to generation.

You can now allocate inference compute, measure width, design verification, adapt width versus depth, compare composed architectures, and run a review-only coding search with bounded evidence. Final handoff is review/receipt.json plus an optional review/selected/pricing.py; the runner writes no candidate into the local target, and CI plus a human reviewer accept the receipt only when target_unchanged=true.