Harbor review and repair: complete prompt reference¶

"""Evidence and bounded decisions for the practical-generation review profile."""

from __future__ import annotations

from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

from repo2rlenv.spec.input import LLMSpec


class Record(BaseModel):
    model_config = ConfigDict(extra="forbid")


ProbeFocus = Literal[
    "general", "lazy_output", "numeric_tolerance", "compiled_execution", "model_behavior"
]


class Citation(Record):
    path: str
    quote: str = Field(min_length=1, max_length=1000)


class Assessment(Record):
    status: Literal["pass", "fail", "unknown"]
    score: int = Field(ge=0, le=4)
    explanation: str
    evidence: list[Citation] = Field(min_length=1)


class Issue(Record):
    category: Literal["instruction", "verifier", "leakage", "reference", "packaging", "probe"]
    severity: Literal["blocking", "improvement"]
    problem: str
    repair: str
    evidence: list[Citation] = Field(min_length=1)


class ReadRequest(Record):
    path: str
    query: str | None = Field(
        description="One exact literal substring to search, not a regex or a list of words. Use separate read requests for different terms."
    )
    start_line: int = Field(ge=1)
    end_line: int = Field(ge=1)

    @model_validator(mode="after")
    def bounded_range(self):
        if not 0 <= self.end_line - self.start_line < 400:
            raise ValueError("Read at most 400 lines in forward order")
        return self


class SemanticProbe(Record):
    name: str = Field(pattern=r"^[a-z][a-z0-9-]{0,35}$")
    kind: Literal["wrong_solution", "valid_alternative"]
    focus: ProbeFocus = "general"
    rationale: str = Field(min_length=1)
    evidence: list[Citation] = Field(min_length=1)
    # Run after the reference in a private probe variant, never on the controller.
    script: str = Field(min_length=1, max_length=16000)


class ProbeManifest(Record):
    bundle_hash: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
    probes: list[SemanticProbe] = Field(min_length=1, max_length=4)


class Review(Record):
    summary: str
    task: Assessment
    verifier: Assessment
    leakage: Assessment
    rollout: Literal[
        "not_run",
        "legitimate_success",
        "legitimate_failure",
        "reward_hack",
        "task_defect",
        "infrastructure_failure",
        "incomplete",
        "insufficient_evidence",
    ]
    issues: list[Issue] = Field(max_length=12)
    probes: list[SemanticProbe] = Field(max_length=4)
    read_requests: list[ReadRequest] = Field(max_length=6)

    @property
    def sound(self) -> bool:
        return (
            not self.read_requests
            and all(
                getattr(self, name).status == "pass" for name in ("task", "verifier", "leakage")
            )
            and not any(issue.severity == "blocking" for issue in self.issues)
            and self.rollout not in {"reward_hack", "task_defect", "insufficient_evidence"}
        )


class Edit(Record):
    path: str
    old: str
    new: str
    executable: bool


class Repair(Record):
    explanation: str = Field(max_length=1500)
    addressed_issues: list[str] = Field(min_length=1)
    edits: list[Edit] = Field(max_length=16)
    append_expected_passes: list[str] = Field(
        default_factory=list,
        max_length=256,
        description=(
            "Append only new exact case IDs to existing tests/contract.json.expected_passes; "
            "preserve all existing IDs and do not also text-edit that contract."
        ),
    )
    probe_replacements: list[SemanticProbe] = Field(default_factory=list, max_length=4)

    @model_validator(mode="after")
    def targeted_changes(self):
        if not self.edits and not self.append_expected_passes and not self.probe_replacements:
            raise ValueError("A repair must change task files or a diagnosed invalid probe")
        if any(not value.strip() for value in self.append_expected_passes):
            raise ValueError("Appended expected-pass IDs must be nonempty")
        if len(set(self.append_expected_passes)) != len(self.append_expected_passes):
            raise ValueError("Duplicate appended expected-pass IDs")
        # Replacement eligibility depends on preserved execution evidence and is
        # enforced by QualityLoop, never inferred from a model-authored patch.
        return self


