Let repair and review agents own the fix workflow

strix/fix/contracts.py

@@ -10,6 +10,8 @@ from typing import TYPE_CHECKING, Literal, cast
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
+from strix.config.settings import DEFAULT_MAX_TURNS
+
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -133,7 +135,7 @@ class CommandSpec(ContractModel):
     required: bool = True
     timeout_seconds: int = Field(default=300, ge=1, le=3600)
     cwd: str = "."
-    purpose: Literal["quality", "regression", "unit"] = "quality"
+    purpose: Literal["quality", "regression", "unit", "security"] = "quality"
 
     _relative_cwd = field_validator("cwd")(_validate_relative_path)
 
@@ -151,7 +153,7 @@ class ReproductionSpec(ContractModel):
 
 
 class RepositoryTestPlan(ContractModel):
-    """The repair agent's native tests, rerun by the controller before review."""
+    """Historical native-test handoff; agent-driven runs record commands directly."""
 
     regression_test: CommandSpec
     regression_files: list[str] = Field(min_length=1)
@@ -218,8 +220,11 @@ class FixPreparationRequestV1(ContractModel):
     repository_id: str | None = None
     candidate: FixCandidateV1
     checks: list[CommandSpec] = []
-    max_repair_attempts: int = Field(default=2, ge=1, le=2)
-    timeout_seconds: int = Field(default=1800, ge=30, le=14400)
+    # Accepted for old callers; the agent loop is bounded by turns/time instead.
+    max_repair_attempts: int | None = Field(default=None, ge=1)
+    max_agent_turns: int = Field(default=DEFAULT_MAX_TURNS, ge=1, le=10000)
+    timeout_seconds: int = Field(default=7200, ge=30, le=14400)
+    max_budget_usd: float | None = Field(default=None, gt=0, allow_inf_nan=False)
     network_allowed: bool = False
     credentials_allowed: list[str] = []
 
@@ -244,7 +249,7 @@ class CheckResult(ContractModel):
     baseline_workspace_root: str | None = None
     source_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
     environment_id: str | None = None
-    purpose: Literal["quality", "regression", "unit"] = "quality"
+    purpose: Literal["quality", "regression", "unit", "security"] = "quality"
     tests_passed: int | None = Field(default=None, ge=0)
 
 
@@ -316,6 +321,8 @@ class VerifierResult(ContractModel):
     unit_test_coverage_valid: bool = False
     # None preserves historical reviews; new reviewers classify every concern.
     concerns: list[ReviewConcern] | None = None
+    source_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+    turns_used: int = Field(default=0, ge=0)
 
 
 class RepairOutcome(ContractModel):
@@ -328,6 +335,7 @@ class RepairOutcome(ContractModel):
     checks: list[CommandSpec] = []
     command_results: list[CheckResult] = []
     test_plan: RepositoryTestPlan | None = None
+    source_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
 
 
 class FixPreparationAttempt(ContractModel):
@@ -352,7 +360,8 @@ class FileManifestEntry(ContractModel):
 class FixPreparationResultV1(ContractModel):
     version: Literal["1"] = "1"
     # Absent on historical records; keep their stronger, paired-proof interpretation.
-    validation_mode: Literal["paired", "native_tests"] = "paired"
+    validation_mode: Literal["paired", "native_tests", "agent_review"] = "paired"
+    prepared_source_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
     state: PreparationState
     stop_reason: str
     source_identity: SourceIdentity | None

strix/fix/prepare.py

@@ -32,8 +32,7 @@ from strix.fix.contracts import (
     VerificationDecision,
     VerifierResult,
 )
-from strix.fix.evidence import command_status, record_test_execution
-from strix.fix.locations import AnchorStatus, anchor_location
+from strix.fix.evidence import command_status
 
 
 class PreparationCancelledError(RuntimeError):
@@ -47,14 +46,9 @@ CancellationCheck = Callable[[], bool]
 
 @dataclass(slots=True)
 class PreparationPolicy:
-    max_repair_attempts: int = 2
-    timeout_seconds: int = 1800
+    timeout_seconds: int = 7200
     max_output_chars: int = 20000
 
-    def __post_init__(self) -> None:
-        if not 1 <= self.max_repair_attempts <= 2:
-            raise ValueError("max_repair_attempts must be between 1 and 2")
-
 
 @dataclass(slots=True)
 class PreparationContext:
@@ -74,7 +68,7 @@ IndependentVerifier = Callable[
     Awaitable[VerifierResult],
 ]
 SourceVerifier = Callable[[PreparationContext], Awaitable[bool]]