class LoopOptions(Record):
    review_model: LLMSpec = Field(
        default_factory=lambda: LLMSpec(provider="anthropic", model="claude-sonnet-4-6")
    )
    repair_model: LLMSpec | None = None
    escalation_model: LLMSpec | None = None
    solver_model: LLMSpec = Field(
        default_factory=lambda: LLMSpec(provider="anthropic", model="claude-sonnet-4-6")
    )
    repair: bool = False
    run_rollout: bool = False
    max_repairs: int = Field(default=3, ge=0, le=5)
    max_read_rounds: int = Field(default=2, ge=0, le=4)
    max_probes: int = Field(default=2, ge=0, le=4)
    context_chars: int = Field(default=100000, ge=16000, le=250000)
    model_tokens: int = Field(default=6000, ge=1024, le=16000)
    max_turns: int = Field(default=24, ge=1, le=100)
    solver_tokens: int = Field(default=4096, ge=256, le=8192)
    trial_timeout_sec: int = Field(default=900, ge=30, le=3600)
    success_reward: float = Field(default=1.0, allow_inf_nan=False)
    model_reservation_usd: str = "1.00"
    solver_reservation_usd: str = "4.00"
    shared_reservation_wait_sec: int = Field(default=30, ge=0, le=300, strict=True)
    max_spend_usd: str = "15.00"

    @model_validator(mode="after")
    def explicit_routes_and_limits(self):
        from decimal import Decimal

        for name in ("model_reservation_usd", "solver_reservation_usd", "max_spend_usd"):
            amount = Decimal(getattr(self, name))
            if not amount.is_finite() or amount <= 0:
                raise ValueError(f"{name} must be finite and positive")
        for spec in (
            self.review_model,
            self.repair_model,
            self.escalation_model,
            self.solver_model,
        ):
            if spec and (spec.provider not in {"openai", "anthropic"} or spec.fallback):
                raise ValueError(
                    "Use explicit OpenAI/Anthropic routes; escalation is metered separately"
                )
        return self


class TrialRecord(Record):
    role: Literal["baseline", "oracle", "rollout", "probe"]
    bundle_hash: str
    result: str
    result_sha256: str
    agent: str
    model: str | None
    reward: float | None = Field(allow_inf_nan=False, strict=True)
    exception_type: str | None
    # Imported, checksum-bound evidence is valid but not cryptographically attested.
    binding: Literal["receipt", "harbor_checksum"]
    agent_exit_code: int | None = None
    probe_installed: bool | None = None
    probe: SemanticProbe | None = None


class LoopResult(Record):
    schema_version: Literal["1"] = "1"
    profile: Literal["practical-generation-v1"] = "practical-generation-v1"
    status: Literal["usable", "reviewed", "needs_repair", "needs_evidence", "budget_exhausted"]
    source_hash: str
    bundle_hash: str
    task_path: str
    repairs: int
    review: Review | None
    trials: list[TrialRecord]
    reasons: list[str]
    accounted_usd: str
    reserved_usd: str

"""Bounded evidence packs, explicit omissions and verifiable model citations."""

from __future__ import annotations

import ast
import hashlib
import json
import re
from pathlib import Path

from repo2rlenv.quality.loop.artifacts import digest
from repo2rlenv.quality.loop.models import ReadRequest, Review, TrialRecord
from repo2rlenv.quality.loop.protocol import citation_path_error
from repo2rlenv.quality.loop.rollout_evidence import rollout_documents


def _search_excerpts(lines: list[str], query: str, *, maximum: int | None = None) -> str:
    """Bound both match count and characters, including minified JSON traces."""
    excerpts = []
    ranges = []
    matched_windows = 0
    for index, line in enumerate(lines):
        match = line.find(query)
        if match < 0:
            continue
        start, end = max(0, index - 5), min(len(lines), index + 26)
        surrounding = "".join(lines[start:end])
        if len(surrounding) <= 3000:
            previous = ranges[-1] if ranges else None
            merged = "".join(lines[previous[0] : end]) if previous else ""
            if previous and start <= previous[1] and len(merged) <= 3000:
                start = previous[0]
                ranges[-1] = (start, end)
                excerpts[-1] = f"[Lines {start + 1}-{end}]\n" + merged
            else:
                excerpts.append(f"[Lines {start + 1}-{end}]\n" + surrounding)
                ranges.append((start, end))
            matched_windows += 1
        else:
            # One line can contain an entire trajectory. Retain literal bytes
            # around each hit and label omissions instead of expanding that line.
            while match >= 0 and matched_windows < 8:
                left = max(0, match - 1000)
                right = min(len(line), left + 3000)
                excerpts.append(
                    f"[Line {index + 1}, columns {left + 1}-{right}; "
                    "surrounding text omitted]\n" + line[left:right]
                )
                ranges.append(None)
                matched_windows += 1
                match = line.find(query, right)
        if matched_windows >= 8:
            break
    note = (
        "\n[Search limited to the first 8 matching windows; use a narrower literal query "
        "or line ranges to inspect further matches.]"
        if matched_windows >= 8
        else ""
    )
    result = ("\n".join(excerpts) or "[No literal matches found]") + note
    if maximum is None or len(result) <= maximum:
        return result
    # Keep complete literal windows. Existing documents and accepted quotations
    # are never shortened to make room; omissions in this new search are explicit.
    for count in range(len(excerpts) - 1, 0, -1):
        bounded = (
            "\n".join(excerpts[:count])
            + (
                f"\n[Context budget: {len(excerpts) - count} additional search window(s) omitted. "
                "Use a narrower literal query or explicit line ranges to inspect them.]"
            )
            + note
        )
        if len(bounded) <= maximum:
            return bounded
    raise ValueError(
        f"Additional reads exceeded context budget: {maximum} characters remain, "
        "insufficient for one complete search window. Request a narrower literal "
        "query or a short explicit line range; existing evidence is unchanged."
    )