-CheckPlanner = Callable[[PreparationContext, list[FileManifestEntry]], Awaitable[list[CommandSpec]]]
+EvidenceReader = Callable[[], Awaitable[list[CheckResult]]]
 
 
 _COMMAND_ENV_ALLOWLIST = frozenset(
@@ -378,7 +372,10 @@ def _result(
 ) -> FixPreparationResultV1:
     return FixPreparationResultV1(
         state=state,
-        validation_mode="native_tests",
+        validation_mode="agent_review",
+        prepared_source_digest=(
+            context.feedback[-1].repair.source_digest if context.feedback else None
+        ),
         test_plan=context.feedback[-1].repair.test_plan if context.feedback else None,
         stop_reason=reason,
         source_identity=context.candidate.source_identity,
@@ -408,128 +405,53 @@ def _repair_outcome(value: RepairOutcome | None) -> RepairOutcome:
     )
 
 
-def _required_checks_pass(checks: list[CheckResult]) -> bool:
-    required = [result for result in checks if result.required]
-    return bool(required) and all(
-        result.status is CheckStatus.PASSED and result.exit_code == 0 for result in required
-    )
-
-
-def _verification_passes(checks: list[CheckResult], verifier: VerifierResult) -> bool:
-    return (
-        _required_checks_pass(checks)
-        and any(
-            item.purpose == "regression" and item.status is CheckStatus.PASSED for item in checks
-        )
-        and any(
-            item.purpose == "unit" and item.required and item.status is CheckStatus.PASSED
+def validation_gaps(
+    checks: list[CheckResult], source_digest: str | None, requested: list[CommandSpec]
+) -> list[str]:
+    """Check execution facts only; agents judge coverage and the security fix."""
+    gaps: list[str] = []
+    if not source_digest:
+        gaps.append("No source digest was recorded for the prepared patch.")
+    for purpose in ("regression", "unit"):
+        if not any(item.purpose == purpose and item.required for item in checks):
+            gaps.append(f"Run and record the required {purpose} tests.")  # noqa: PERF401
+    for item in checks:
+        if not item.required:
+            continue
+        if item.source_digest != source_digest or not item.environment_id:
+            gaps.append(f"Rerun {item.name}: its results are not for the current patch.")
+        elif item.status is not CheckStatus.PASSED or item.exit_code != 0 or item.tests_passed == 0:
+            gaps.append(f"Resolve and rerun {item.name}: {item.status}.")
+    if len({item.environment_id for item in checks if item.required}) > 1:
+        gaps.append("Required checks refer to different execution environments.")
+    for command in requested:
+        if command.required and not any(
+            item.required and item.argv == command.argv and item.cwd == command.cwd
             for item in checks
-        )
-        and len({item.environment_id for item in checks if item.environment_id}) <= 1
-        and len({item.source_digest for item in checks if item.source_digest}) <= 1
-        and verifier.decision is VerificationDecision.VERIFIED
-        and verifier.security_invariant_closed
-        and verifier.regression_test_valid
-        and verifier.unit_test_coverage_valid
-        and verifier.review_basis in {"execution", "code_review"}
-        and not verifier.gaps
-        and verifier.blocker is None
-    )
-
-
-def _test_plan_gaps(repair: RepairOutcome, manifest: list[FileManifestEntry]) -> list[str]:
-    plan = repair.test_plan
-    if plan is None:
-        return ["The repair must supply a regression test and the repository unit-test commands."]
-    changed = {entry.path for entry in manifest if entry.operation != "delete"}
-    missing = [path for path in plan.regression_files if path not in changed]
-    gaps = [f"Regression test must be added or updated in the patch: {path}" for path in missing]
-    if not plan.unit_tests:
-        gaps.append("Supply the customer's existing unit-test suite for the affected code.")
-        if plan.no_unit_tests_reason:
-            gaps.append(plan.no_unit_tests_reason)
+        ):
+            gaps.append(f"Run the requested check: {command.name}.")  # noqa: PERF401
     return gaps
 
 
-def _resolve_review(verifier: VerifierResult) -> VerifierResult:
-    """Derive the outcome from actionable concerns, retaining legacy record support."""
-    if verifier.concerns is None:
-        return verifier
-    repairs = [c.summary for c in verifier.concerns if c.kind == "repair_needed"]
-    prerequisites = [c.summary for c in verifier.concerns if c.kind == "customer_prerequisite"]
-    notes = [c.summary for c in verifier.concerns if c.kind == "optional_follow_up"]
-    approved = (
-        verifier.security_invariant_closed
-        and verifier.regression_test_valid
-        and verifier.unit_test_coverage_valid
-    )
-    decision = (
-        VerificationDecision.REJECTED
-        if repairs
-        else VerificationDecision.INCONCLUSIVE
-        if prerequisites or not approved
-        else VerificationDecision.VERIFIED
-    )
-    return verifier.model_copy(
-        update={
-            "decision": decision,
-            "repairable": bool(repairs),
-            "gaps": [*repairs, *prerequisites],
-            "notes": list(dict.fromkeys([*verifier.notes, *notes])),
-            "blocker": PreparationBlocker(
-                kind=BlockerKind.EXTERNAL_CONFIGURATION,
-                summary="A required customer prerequisite is missing.",
-                user_action="\n".join(prerequisites),
-                details=prerequisites,
-            )
-            if prerequisites
-            else None,
-        }
-    )
-
-
-def _test_commands(repair: RepairOutcome) -> list[CommandSpec]:
-    plan = repair.test_plan
-    if plan is None:
-        return []
-    return [
-        plan.regression_test.model_copy(update={"purpose": "regression", "required": True}),
-        *(
-            item.model_copy(update={"purpose": "unit", "required": True})
-            for item in plan.unit_tests
-        ),
-    ]
-
-
-async def prepare_fix(  # noqa: PLR0915
+async def prepare_fix(  # noqa: PLR0915 - thin orchestration and cleanup
     request: FixPreparationRequestV1,
     workspace: Path,
     *,
     repair: RepairAgent,
     verify: IndependentVerifier,
-    command_runner: CommandRunner = run_command,
     manifest_builder: ManifestBuilder = build_git_manifest,
     source_verifier: SourceVerifier = _verify_source,
-    check_planner: CheckPlanner | None = None,
+    evidence_reader: EvidenceReader | None = None,
     cancelled: CancellationCheck = lambda: False,
     policy: PreparationPolicy | None = None,
 ) -> FixPreparationResultV1:
+    """Run repair/review conversations; agents own setup, tests and corrections."""
     started = time.monotonic()
-    resolved_policy = policy or PreparationPolicy(
-        max_repair_attempts=request.max_repair_attempts,
-        timeout_seconds=request.timeout_seconds,
-    )
+    resolved_policy = policy or PreparationPolicy(timeout_seconds=request.timeout_seconds)
     context = PreparationContext(request=request, workspace=workspace, candidate=request.candidate)
-    runner = command_runner
-    if runner is run_command:
-        runner = functools.partial(
-            run_command,
-            credentials_allowed=request.credentials_allowed,
-            network_allowed=request.network_allowed,
-        )
     checks: list[CheckResult] = []
     verifier: VerifierResult | None = None
-    reproduction: CheckResult | None = None
+    repair_turns = review_turns = 0
 
     async def finish(
         state: PreparationState,
@@ -538,201 +460,110 @@ async def prepare_fix(  # noqa: PLR0915
         blocker: PreparationBlocker | None = None,
         gaps: list[str] | None = None,
     ) -> FixPreparationResultV1:
+        nonlocal checks
+        if evidence_reader:
+            checks = await evidence_reader()
         manifest, summary, artifact = await manifest_builder(workspace)
-        retained_verifier = verifier or (
-            context.feedback[-1].verifier if context.feedback else None
-        )
         return _result(
             context,
             state=state,
             reason=reason,
             started=started,
             checks=checks,
-            verifier=retained_verifier,
-            reproduction=reproduction,
+            verifier=verifier,
             manifest=manifest,
             diff_summary=summary,
             artifact_ref=artifact,
             attempt_history=context.feedback,
             blocker=blocker,
             gaps=gaps,
+            reproduction=next((c for c in checks if c.purpose == "regression"), None),
         )
 
-    async def execute() -> FixPreparationResultV1:  # noqa: PLR0911, PLR0912, PLR0915
-        nonlocal checks, verifier, reproduction
-        blocker: PreparationBlocker | None
+    async def execute() -> FixPreparationResultV1:  # noqa: PLR0912
+        nonlocal checks, verifier, repair_turns, review_turns
         if cancelled():
             raise PreparationCancelledError
         if not await source_verifier(context):
-            blocker = PreparationBlocker(
-                kind=BlockerKind.SOURCE,
-                summary="The repository no longer matches the finding source.",
-                user_action="Refresh the finding against the current repository revision.",
+            return await finish(
+                PreparationState.STALE,
+                "The repository no longer matches the finding source.",
+                blocker=PreparationBlocker(
+                    kind=BlockerKind.SOURCE,
+                    summary="The repository no longer matches the finding source.",
+                    user_action="Refresh the finding against the current repository revision.",
+                ),
             )
-            return _result(
-                context,
-                state=PreparationState.STALE,
-                reason=blocker.summary,
-                blocker=blocker,
-                started=started,
-            )
-        anchors = [
-            anchor_location(workspace, location, exact_source=True)
-            for location in context.candidate.finding_locations
-        ]
-        unresolved = [item for item in anchors if item.status is not AnchorStatus.UNIQUE]
-        if unresolved:
-            blocker = PreparationBlocker(
-                kind=BlockerKind.SOURCE,
-                summary="The reported finding locations could not be resolved uniquely.",
-                user_action="Refresh the finding or identify the affected source location.",
-                details=[f"{item.location.file}: {item.status}" for item in unresolved],
-            )
-            return await finish(PreparationState.BLOCKED, blocker.summary, blocker=blocker)
-        context.candidate = context.candidate.model_copy(
-            update={"finding_locations": [item.location for item in anchors]}
-        )
-        for attempt in range(1, resolved_policy.max_repair_attempts + 1):
-            context.attempt = attempt
+        # Location/snippet interpretation belongs to repair. Exact source identity is checked above.
+        while repair_turns < request.max_agent_turns and review_turns < request.max_agent_turns:
             if cancelled():
                 raise PreparationCancelledError
+            context.attempt += 1
             verifier = None
-            reproduction = None
             record = FixPreparationAttempt(
-                attempt=attempt,
+                attempt=context.attempt,
                 repair=RepairOutcome(status=RepairStatus.INCOMPLETE, summary="Repair started."),
                 workspace_digest=await _workspace_digest(workspace),
             )
             context.feedback.append(record)
             record.repair = _repair_outcome(await repair(context, checks))
-            checks = record.checks
+            repair_turns += max(1, record.repair.turns_used)
+            checks = await evidence_reader() if evidence_reader else record.repair.command_results
+            record.checks = list(checks)
             record.workspace_digest = await _workspace_digest(workspace)
-            manifest, _, _ = await build_git_manifest(workspace)
-            can_retry = (
-                attempt < resolved_policy.max_repair_attempts
-                and record.repair.status is not RepairStatus.BLOCKED
-            )
-            if not manifest:
-                if record.repair.blocker:
-                    return await finish(
-                        PreparationState.BLOCKED,
-                        record.repair.summary,
-                        blocker=record.repair.blocker,
-                        gaps=record.repair.gaps,
-                    )
-                record.repair.gaps.append(
-                    "No changed files were found. Use the file tools to implement the fix and "
-                    "regression test, then submit their validation commands."
-                )
-                if can_retry:
-                    continue
-                return await finish(
-                    PreparationState.FAILED,
-                    "The repair did not change repository source.",
-                    gaps=record.repair.gaps,
-                )
-            await manifest_builder(workspace)  # Save the patch before any test can fail.
-            plan_gaps = _test_plan_gaps(record.repair, manifest)
-            if plan_gaps:
-                record.repair.gaps = list(dict.fromkeys([*record.repair.gaps, *plan_gaps]))
-                if can_retry:
-                    continue
-                return await finish(
-                    PreparationState.BLOCKED,
-                    "The patch is saved, but its required test handoff is incomplete.",
-                    blocker=record.repair.blocker,
-                    gaps=record.repair.gaps,
-                )
-            additional = await check_planner(context, manifest) if check_planner else request.checks
-            planned = [*_test_commands(record.repair), *additional]
-            unique: dict[tuple[str, tuple[str, ...], str], CommandSpec] = {}
-            for command in planned:
-                key = (command.purpose, tuple(command.argv), command.cwd)
-                previous = unique.get(key)
-                unique[key] = command.model_copy(
-                    update={"required": command.required or bool(previous and previous.required)}
-                )
-            checks = record.checks
-            for command in unique.values():
-                if cancelled():
-                    raise PreparationCancelledError
-                checks.append(record_test_execution(await runner(workspace, command), command))
-            reproduction = next((item for item in checks if item.purpose == "regression"), None)
-            record.security_reproduction = reproduction
-            failed = [
-                item
-                for item in checks
-                if item.required and (item.status is not CheckStatus.PASSED or item.exit_code != 0)
-            ]
-            if failed:
-                if can_retry:
-                    continue
-                return await finish(
-                    PreparationState.BLOCKED,
-                    "The patch is saved, but required tests or checks have not passed.",
-                    blocker=PreparationBlocker(
-                        kind=BlockerKind.VERIFICATION_RUNTIME,
-                        summary="Required tests or checks have not passed.",
-                        user_action="Review the recorded failures and retry after resolving them.",
-                        details=[f"{item.name}: {item.status}" for item in failed],
-                    ),
-                    gaps=[f"{item.name}: {item.status}" for item in failed],
-                )
-            # Review the exact repair and its native test results. No second harness is required.
-            verifier = _resolve_review(await verify(context, checks))
-            record.verifier = verifier
-            if record.workspace_digest != await _workspace_digest(workspace):
-                return await finish(
-                    PreparationState.FAILED,
-                    "Review changed the prepared source; validation must be rerun.",
-                )
-            if verifier.decision is VerificationDecision.REJECTED:
-                if verifier.repairable and can_retry:
-                    continue
-                return await finish(
-                    PreparationState.FAILED,
-                    "The independent reviewer found a repair or regression-test defect.",
-                    gaps=verifier.gaps,
-                )
-            if verifier.blocker:
-                return await finish(
-                    PreparationState.BLOCKED,
-                    verifier.blocker.summary,
-                    blocker=verifier.blocker,
-                    gaps=verifier.gaps,
-                )
-            if record.repair.status is RepairStatus.BLOCKED:
+            manifest, _, _ = await manifest_builder(workspace)
+            if record.repair.status is not RepairStatus.COMPLETE:
                 return await finish(
                     PreparationState.BLOCKED,
                     record.repair.summary,
                     blocker=record.repair.blocker,
                     gaps=record.repair.gaps,
                 )
-            # Typed review explicitly resolves the repair agent's observations. Historical
-            # free-text gaps remain blocking until a reviewer classifies them.
-            repair_gaps = record.repair.gaps if verifier.concerns is None else []
-            if _verification_passes(checks, verifier) and not repair_gaps:
-                return await finish(
-                    PreparationState.READY,
-                    "Tests and required checks passed; independent review approved a draft PR.",
+            if not manifest:
+                record.repair.gaps.append(
+                    "No changed files were found. Implement the fix and regression test."
                 )
-            gaps = list(dict.fromkeys([*repair_gaps, *verifier.gaps]))
-            if not gaps:
-                gaps = ["Independent review could not confirm the repair and native-test coverage."]
+                continue
+            if cancelled():
+                raise PreparationCancelledError
+            # Review can investigate even incomplete validation and run the missing checks itself.
+            verifier = await verify(context, checks)
+            review_turns += max(1, verifier.turns_used)
+            record.verifier = verifier
+            if evidence_reader:
+                checks = await evidence_reader()
+                record.checks = list(checks)
+            if verifier.blocker:
+                return await finish(
+                    PreparationState.BLOCKED,
+                    verifier.summary,
+                    blocker=verifier.blocker,
+                    gaps=verifier.gaps,
+                )
+            if verifier.decision is VerificationDecision.REJECTED:
+                record.repair.gaps.extend(verifier.gaps or [verifier.summary])
+                continue
+            if verifier.decision is not VerificationDecision.VERIFIED:
+                return await finish(PreparationState.BLOCKED, verifier.summary, gaps=verifier.gaps)
+            gaps = validation_gaps(checks, record.repair.source_digest, request.checks)
+            if (
+                record.workspace_digest != await _workspace_digest(workspace)
+                or verifier.source_digest != record.repair.source_digest
+            ):
+                gaps.append(
+                    "The deliverable changed during review. Inspect the diff, clean up temporary "
+                    "files, and rerun affected checks before requesting review again."
+                )
+            if gaps:
+                record.repair.gaps.extend(gaps)
+                continue
             return await finish(
-                PreparationState.BLOCKED,
-                "The patch and test results are saved, but independent review is incomplete.",
-                blocker=verifier.blocker
-                or PreparationBlocker(
-                    kind=BlockerKind.VERIFICATION_RUNTIME,
-                    summary="Independent review is incomplete.",
-                    user_action="Retry review of the saved changes and test evidence.",
-                    details=gaps,
-                ),
-                gaps=gaps,
+                PreparationState.READY,
+                "Required tests passed and independent review approved the draft PR.",
             )
         return await finish(
-            PreparationState.FAILED, "The repair exhausted its configured attempts."
+            PreparationState.BLOCKED,
+            "The agent turn budget was reached; partial work was retained.",
         )
 
     try:
@@ -746,7 +577,6 @@ async def prepare_fix(  # noqa: PLR0915
         logging.getLogger(__name__).exception(
             "Fix preparation failed during attempt %s", context.attempt
         )
-        # Preserve partial work without exposing unredacted exception text.
         return await finish(
             PreparationState.FAILED,
             f"Fix preparation stopped after {type(error).__name__}; partial work was retained.",

tests/test_fix_preparation.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
 import subprocess
@@ -23,9 +24,7 @@ from strix.fix.contracts import (
     RegressionTestResult,
     RepairOutcome,
     RepairStatus,
-    RepositoryTestPlan,
     ReproductionSpec,
-    ReviewConcern,
     SourceIdentity,
     SourceIdentityKind,
     VerificationDecision,
@@ -37,7 +36,9 @@ from strix.fix.evidence import command_status, passed_test_count, record_test_ex
 from strix.fix.locations import AnchorStatus, anchor_location
 from strix.fix.prepare import (
     PreparationContext,
+    PreparationPolicy,
     _network_isolation_prefix,
+    _workspace_digest,
     build_git_manifest,
     build_git_patch,
     prepare_fix,
@@ -160,6 +161,7 @@ def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationR
             )
         ],
         max_repair_attempts=attempts,
+        max_agent_turns=8,
         network_allowed=True,
     )
 
@@ -178,20 +180,34 @@ async def _noop_repair(
         "import unittest\nfrom app import result\nclass SecurityTest(unittest.TestCase):\n"
         "    def test_safe(self): self.assertEqual(result(), 'safe')\n"
     )
-    return RepairOutcome(
-        test_plan=RepositoryTestPlan(
-            regression_files=["test_app.py"],
-            regression_test=CommandSpec(
-                name="regression", argv=[sys.executable, "-m", "unittest", "test_app"]
-            ),
-            unit_tests=[
-                CommandSpec(
-                    name="existing suite", argv=[sys.executable, "-m", "unittest", "test_existing"]
-                )
-            ],
+    commands = [
+        CommandSpec(
+            name="regression",
+            argv=[sys.executable, "-m", "unittest", "test_app"],
+            purpose="regression",
         ),
+        CommandSpec(
+            name="existing suite",
+            argv=[sys.executable, "-m", "unittest", "test_existing"],
+            purpose="unit",
+        ),
+        *context.request.checks,
+    ]
+    digest = await _workspace_digest(context.workspace)
+    results = []
+    for command in commands:
+        result = record_test_execution(
+            await run_command(context.workspace, command, network_allowed=True), command
+        )
+        results.append(
+            result.model_copy(update={"source_digest": digest, "environment_id": "test"})
+        )
+    return RepairOutcome(
         status=RepairStatus.COMPLETE,
-        summary="The repository fix is ready for independent evaluation.",
+        summary="Fixed and tested.",
+        command_results=results,
+        source_digest=digest,
+        turns_used=1,
     )
 
 
@@ -201,6 +217,7 @@ async def _verified(
 ) -> VerifierResult:
     return VerifierResult(
         decision=VerificationDecision.VERIFIED,
+        source_digest=_context.feedback[-1].repair.source_digest,
         summary="The invariant is closed.",
         security_invariant_closed=True,
         review_basis="code_review",
@@ -475,252 +492,176 @@ def test_regression_requires_consistent_execution_provenance() -> None:
 
 
 @pytest.mark.asyncio
-async def test_native_flow_runs_regression_before_review_and_saves_test(tmp_path: Path) -> None:
+async def test_agent_tests_are_reused_without_controller_execution(tmp_path: Path) -> None:
     workspace, commit = _workspace(tmp_path)
-
-    async def review(context, checks):
-        assert checks[0].purpose == "regression"
-        assert checks[0].tests_passed == 1
-        assert checks[0].exit_code == 0
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=_noop_repair, verify=review
-    )
-    assert result.state is PreparationState.READY, result.model_dump_json()
-    assert result.validation_mode == "native_tests"
-    assert result.test_plan.regression_files == ["test_app.py"]
-    assert set(result.changed_files) == {"app.py", "test_app.py"}
-    assert result.verifier.review_basis == "code_review"
-
-
-@pytest.mark.asyncio
-async def test_failing_unit_suite_is_returned_to_repair_before_review(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-    calls = []
+    outcome = None
 
     async def repair(context, checks):
-        calls.append(context.attempt)
+        nonlocal outcome
         outcome = await _noop_repair(context, checks)
-        (workspace / "test_existing.py").write_text(
-            "import unittest\nclass Existing(unittest.TestCase):\n"
-            f"    def test_existing(self): self.assertTrue({context.attempt > 1})\n"
-        )
-        outcome.test_plan.unit_tests = [
-            CommandSpec(
-                name="existing suite", argv=[sys.executable, "-m", "unittest", "test_existing"]
-            )
-        ]
-        outcome.test_plan.no_unit_tests_reason = None
-        if context.attempt == 2:
-            assert any(c.purpose == "unit" and c.status is CheckStatus.FAILED for c in checks)
         return outcome
 
     result = await prepare_fix(
         _request(_candidate(commit)), workspace, repair=repair, verify=_verified
     )
     assert result.state is PreparationState.READY
-    assert calls == [1, 2]
-    assert result.attempt_history[0].verifier is None
-    assert result.checks[1].tests_passed == 1
+    assert result.validation_mode == "agent_review"
+    assert result.test_plan is None
+    assert result.checks == outcome.command_results
+    assert result.checks[0].tests_passed == 1
+    assert result.prepared_source_digest == result.verifier.source_digest
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "output", ["Ran 0 tests in 0.000s\n\nOK", "3 skipped", "collected 0 items"]
-)
-async def test_empty_or_skipped_tests_never_approve(tmp_path: Path, output: str) -> None:
+async def test_review_can_request_more_than_two_repairs(tmp_path: Path) -> None:
     workspace, commit = _workspace(tmp_path)
 
-    async def runner(_root, command):
-        return CheckResult(
-            name=command.name,
-            argv=command.argv,
-            status=CheckStatus.PASSED,
-            exit_code=0,
-            output=output,
-            duration_seconds=0,
-        )
+    async def review(context, checks):
+        if context.attempt < 4:
+            return VerifierResult(
+                decision=VerificationDecision.REJECTED,
+                summary="Inspect sibling path",
+                gaps=["Inspect sibling path"],
+            )
+        assert context.feedback[-2].verifier.summary == "Inspect sibling path"
+        return await _verified(context, checks)
 
-    async def review(*_args):
-        pytest.fail("Review must not run without passing tests")
+    result = await prepare_fix(
+        _request(_candidate(commit)), workspace, repair=_noop_repair, verify=review
+    )
+    assert result.state is PreparationState.READY
+    assert len(result.attempt_history) == 4  # Legacy max_repair_attempts=2 is not a loop cap.
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("defect", ["missing_unit", "failed", "empty", "stale", "missing_request"])
+async def test_approval_cannot_waive_required_execution_evidence(
+    tmp_path: Path, defect: str
+) -> None:
+    workspace, commit = _workspace(tmp_path)
+
+    async def repair(context, checks):
+        result = await _noop_repair(context, checks)
+        if defect == "missing_unit":
+            result.command_results = [c for c in result.command_results if c.purpose != "unit"]
+        elif defect == "missing_request":
+            result.command_results = [c for c in result.command_results if c.purpose != "quality"]
+        else:
+            check = result.command_results[0]
+            if defect == "failed":
+                check.status, check.exit_code = CheckStatus.FAILED, 1
+            elif defect == "empty":
+                check.tests_passed = 0
+            else:
+                check.source_digest = "a" * 64
+        return result
+
+    request = _request(_candidate(commit))
+    request.max_agent_turns = 2
+    result = await prepare_fix(request, workspace, repair=repair, verify=_verified)
+    assert result.state is PreparationState.BLOCKED
+    assert result.final_file_manifest
+    assert result.attempt_history[0].repair.gaps
+
+
+@pytest.mark.asyncio
+async def test_incomplete_validation_can_be_finished_by_reviewer(tmp_path: Path) -> None:
+    workspace, commit = _workspace(tmp_path)
+    evidence = []
+
+    async def repair(context, checks):
+        result = await _noop_repair(context, checks)
+        evidence.extend(result.command_results)
+        evidence[1].status = CheckStatus.FAILED
+        return result
+
+    async def review(context, checks):
+        assert checks[1].status is CheckStatus.FAILED
+        command = CommandSpec(name="existing suite", argv=checks[1].argv, purpose="unit")
+        executed = record_test_execution(
+            await run_command(workspace, command, network_allowed=True), command
+        )
+        evidence[1] = executed.model_copy(
+            update={"source_digest": evidence[0].source_digest, "environment_id": "test"}
+        )
+        return await _verified(context, checks)
+
+    async def read_evidence():
+        return list(evidence)
 
     result = await prepare_fix(
         _request(_candidate(commit)),
         workspace,
-        repair=_noop_repair,
+        repair=repair,
         verify=review,
-        command_runner=runner,
-    )
-    assert result.state is PreparationState.BLOCKED
-    assert result.final_file_manifest
-    assert result.checks[0].status is CheckStatus.SKIPPED
-
-
-@pytest.mark.asyncio
-async def test_reviewer_concrete_defect_retries_and_reruns_tests(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def review(context, checks):
-        if context.attempt == 1:
-            return VerifierResult(
-                decision=VerificationDecision.REJECTED,
-                summary="Missing sibling guard",
-                gaps=["Check the second entry point"],
-                repairable=True,
-            )
-        assert context.feedback[0].verifier.gaps == ["Check the second entry point"]
-        return await _verified(context, checks)
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=_noop_repair, verify=review
+        evidence_reader=read_evidence,
     )
     assert result.state is PreparationState.READY
-    assert len(result.attempt_history) == 2
-    assert all(a.checks[0].tests_passed == 1 for a in result.attempt_history)
+    assert result.checks[1].status is CheckStatus.PASSED
 
 
 @pytest.mark.asyncio
-async def test_exhausted_repair_can_be_validated_from_saved_work(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def repair(context, checks):
-        outcome = await _noop_repair(context, checks)
-        outcome.status = RepairStatus.BUDGET_EXHAUSTED
-        return outcome
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=repair, verify=_verified
-    )
-    assert result.state is PreparationState.READY
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("recover", [True, False])
-async def test_empty_completion_gets_actionable_bounded_retry(
-    tmp_path: Path, recover: bool
+@pytest.mark.parametrize("stop", ["budget", "blocked", "exception", "timeout", "cancel"])
+async def test_interruptions_preserve_partial_patch_without_approval(
+    tmp_path: Path, stop: str
 ) -> None:
+
     workspace, commit = _workspace(tmp_path)
+    cancel = False
 
     async def repair(context, checks):
-        if context.attempt == 2:
-            assert "No changed files" in context.feedback[0].repair.gaps[0]
-            if recover:
-                return await _noop_repair(context, checks)
-        return RepairOutcome(status=RepairStatus.COMPLETE, summary="Claimed completion")
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=repair, verify=_verified
-    )
-    assert result.attempts == 2
-    assert result.state is (PreparationState.READY if recover else PreparationState.FAILED)
-    if not recover:
-        assert not result.final_file_manifest
-        assert "No changed files" in result.gaps[0]
-
-
-@pytest.mark.asyncio
-async def test_missing_customer_suite_cannot_be_waived_with_an_explanation(tmp_path: Path) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def repair(context, checks):
-        outcome = await _noop_repair(context, checks)
-        outcome.test_plan.unit_tests = []
-        outcome.test_plan.no_unit_tests_reason = "No existing suite was found."
-        return outcome
+        nonlocal cancel
+        result = await _noop_repair(context, checks)
+        if stop == "budget":
+            result.status = RepairStatus.BUDGET_EXHAUSTED
+        elif stop == "blocked":
+            result.status = RepairStatus.BLOCKED
+        elif stop == "exception":
+            raise RuntimeError("provider error")
+        elif stop == "timeout":
+            await asyncio.sleep(10)
+        else:
+            cancel = True
+        return result
 
     async def review(*_args):
-        pytest.fail("A reviewer cannot waive required customer unit tests")
+        raise AssertionError("Stopped repair must not be approved")
 
     result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=repair, verify=review
-    )
-    assert result.state is PreparationState.BLOCKED
-    assert result.final_file_manifest
-    assert any("existing unit-test suite" in gap for gap in result.gaps)
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize("kind", ["repair_needed", "customer_prerequisite", "optional_follow_up"])
-async def test_controller_derives_review_action_and_resolves_repair_observations(
-    tmp_path: Path, kind: str
-) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def repair(context, checks):
-        outcome = await _noop_repair(context, checks)
-        outcome.gaps = ["An additional deployment check could help."]
-        if context.attempt == 2:
-            assert context.feedback[0].verifier.gaps == ["Inspect the companion configuration."]
-        return outcome
-
-    async def review(context, checks):
-        result = await _verified(context, checks)
-        # The controller must derive the outcome even if the proposed decision says verified.
-        result.concerns = (
-            []
-            if context.attempt == 2
-            else [ReviewConcern(kind=kind, summary="Inspect the companion configuration.")]
-        )
-        return result
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=repair, verify=review
-    )
-    assert result.attempts == (2 if kind == "repair_needed" else 1)
-    if kind == "customer_prerequisite":
-        assert result.state is PreparationState.BLOCKED
-        assert result.blocker.user_action == "Inspect the companion configuration."
-    else:
-        assert result.state is PreparationState.READY
-        assert not result.gaps
-        assert not result.verifier.gaps
-        if kind == "optional_follow_up":
-            assert result.verifier.notes == ["Inspect the companion configuration."]
-
-
-@pytest.mark.asyncio
-@pytest.mark.parametrize(
-    "defect",
-    [
-        "missing_plan",
-        "missing_file",
-        "rejected_test",
-        "inconclusive_review",
-        "exception",
-        "source_change",
-    ],
-)
-async def test_incomplete_work_is_retained_without_ready(tmp_path: Path, defect: str) -> None:
-    workspace, commit = _workspace(tmp_path)
-
-    async def repair(context, checks):
-        outcome = await _noop_repair(context, checks)
-        if defect == "missing_plan":
-            outcome.test_plan = None
-        if defect == "missing_file":
-            outcome.test_plan.regression_files = ["invented.py"]
-        return outcome
-
-    async def review(context, checks):
-        if defect == "exception":
-            raise RuntimeError("review provider failed")
-        if defect == "source_change":
-            (workspace / "app.py").write_text("bad=True\n")
-        result = await _verified(context, checks)
-        if defect == "rejected_test":
-            result.regression_test_valid = False
-        if defect == "inconclusive_review":
-            result.decision = VerificationDecision.INCONCLUSIVE
-        return result
-
-    result = await prepare_fix(
-        _request(_candidate(commit)), workspace, repair=repair, verify=review
+        _request(_candidate(commit)),
+        workspace,
+        repair=repair,
+        verify=review,
+        cancelled=lambda: cancel,
+        policy=PreparationPolicy(timeout_seconds=1 if stop == "timeout" else 30),
     )
     assert result.state in {PreparationState.BLOCKED, PreparationState.FAILED}
     assert result.final_file_manifest
 
 
+@pytest.mark.asyncio
+async def test_review_mutation_returns_to_repair_instead_of_destroying_run(tmp_path: Path) -> None:
+    workspace, commit = _workspace(tmp_path)
+
+    async def review(context, checks):
+        result = await _verified(context, checks)
+        if context.attempt == 1:
+            (workspace / "review.tmp").write_text("temporary")
+        return result
+
+    async def repair(context, checks):
+        if context.attempt > 1:
+            assert "deliverable changed" in " ".join(context.feedback[-2].repair.gaps)
+            (workspace / "review.tmp").unlink()
+        return await _noop_repair(context, checks)
+
+    result = await prepare_fix(
+        _request(_candidate(commit)), workspace, repair=repair, verify=review
+    )
+    assert result.state is PreparationState.READY
+    assert result.attempts == 2
+
+
 @pytest.mark.parametrize(
     ("output", "expected"),
     [