class EvidenceContext:
    def __init__(self, task: Path, trials: list[TrialRecord], *, limit: int):
        self.task, self.limit = task, limit
        self.documents: dict[str, str] = {}
        self.inventory: list[dict] = []
        self.omitted: list[str] = []
        self._paths: dict[str, Path] = {}
        self._texts: dict[str, str] = {}
        self._initial_limit = int(limit * (0.35 if trials else 0.7))
        for path in sorted(task.rglob("*")):
            if path.is_file():
                key = path.relative_to(task).as_posix()
                self._paths[key] = path
                self.inventory.append(
                    {"path": key, "bytes": path.stat().st_size, "sha256": digest(path)}
                )
        priority = [
            "instruction.md",
            "task.toml",
            "environment/Dockerfile",
            "solution/solve.sh",
            "tests/test.sh",
            "tests/contract.json",
        ]
        for key in priority:
            if key in self._paths:
                self._include(key, self._paths[key], maximum=12000, tail=key.endswith(".json"))
        instruction = (task / "instruction.md").read_text()
        symbols = set(re.findall(r"\bdef\s+([A-Za-z_]\w*)", instruction))
        symbols.update(re.findall(r"`(?:[A-Za-z_]\w*\.)*([A-Za-z_]\w*)\s*(?:\(|`)", instruction))
        test_symbols: dict[str, set[str]] = {}
        whole_test_files: set[str] = set()
        contract = self._paths.get("tests/contract.json")
        if contract is not None:
            try:
                data = json.loads(contract.read_text())
                methods = {
                    identity.rsplit("::", 1)[-1].split("[", 1)[0]
                    for identity in data.get("expected_passes", [])[:64]
                }
                for selector in data.get("test_paths", []):
                    path, *nodes = selector.split("::")
                    selected = {node.split("[", 1)[0] for node in nodes} or methods
                    test_symbols.setdefault("tests/source/" + path, set()).update(selected)
                    if not nodes and path.endswith(".py"):
                        whole_test_files.add("tests/source/" + path)
                for key in self._paths:
                    if (
                        key in test_symbols
                        or not key.startswith("tests/source/")
                        or not key.endswith(".py")
                    ):
                        continue
                    module = key.removeprefix("tests/source/")[:-3].replace("/", ".")
                    nodes = set()
                    for identity in data.get("expected_passes", [])[:64]:
                        if identity.startswith((module + ".", module + "::")):
                            nodes.update(identity[len(module) :].lstrip(".:").split("::"))
                    if nodes:
                        test_symbols[key] = nodes
            except (ValueError, AttributeError):
                pass
        self.verifier_paths = sorted(key for key in test_symbols if key in self._paths)
        # Selected assertions precede reference patches and generic grading
        # helpers, which can otherwise fill the entire task share. Every file
        # remains in the inventory, even when its initial excerpt does not fit.
        # Generic method names such as test_empty must never select unrelated
        # source functions/classes.
        helpers = {
            key
            for key in self._paths
            if key.startswith(("tests/", "solution/")) and key.count("/") == 1
        }
        ordered = sorted(
            self._paths.items(),
            key=lambda item: (
                0
                if item[0] in test_symbols
                else 1
                if item[0] in helpers
                else 2
                if item[0].startswith(("environment/", "solution/"))
                else 3,
                item[0],
            ),
        )
        for key, path in ordered:
            if key in self.documents:
                continue
            if key in helpers:
                self._include(key, path, maximum=12000, tail=key.endswith(".json"))
                continue
            if path.suffix == ".py" and path.stat().st_size < 2_000_000:
                if key in whole_test_files and path.stat().st_size <= 24000:
                    # Small selected suites include their fixture helpers and
                    # final assertions, avoiding needless reads of omitted tests.
                    # Larger or individually selected suites retain symbol excerpts.
                    self._include(key, path, maximum=24000)
                else:
                    self._include_symbols(key, path, test_symbols.get(key, symbols))
        # Reserve an execution share before reading large repository files. Collect
        # every trial first: a baseline's long test inventory must not crowd out a
        # later counterexample's actual assertion failure.
        candidates = []
        for index, trial in enumerate(trials):
            path = Path(trial.result)
            if digest(path) != trial.result_sha256:
                raise ValueError("Trial result changed since ingestion")
            prefix = f"evidence/{index}-{trial.role}/"
            summary = trial.model_dump(mode="json")
            native = json.loads(path.read_text())
            message = (native.get("exception_info") or {}).get("exception_message")
            if message:
                summary["exception_message_tail"] = str(message)[-6000:]

            if trial.probe:
                script_key = prefix + "probe-script.sh"
                script = summary["probe"].pop("script")
                summary["probe"]["script_path"] = script_key
                self._texts[script_key] = script
                self.inventory.append(
                    {
                        "path": script_key,
                        "bytes": len(script.encode()),
                        "sha256": hashlib.sha256(script.encode()).hexdigest(),
                    }
                )
                candidates.append((2, index, script_key, 8000))
            self.documents[prefix + "result.json"] = json.dumps(summary, indent=2)
            root = path.parent
            if trial.role == "rollout":
                for name, text in rollout_documents(task, root).items():
                    key = prefix + name
                    self._texts[key] = text
                    self.inventory.append(
                        {
                            "path": key,
                            "bytes": len(text.encode()),
                            "sha256": hashlib.sha256(text.encode()).hexdigest(),
                        }
                    )
                    candidates.append((1, index, key, 18000))
            # Failure logs precede trajectories, verbose test inventories and
            # captured source. All remain addressable through bounded read requests.
            selected = [
                (0, "verifier/stdout.txt", 6000),
                (0, "verifier/stderr.txt", 3000),
                (0, "exception.txt", 3000),
                (1, "verifier/test-stdout.txt", 4000),
                (1, "verifier/test-stderr.txt", 3000),
                (1, "agent/exit-code.txt", 100),
                (3, "agent/oracle.txt", 4000),
                (3, "agent/trajectory.json", 12000),
                (4, "verifier/result.json", 3000),
                (4, "verifier/results.xml", 4000),
            ]
            selected += (
                [
                    (5, item.relative_to(root).as_posix(), 3000)
                    for item in sorted((root / "artifacts").rglob("*"))
                ]
                if (root / "artifacts").is_dir()
                else []
            )
            for priority, relative, maximum in selected:
                candidate = root / relative
                if candidate.is_symlink() or not candidate.resolve().is_relative_to(root.resolve()):
                    raise ValueError("Trial evidence cannot contain symlinks")
                if candidate.is_file():
                    key = prefix + relative
                    self._paths[key] = candidate
                    self.inventory.append(
                        {
                            "path": key,
                            "bytes": candidate.stat().st_size,
                            "sha256": digest(candidate),
                        }
                    )
                    # Expected baseline failures are less useful for diagnosis than
                    # oracle/probe/solver failures on the same revision.
                    candidates.append(
                        (priority + (6 if trial.role == "baseline" else 0), index, key, maximum)
                    )
        self._initial_limit = int(limit * 0.7)
        for _, _, key, maximum in sorted(candidates):
            if key in self._texts:
                self._include_text(key, self._texts[key], maximum=maximum, tail=True)
            else:
                self._include(key, self._paths[key], maximum=maximum, tail=True)

    def _include(self, key: str, path: Path, *, maximum: int, tail: bool = False):
        if path.stat().st_size > 2_000_000:
            self.omitted.append(key + ": larger than 2 MB")
            return
        try:
            text = path.read_text()
        except UnicodeError:
            self.omitted.append(key + ": binary")
            return
        self._include_text(key, text, maximum=maximum, tail=tail)

    def _include_text(self, key: str, text: str, *, maximum: int, tail: bool):
        # Leave room for model-requested source excerpts instead of filling the
        # initial prompt with unrelated repository documentation and test names.
        remaining = self._initial_limit - sum(len(value) for value in self.documents.values())
        maximum = min(maximum, remaining)
        if maximum < 100:
            self.omitted.append(key + ": context budget")
            return
        if len(text) > maximum:
            self.omitted.append(key + ": partial text; request line ranges if needed")
            text = (
                (text[: maximum // 2] + "\n[... omitted ...]\n" + text[-maximum // 2 :])
                if tail
                else text[:maximum]
            )
        self.documents[key] = text

    def _include_symbols(self, key: str, path: Path, symbols: set[str]):
        if not symbols:
            return
        try:
            text = path.read_text()
            tree = ast.parse(text)
        except (UnicodeError, SyntaxError):
            return
        lines = text.splitlines(keepends=True)
        chunks = []
        covered = []
        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and any(
                symbol.lower().replace("_", "") == node.name.lower().replace("_", "")
                or (
                    isinstance(node, ast.ClassDef)
                    and node.name.endswith("Tests")
                    and symbol.lower().replace("_", "") == node.name[:-5].lower().replace("_", "")
                )
                for symbol in symbols
                if len(symbol) >= 3
            ):
                if any(start <= node.lineno <= end for start, end in covered):
                    continue
                start = max(1, node.lineno - 1)
                end = min(node.end_lineno or node.lineno, start + 399)
                covered.append((start, end))
                chunks.append(f"[Lines {start}-{end}]\n" + "".join(lines[start - 1 : end]))
                if sum(map(len, chunks)) >= 12000:
                    break
        if chunks:
            # Repairs need the module's real imports and aliases as well as the
            # selected function/class. Keep the header bounded and explicit.
            header_end = min(60, len(lines))
            header = "".join(lines[:header_end])[:4000]
            chunks.insert(
                0, f"[Module header, lines 1-{header_end}, at most 4000 chars]\n" + header
            )
        excerpt = "\n".join(chunks)[:16000]
        if excerpt and sum(map(len, self.documents.values())) + len(excerpt) <= self._initial_limit:
            self.documents[key] = excerpt
            self.omitted.append(
                key + ": symbol excerpts only; request ranges/search for other code"
            )

    def add_text(self, key: str, text: str, *, maximum: int):
        """Register controller context with bounded excerpts and a readable full source."""
        if key in self._paths or key in self._texts or key in self.documents:
            raise ValueError("Evidence document already exists")
        self._texts[key] = text
        self.inventory.append(
            {
                "path": key,
                "bytes": len(text.encode()),
                "sha256": hashlib.sha256(text.encode()).hexdigest(),
            }
        )
        available = max(0, min(maximum, self.limit - sum(map(len, self.documents.values()))))
        marker = (
            "\n[Excerpt ends; request a range or literal search in this document for omitted text.]"
        )
        if len(text) <= available:
            self.documents[key] = text
        else:
            if available > len(marker):
                self.documents[key] = text[: available - len(marker)] + marker
            self.omitted.append(
                key + ": controller context excerpt; full text available through reads"
            )

    def read_more(self, requests: list[ReadRequest]):
        additions = {}
        for request in requests:
            if request.path not in self._paths and request.path not in self._texts:
                matches = sorted(
                    (
                        key
                        for key in self._paths.keys() | self._texts.keys()
                        if Path(key).name == Path(request.path).name
                    ),
                    key=lambda key: (key not in self.verifier_paths, key),
                )
                guidance = (
                    " Exact basename matches: " + ", ".join(matches[:8]) + "."
                    if matches
                    else " Use a path from the supplied inventory or search its catalogue."
                )
                raise ValueError(
                    f"Requested file is not in evidence inventory: {request.path}." + guidance
                )
            if request.path in self._texts:
                text = self._texts[request.path]
            else:
                path = self._paths[request.path]
                if path.stat().st_size > 2_000_000:
                    raise ValueError("Requested file exceeds bounded text reader")
                text = path.read_text()
            lines = text.splitlines(keepends=True)
            suffix = (
                f"search={request.query}"
                if request.query is not None
                else f"L{request.start_line}-L{request.end_line}"
            )
            key = f"{request.path}:{suffix}"
            if key in self.documents or key in additions:
                continue
            remaining = (
                self.limit
                - sum(map(len, self.documents.values()))
                - sum(map(len, additions.values()))
            )
            if request.query is not None:
                if not request.query.strip() or len(request.query) > 200:
                    raise ValueError("Search query must have 1-200 characters")
                try:
                    text = _search_excerpts(lines, request.query, maximum=remaining)
                except ValueError as exc:
                    raise ValueError(f"{request.path}: {exc}") from exc
            else:
                text = "".join(lines[request.start_line - 1 : request.end_line])
            if not text:
                raise ValueError("Requested range is empty")
            if len(text) > remaining:
                raise ValueError(
                    f"Additional reads exceeded context budget: {key} requires {len(text)} "
                    f"characters but {remaining} remain. Request a shorter line range or "
                    "literal search; existing evidence is unchanged."
                )
            additions[key] = text
        updated = {**self.documents, **additions}
        if sum(map(len, updated.values())) > self.limit:
            raise ValueError("Additional reads exceeded context budget")
        self.documents = updated

    def _inventory_payload(self) -> dict:
        entries = [{"path": item["path"], "bytes": item["bytes"]} for item in self.inventory]
        if len(json.dumps(entries)) <= 30000:
            return {"inventory": entries}
        # Trial captures repeat long directory prefixes for hundreds of modules.
        # Group them losslessly; all full paths remain available to read_more.
        directories: dict[str, dict[str, int]] = {}
        for item in entries:
            prefix, _, name = item["path"].rpartition("/")
            directories.setdefault(prefix, {})[name] = item["bytes"]
        if len(json.dumps(directories)) > 30000:
            # Large repositories still need a bounded first review. Keep every
            # path addressable through a searchable catalogue instead of dropping
            # files or refusing the review before the model can request evidence.
            key = "evidence/full-file-inventory.jsonl"
            if key in self._paths:
                raise ValueError("Task uses the reserved quality inventory path")
            self._texts[key] = "\n".join(json.dumps(item) for item in entries) + "\n"
            counts = [
                {"directory": name, "files": len(items)} for name, items in directories.items()
            ]
            return {
                "inventory_document": key,
                "inventory_files": len(entries),
                "inventory_directories": counts[:100],
                "inventory_directories_omitted": max(0, len(counts) - 100),
                "inventory_format": (
                    "The complete JSONL catalogue is available through read requests: "
                    "one file path and byte count per line. Search it by filename or "
                    "request bounded line ranges, then read the required source file. "
                    "All original file paths remain directly readable."
                ),
            }
        return {
            "inventory_by_directory": directories,
            "inventory_format": (
                "Each directory maps filenames to byte counts. Read a file using "
                "directory/filename (or filename for the empty directory). "
                "This is the complete inventory, without omitted paths."
            ),
        }

    def payload(self, **extra) -> str:
        result = json.dumps(
            {
                "documents": self.documents,
                "selected_verifier_files": self.verifier_paths[:32],
                "selected_verifier_files_omitted": max(0, len(self.verifier_paths) - 32),
                # Hashes remain in the local evidence record. They add no useful
                # review context and repeat for every source copy and trial.
                **self._inventory_payload(),
                "omitted": [item for item in self.omitted if not item.endswith(": context budget")],
                "budget_omissions": {
                    "count": sum(item.endswith(": context budget") for item in self.omitted),
                    "note": "Inventory files absent from documents remain available through bounded reads.",
                },
                **extra,
            },
            ensure_ascii=False,
        )
        # Inventory and decision metadata have their own small allowance.
        if len(result) > self.limit + 70000:
            raise ValueError("Evidence pack exceeds context and inventory limits")
        return result

    def validate_review(self, review: Review):
        citations = [
            citation
            for name in ("task", "verifier", "leakage")
            for citation in getattr(review, name).evidence
        ]
        citations += [
            citation for item in [*review.issues, *review.probes] for citation in item.evidence
        ]
        for citation in citations:
            document = self.documents.get(citation.path, "")
            if not document or " ".join(citation.quote.split()) not in " ".join(document.split()):
                raise ValueError(citation_path_error(citation, self.documents))

"""Evidence-bound correction of controls that never completed installation."""

from __future__ import annotations

import json
from pathlib import Path

from repo2rlenv.execution.lifecycle import save_record
from repo2rlenv.quality.loop.artifacts import digest, import_trial, task_identity
from repo2rlenv.quality.loop.models import SemanticProbe, TrialRecord


def _files(trial: TrialRecord) -> dict[str, str | None]:
    result = Path(trial.result)
    paths = [
        result,
        result.parents[2] / "trial.json",
        result.parent / "agent/oracle.txt",
        result.parent / "agent/exit-code.txt",
    ]
    hashes = {}
    for path in paths:
        if any(component.is_symlink() for component in (path, *path.parents)) or (
            path.exists() and not path.is_file()
        ):
            raise ValueError("Probe installation evidence must be regular files")
        hashes[str(path)] = digest(path) if path.exists() else None
    return hashes


def capture_attempt(parent_hash: str, trial: TrialRecord) -> dict:
    from repo2rlenv.quality.loop.probe_behavior import BEHAVIOR_FOCI, behavior_files

    attempt = {
        "parent_hash": parent_hash,
        "trial": trial.model_dump(mode="json"),
        "files": _files(trial),
    }
    if trial.probe is not None and trial.probe.focus in BEHAVIOR_FOCI:
        attempt["behavior_files"] = behavior_files(trial)
    return attempt


def record_attempt(directory: Path, key: str, parent_hash: str, trial: TrialRecord) -> dict:
    """Keep prior installations even after current-revision trials are discarded."""
    attempt = capture_attempt(parent_hash, trial)
    path = directory / "probe-attempts" / f"{key}.json"
    if path.exists():
        if path.is_symlink():
            raise ValueError("Preserved probe installation evidence changed")
        saved = json.loads(path.read_text())
        # Older journals retain their original evidence boundary. Do not seal
        # previously unbound verifier logs retroactively on resume.
        compared = (
            attempt
            if "behavior_files" in saved
            else {key: value for key, value in attempt.items() if key != "behavior_files"}
        )
        if saved != compared:
            raise ValueError("Preserved probe installation evidence changed")
        return saved
    else:
        save_record(path, attempt)
    return attempt


def installation(attempt: dict) -> bool:
    """Reconstruct completion from the owned receipt, exact variant and saved logs.

    An absent receipt/log or uncertain exit is not proof of a failed installation.
    These are content bindings, not attestations for externally supplied evidence.
    """
    trial = TrialRecord.model_validate(attempt["trial"])
    if trial.role != "probe" or trial.probe is None or trial.binding != "receipt":
        raise ValueError("Probe correction requires an owned probe execution receipt")
    if attempt["files"] != _files(trial) or any(
        value is None for value in attempt["files"].values()
    ):
        raise ValueError("Probe installation evidence is missing or changed")
    result = Path(trial.result)
    if digest(result) != trial.result_sha256:
        raise ValueError("Probe result changed after collection")
    raw = json.loads(result.read_text())
    variant = Path(raw.get("config", {}).get("task", {}).get("path", ""))
    if not variant.is_dir():
        variant = result.parents[4] / "probes" / result.parents[2].name / variant.name
    manifest_path = variant.parent / "probe.json"
    if (
        any(component.is_symlink() for component in (manifest_path, *manifest_path.parents))
        or not manifest_path.is_file()
    ):
        raise ValueError("Probe creation receipt is unavailable")
    manifest = json.loads(manifest_path.read_text())
    if (
        manifest
        != {
            "parent_hash": attempt["parent_hash"],
            "bundle_hash": trial.bundle_hash,
            "probe": trial.probe.model_dump(mode="json"),
        }
        or task_identity(variant) != trial.bundle_hash
    ):
        raise ValueError("Probe installation belongs to another definition or parent")
    imported = import_trial(result.parents[2] / "trial.json", variant, "oracle")
    expected = trial.model_copy(update={"role": "oracle", "probe": None, "probe_installed": None})
    if imported != expected:
        raise ValueError("Probe summary disagrees with its execution evidence")
    log = result.parent / "agent/oracle.txt"
    if log.stat().st_size > 16 * 1024 * 1024:
        raise ValueError("Probe installation log exceeds the evidence limit")
    completed = "__QUALITY_PROBE_COMPLETED__" in log.read_text()
    if completed != trial.probe_installed or trial.agent_exit_code is None:
        raise ValueError("Probe installation summary disagrees with its completion evidence")
    if not completed and trial.agent_exit_code == 0:
        raise ValueError("Missing marker without a failed agent exit is inconclusive")
    return completed


def failed_installations(task: Path, trials: list[TrialRecord]) -> list[dict]:
    """Only expose independently reconstructed failures to the model."""
    failures = []
    parent_hash = task_identity(task)
    for index, trial in enumerate(trials):
        if (
            trial.role != "probe"
            or trial.probe is None
            or trial.probe.kind != "wrong_solution"
            or trial.probe_installed is not False
            or trial.agent_exit_code in {None, 0}
        ):
            continue
        try:
            attempt = capture_attempt(parent_hash, trial)
            if installation(attempt):
                continue
        except (ValueError, OSError, KeyError, IndexError):
            continue
        failures.append(
            {
                "name": trial.probe.name,
                "result": trial.result,
                "result_sha256": trial.result_sha256,
                "agent_exit_code": trial.agent_exit_code,
                "summary_path": f"evidence/{index}-probe/result.json",
                "log_path": f"evidence/{index}-probe/agent/oracle.txt",
                "diagnosis": "Probe installation did not complete; reward is not a semantic control result.",
            }
        )
    return failures


def grounded_diagnosis(failure: dict, review, context) -> bool:
    """Require a probe-category issue citing this attempt, not an unrelated defect."""
    for issue in review.issues:
        if issue.category != "probe":
            continue
        for citation in issue.evidence:
            if citation.path not in {
                failure["summary_path"],
                failure["log_path"],
                *failure.get("evidence_paths", []),
            }:
                continue
            document = context.documents.get(citation.path, "")
            if document and " ".join(citation.quote.split()) in " ".join(document.split()):
                return True
    return False


def replacement_evidence(
    task: Path, probe: SemanticProbe, history: list[dict], failures: list[dict], review, context
) -> dict | None:
    """A completed installation under this name permanently closes this exception."""
    if probe.kind != "wrong_solution" or not any(
        failure["name"] == probe.name for failure in failures
    ):
        return None
    attempts = [
        attempt
        for attempt in history
        if (attempt["trial"].get("probe") or {}).get("name") == probe.name
    ]
    if not attempts:
        return None
    try:
        if any(installation(attempt) for attempt in attempts):
            return None
    except (ValueError, OSError, KeyError, IndexError):
        return None
    exact = [
        attempt
        for attempt in attempts
        if attempt["parent_hash"] == task_identity(task)
        and attempt["trial"]["probe"] == probe.model_dump(mode="json")
    ]
    for failure in failures:
        if (
            failure["name"] == probe.name
            and any(attempt["trial"]["result"] == failure["result"] for attempt in exact)
            and grounded_diagnosis(failure, review, context)
        ):
            return {"probe": probe.model_dump(mode="json"), "attempts": attempts}
    return None