diff --git a/strix/fix/__init__.py b/strix/fix/__init__.py index de4ef9321..44310b235 100644 --- a/strix/fix/__init__.py +++ b/strix/fix/__init__.py @@ -18,6 +18,7 @@ from strix.fix.contracts import ( RegressionTestResult, RepairOutcome, RepairStatus, + RepositoryTestPlan, ReproductionSpec, SourceIdentity, SourceIdentityKind, @@ -58,6 +59,7 @@ __all__ = [ "RegressionTestResult", "RepairOutcome", "RepairStatus", + "RepositoryTestPlan", "ReproductionSpec", "SourceIdentity", "SourceIdentityKind", diff --git a/strix/fix/contracts.py b/strix/fix/contracts.py index e99f2b4f5..df7c796ab 100644 --- a/strix/fix/contracts.py +++ b/strix/fix/contracts.py @@ -133,6 +133,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" _relative_cwd = field_validator("cwd")(_validate_relative_path) @@ -149,6 +150,28 @@ class ReproductionSpec(ContractModel): command: CommandSpec | None = None +class RepositoryTestPlan(ContractModel): + """The repair agent's native tests, rerun by the controller before review.""" + + regression_test: CommandSpec + regression_files: list[str] = Field(min_length=1) + unit_tests: list[CommandSpec] = [] + no_unit_tests_reason: str | None = None + + @field_validator("regression_files") + @classmethod + def validate_files(cls, paths: list[str]) -> list[str]: + return list(dict.fromkeys(_validate_relative_path(path) for path in paths)) + + @model_validator(mode="after") + def explain_missing_suite(self) -> RepositoryTestPlan: + if not self.unit_tests and not (self.no_unit_tests_reason or "").strip(): + raise ValueError( + "Provide the existing unit-test commands or explain why no suite exists." + ) + return self + + class ReportedCheck(ContractModel): name: str result: str @@ -177,6 +200,8 @@ class FixCandidateV1(ContractModel): data = self.model_dump(mode="json") if self.finding is None: data.pop("finding", None) # Preserve digests for stored legacy candidates. + if data.get("reproduction") and data["reproduction"].get("command"): + data["reproduction"]["command"].pop("purpose", None) payload = json.dumps( data, sort_keys=True, @@ -219,6 +244,8 @@ 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" + tests_passed: int | None = Field(default=None, ge=0) class RegressionTestResult(ContractModel): @@ -279,6 +306,9 @@ class VerifierResult(ContractModel): regression_tests: list[RegressionTestResult] = [] harnesses: list[VerificationHarness] = [] notes: list[str] = [] + review_basis: Literal["execution", "code_review"] | None = None + regression_test_valid: bool = False + unit_test_coverage_valid: bool = False class RepairOutcome(ContractModel): @@ -290,6 +320,7 @@ class RepairOutcome(ContractModel): blocker: PreparationBlocker | None = None checks: list[CommandSpec] = [] command_results: list[CheckResult] = [] + test_plan: RepositoryTestPlan | None = None class FixPreparationAttempt(ContractModel): @@ -313,6 +344,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" state: PreparationState stop_reason: str source_identity: SourceIdentity | None @@ -332,6 +365,7 @@ class FixPreparationResultV1(ContractModel): attempts: int = Field(default=0, ge=0) elapsed_seconds: float = Field(default=0, ge=0) cost_usd: float | None = Field(default=None, ge=0) + test_plan: RepositoryTestPlan | None = None def candidate_from_legacy_report( diff --git a/strix/fix/evidence.py b/strix/fix/evidence.py index 4f56446f3..3361a4321 100644 --- a/strix/fix/evidence.py +++ b/strix/fix/evidence.py @@ -5,7 +5,56 @@ from __future__ import annotations import re from typing import Literal -from strix.fix.contracts import CheckStatus +from strix.fix.contracts import CheckResult, CheckStatus, CommandSpec + + +def passed_test_count(output: str) -> int | None: # noqa: PLR0911 - native summary formats + """Recognize native test-runner summaries, not the model's account of a run. + + Counts are optional: runners use different output formats. This is execution + evidence, not an assertion that a test covers the finding; the reviewer checks that. + """ + plain = re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", output) + if command_status(0, plain)[0] is CheckStatus.SKIPPED: + return 0 + counts = re.findall(r"\b(\d+)\s+(?:passed|pass)\b", plain, re.IGNORECASE) + counts += re.findall(r"(?:#|\u2139)\s+pass\s+(\d+)\b", plain) + if counts: + return max(int(count) for count in counts) + unittest = re.search(r"Ran (\d+) tests? in [^\n]+\n\s*OK(?: \(skipped=(\d+)\))?", plain) + if unittest: + return max(0, int(unittest[1]) - int(unittest[2] or 0)) + rspec = re.search(r"(\d+) examples?, (\d+) failures?(?:, (\d+) pending)?", plain) + if rspec: + return max(0, int(rspec[1]) - int(rspec[2]) - int(rspec[3] or 0)) + minitest = re.search( + r"(\d+) runs, \d+ assertions, (\d+) failures, (\d+) errors, (\d+) skips", plain + ) + if minitest: + return max(0, int(minitest[1]) - sum(int(minitest[i]) for i in (2, 3, 4))) + if re.search(r"^(?:=+\s*)?[1-9]\d* skipped(?:\s+in [^\n=]+)?(?:\s*=+)?$", plain, re.MULTILINE): + return 0 + go = re.findall(r"^\s*--- PASS: ", plain, re.MULTILINE) + return len(go) if go else None + + +def record_test_execution(result: CheckResult, command: CommandSpec) -> CheckResult: + """Known empty/skipped runs cannot pass; unknown summary formats go to review.""" + count = passed_test_count(result.output) if command.purpose != "quality" else None + update: dict[str, object] = { + "purpose": command.purpose, + "tests_passed": count, + "required": command.required, + "name": command.name, + "argv": command.argv, + "cwd": command.cwd, + } + if command.purpose != "quality" and result.status is CheckStatus.PASSED and count == 0: + update.update( + status=CheckStatus.SKIPPED, + output=result.output + "\nThe runner reported no passing tests.", + ) + return result.model_copy(update=update) def command_status( diff --git a/strix/fix/prepare.py b/strix/fix/prepare.py index 0f6ad3078..da505547c 100644 --- a/strix/fix/prepare.py +++ b/strix/fix/prepare.py @@ -6,6 +6,7 @@ import asyncio import functools import hashlib import json +import logging import os import subprocess import time @@ -29,10 +30,9 @@ from strix.fix.contracts import ( RepairOutcome, RepairStatus, VerificationDecision, - VerificationTarget, VerifierResult, ) -from strix.fix.evidence import command_status +from strix.fix.evidence import command_status, record_test_execution from strix.fix.locations import AnchorStatus, anchor_location @@ -378,6 +378,8 @@ def _result( ) -> FixPreparationResultV1: return FixPreparationResultV1( state=state, + validation_mode="native_tests", + test_plan=context.feedback[-1].repair.test_plan if context.feedback else None, stop_reason=reason, source_identity=context.candidate.source_identity, candidate=context.candidate, @@ -408,92 +410,49 @@ 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 for result in 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: - evidence = [ - *checks, - *( - leg - for test in verifier.regression_tests - for leg in (test.base, test.patched, test.behavior) - ), - ] - environments = {item.environment_id for item in evidence if item.environment_id} - patched_sources = { - item.source_digest - for item in evidence - if item.target is not VerificationTarget.BASE and item.source_digest - } +def _verification_passes(checks: list[CheckResult], verifier: VerifierResult) -> bool: return ( _required_checks_pass(checks) - and len(environments) <= 1 - and len(patched_sources) <= 1 + and any( + item.purpose == "regression" and item.status is CheckStatus.PASSED 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.reproduction_executed + 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 - and bool(verifier.regression_tests) - and all(result.passed() for result in verifier.regression_tests) ) -def _verification_gaps( - repair_outcome: RepairOutcome, - checks: list[CheckResult], - verifier: VerifierResult, -) -> list[str]: - gaps = list(repair_outcome.gaps) - required = [result for result in checks if result.required] - if not required: - gaps.append("No required repository check was configured.") - gaps.extend( - f"{result.name}: required check {result.status}" - for result in required - if result.status is not CheckStatus.PASSED - ) - gaps.extend( - f"{result.name}: optional check {result.status}" - for result in checks - if not result.required and result.status is not CheckStatus.PASSED - ) - if not verifier.regression_tests or not all( - item.passed() for item in verifier.regression_tests - ): - gaps.append( - "A paired functional regression and legitimate-behavior test is still required." - ) - if not verifier.reproduction_executed: - gaps.append("The independent verifier did not execute the security reproduction.") - if not verifier.security_invariant_closed: - gaps.append("The independent verifier did not prove the security invariant.") - gaps.extend(verifier.gaps) - return list(dict.fromkeys(gaps)) +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] + return [f"Regression test must be added or updated in the patch: {path}" for path in missing] -def _matches_baseline_failure(result: CheckResult) -> bool: - if result.baseline_status is not result.status or result.status not in { - CheckStatus.FAILED, - CheckStatus.UNAVAILABLE, - CheckStatus.SKIPPED, - }: - return False - if result.baseline_exit_code != result.exit_code: - return False - candidate_output = result.output - baseline_output = result.baseline_output or "" - if result.workspace_root: - candidate_output = candidate_output.replace(result.workspace_root, "") - if result.baseline_workspace_root: - baseline_output = baseline_output.replace(result.baseline_workspace_root, "") - candidate_output = " ".join(candidate_output.split()) - baseline_output = " ".join(baseline_output.split()) - return bool(candidate_output and candidate_output == baseline_output) +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 @@ -600,6 +559,7 @@ async def prepare_fix( # noqa: PLR0915 ) context.feedback.append(record) record.repair = _repair_outcome(await repair(context, checks)) + checks = record.checks record.workspace_digest = await _workspace_digest(workspace) manifest, _, _ = await build_git_manifest(workspace) if not manifest: @@ -614,146 +574,98 @@ async def prepare_fix( # noqa: PLR0915 PreparationState.FAILED, "The repair did not change repository source." ) await manifest_builder(workspace) # Save the patch before any test can fail. - planned = await check_planner(context, manifest) if check_planner else request.checks - # Retain partial execution if a later command is interrupted. - checks = record.checks - for command in planned: - if cancelled(): - raise PreparationCancelledError - checks.append(await runner(workspace, command)) - # A late setup recovery invalidates checks from the old environment. - # Refresh earlier checks, with a bound even if setup keeps changing. - epoch = next( - (item.environment_id for item in reversed(checks) if item.environment_id), None - ) - for _refresh in range(2): - outdated = [ - index - for index, item in enumerate(checks) - if epoch and item.environment_id != epoch - ] - if not outdated: - break - for index in outdated: - checks[index] = await runner(workspace, planned[index]) - epoch = checks[index].environment_id or epoch - required = [item for item in checks if item.required] - regressions = [ - item - for item in required - if item.status is CheckStatus.FAILED - and item.failure_kind != "source_changed" - and not _matches_baseline_failure(item) - ] - unchanged_retry = ( - len(context.feedback) > 1 - and context.feedback[-2].workspace_digest == record.workspace_digest - ) can_retry = ( - record.repair.status is RepairStatus.COMPLETE - and attempt < resolved_policy.max_repair_attempts - and not unchanged_retry + attempt < resolved_policy.max_repair_attempts + and record.repair.status is not RepairStatus.BLOCKED ) - if regressions: + 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.FAILED, - "The repair did not pass the repository quality gate.", - gaps=[f"{item.name}: {item.output}" for item in regressions], + PreparationState.BLOCKED, + "The patch is saved, but its regression-test handoff is incomplete.", + blocker=record.repair.blocker, + gaps=record.repair.gaps, ) - # Missing tools and pre-existing failures do not erase attainable security evidence. + 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 = await verify(context, checks) record.verifier = verifier - reproduction = next( - (item.patched for item in reversed(verifier.regression_tests) if item.passed()), - verifier.regression_tests[-1].patched if verifier.regression_tests else None, - ) - record.security_reproduction = reproduction - gaps = _verification_gaps(record.repair, checks, verifier) if record.workspace_digest != await _workspace_digest(workspace): return await finish( PreparationState.FAILED, - "Verification changed the prepared source; its evidence cannot " - "approve this artifact.", + "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 security verifier found a repair defect.", - gaps=gaps, + "The independent reviewer found a repair or regression-test defect.", + gaps=verifier.gaps, ) if record.repair.status is RepairStatus.BLOCKED: return await finish( PreparationState.BLOCKED, record.repair.summary, blocker=record.repair.blocker, - gaps=gaps, + gaps=record.repair.gaps, ) if _verification_passes(checks, verifier) and not record.repair.gaps: return await finish( PreparationState.READY, - "The fix passed relevant repository checks and functional security " - "verification.", - gaps=gaps, + "Tests and required checks passed; independent review approved a draft PR.", ) - unavailable = [ - item - for item in required - if item.status - in {CheckStatus.UNAVAILABLE, CheckStatus.SKIPPED, CheckStatus.CANCELLED} - ] - baseline = [item for item in required if _matches_baseline_failure(item)] - blocker = verifier.blocker - if blocker is None: - internal_failures = [ - item - for item in verifier.security_tests - if item.failure_kind in {"harness", "unknown", "source_changed", "timeout"} - ] - if internal_failures or (not verifier.security_tests and not verifier.gaps): - blocker = PreparationBlocker( - kind=BlockerKind.VERIFICATION_RUNTIME, - summary="Fix prepared; Strix could not complete verification.", - user_action="Retry verification or review the prepared draft.", - details=[item.name for item in internal_failures] or gaps, - ) - elif unavailable or not required: - blocker = PreparationBlocker( - kind=BlockerKind.ENVIRONMENT, - summary=( - "The patch is prepared, but required repository checks could not run." - ), - user_action=( - "Resolve the reported setup requirement and rerun verification." - ), - details=[f"{item.name}: {item.output}" for item in unavailable] - or ["No relevant repository quality check was identified."], - ) - elif baseline: - blocker = PreparationBlocker( - kind=BlockerKind.REPOSITORY_BASELINE, - summary="Required checks also fail on the unchanged repository.", - user_action=( - "Resolve the existing check failure or provide an authoritative " - "replacement check." - ), - details=[item.name for item in baseline], - ) - else: - blocker = PreparationBlocker( - kind=BlockerKind.SECURITY_EVIDENCE, - summary="The patch is prepared, but functional verification is incomplete.", - user_action=( - "Provide the missing test prerequisites or review the remaining " - "evidence gaps." - ), - details=gaps, - ) + gaps = list(dict.fromkeys([*record.repair.gaps, *verifier.gaps])) + if not gaps: + gaps = ["Independent review could not confirm the repair and native-test coverage."] return await finish( - PreparationState.BLOCKED, blocker.summary, blocker=blocker, gaps=gaps + 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, ) return await finish( PreparationState.FAILED, "The repair exhausted its configured attempts." @@ -766,7 +678,10 @@ async def prepare_fix( # noqa: PLR0915 return await finish(PreparationState.FAILED, "Fix preparation was cancelled.") except TimeoutError: return await finish(PreparationState.FAILED, "Fix preparation exceeded its time limit.") - except Exception as error: # noqa: BLE001 + except Exception as error: + 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, diff --git a/tests/test_fix_preparation.py b/tests/test_fix_preparation.py index fc3142b42..0985dfe34 100644 --- a/tests/test_fix_preparation.py +++ b/tests/test_fix_preparation.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +import json import subprocess import sys from typing import TYPE_CHECKING @@ -15,15 +16,14 @@ from strix.fix.contracts import ( CheckResult, CheckStatus, CommandSpec, - FileManifestEntry, FixCandidateV1, FixEdit, FixPreparationRequestV1, - PreparationBlocker, PreparationState, RegressionTestResult, RepairOutcome, RepairStatus, + RepositoryTestPlan, ReproductionSpec, SourceIdentity, SourceIdentityKind, @@ -32,11 +32,10 @@ from strix.fix.contracts import ( VerifierResult, candidate_from_legacy_report, ) -from strix.fix.evidence import command_status +from strix.fix.evidence import command_status, passed_test_count, record_test_execution from strix.fix.locations import AnchorStatus, anchor_location from strix.fix.prepare import ( PreparationContext, - _matches_baseline_failure, _network_isolation_prefix, build_git_manifest, build_git_patch, @@ -170,7 +169,18 @@ async def _noop_repair( path.read_text(encoding="utf-8").replace("'unsafe'", "'safe'"), encoding="utf-8", ) + (context.workspace / "test_app.py").write_text( + "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"] + ), + no_unit_tests_reason="The fixture has no prior unit suite.", + ), status=RepairStatus.COMPLETE, summary="The repository fix is ready for independent evaluation.", ) @@ -184,7 +194,10 @@ async def _verified( decision=VerificationDecision.VERIFIED, summary="The invariant is closed.", security_invariant_closed=True, - reproduction_executed=True, + review_basis="code_review", + regression_test_valid=True, + unit_test_coverage_valid=True, + reproduction_executed=False, reproduction_summary="The vulnerable input is rejected.", sibling_paths_reviewed=["app.py"], preserved_behaviors=["The module compiles."], @@ -283,526 +296,6 @@ def test_anchor_location_detects_stale_file_digest(tmp_path: Path) -> None: assert anchor_location(tmp_path, edit).status is AnchorStatus.STALE -@pytest.mark.asyncio -async def test_prepare_fix_returns_ready_with_manifest(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=_verified, - ) - - assert result.state is PreparationState.READY - assert result.attempts == 1 - assert [entry.path for entry in result.final_file_manifest] == ["app.py"] - assert result.security_reproduction is not None - assert result.security_reproduction.target is VerificationTarget.PATCHED - - -@pytest.mark.asyncio -async def test_prepare_fix_does_not_apply_draft_before_repair(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - observed_source: list[str] = [] - - async def repair( - context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - observed_source.append((context.workspace / "app.py").read_text(encoding="utf-8")) - (context.workspace / "app.py").write_text( - "def result():\n return 'safe'\n", - encoding="utf-8", - ) - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="Implemented the repair from repository context.", - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=_verified, - ) - - assert result.state is PreparationState.READY - assert observed_source == ["def result():\n return 'unsafe'\n"] - - -@pytest.mark.asyncio -async def test_quality_gate_retries_once_before_verification(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - compile_calls = 0 - repair_calls = 0 - verification_calls = 0 - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - nonlocal compile_calls - compile_calls += 1 - failed = compile_calls == 1 - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.FAILED if failed else CheckStatus.PASSED, - exit_code=1 if failed else 0, - duration_seconds=0, - required=command.required, - ) - - async def repair( - context: PreparationContext, - checks: list[CheckResult], - ) -> RepairOutcome: - nonlocal repair_calls - repair_calls += 1 - assert checks == [] if repair_calls == 1 else checks[0].status is CheckStatus.FAILED - (context.workspace / "app.py").write_text( - "def result():\n return 'safe'\n", - encoding="utf-8", - ) - return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.") - - async def verify( - context: PreparationContext, - checks: list[CheckResult], - ) -> VerifierResult: - nonlocal verification_calls - verification_calls += 1 - return await _verified(context, checks) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=verify, - command_runner=runner, - ) - - assert result.state is PreparationState.READY - assert repair_calls == 2 - assert compile_calls == 2 - assert verification_calls == 1 - - -@pytest.mark.asyncio -async def test_failed_quality_gate_never_runs_security_verifier(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - verifier_called = False - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.FAILED, - exit_code=1, - duration_seconds=0, - required=command.required, - ) - - async def verify( - context: PreparationContext, - checks: list[CheckResult], - ) -> VerifierResult: - nonlocal verifier_called - verifier_called = True - return await _verified(context, checks) - - result = await prepare_fix( - _request(_candidate(commit), attempts=1), - workspace, - repair=_noop_repair, - verify=verify, - command_runner=runner, - ) - - assert result.state is PreparationState.FAILED - assert verifier_called is False - assert "quality gate" in result.stop_reason - - -@pytest.mark.asyncio -async def test_repository_baseline_failure_is_a_typed_blocker( - tmp_path: Path, -) -> None: - workspace, commit = _workspace(tmp_path) - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.FAILED, - exit_code=1, - duration_seconds=0, - output="existing failure", - required=command.required, - baseline_status=CheckStatus.FAILED, - baseline_exit_code=1, - baseline_output="existing failure", - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=_verified, - command_runner=runner, - ) - - assert result.state is PreparationState.BLOCKED - assert result.blocker is not None - assert result.blocker.kind is BlockerKind.REPOSITORY_BASELINE - assert result.attempts == 1 - - -@pytest.mark.asyncio -async def test_different_candidate_and_baseline_failures_are_repairable( - tmp_path: Path, -) -> None: - workspace, commit = _workspace(tmp_path) - repairs = 0 - - async def repair( - _context: PreparationContext, - _feedback: list[CheckResult], - ) -> RepairOutcome: - nonlocal repairs - repairs += 1 - return await _noop_repair(_context, _feedback) - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.FAILED, - exit_code=1, - duration_seconds=0, - output="candidate-specific failure", - required=command.required, - baseline_status=CheckStatus.FAILED, - baseline_exit_code=1, - baseline_output="different pre-existing failure", - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=_verified, - command_runner=runner, - ) - - assert result.state is PreparationState.FAILED - assert result.blocker is None - assert result.attempts == 2 - assert repairs == 2 - - -@pytest.mark.asyncio -async def test_unavailable_required_check_is_a_typed_blocker(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.UNAVAILABLE, - duration_seconds=0, - output="compiler unavailable", - required=command.required, - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=_verified, - command_runner=runner, - ) - - assert result.state is PreparationState.BLOCKED - assert result.blocker is not None - assert result.blocker.kind is BlockerKind.ENVIRONMENT - assert result.attempts == 1 - - -@pytest.mark.asyncio -async def test_repairable_security_rejection_gets_one_retry(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - repair_feedback: list[list[str]] = [] - verifier_calls = 0 - - async def repair( - context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - repair_feedback.append( - [ - gap - for attempt in context.feedback - if attempt.verifier is not None - for gap in attempt.verifier.gaps - ] - ) - (context.workspace / "app.py").write_text( - "def result():\n return 'safe'\n", - encoding="utf-8", - ) - return RepairOutcome(status=RepairStatus.COMPLETE, summary="Repair complete.") - - async def verify( - context: PreparationContext, - checks: list[CheckResult], - ) -> VerifierResult: - nonlocal verifier_calls - verifier_calls += 1 - if verifier_calls == 1: - return VerifierResult( - decision=VerificationDecision.REJECTED, - summary="A sibling path remains vulnerable.", - gaps=["Harden the sibling path."], - repairable=True, - ) - return await _verified(context, checks) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=verify, - ) - - assert result.state is PreparationState.READY - assert result.attempts == 2 - assert repair_feedback == [[], ["Harden the sibling path."]] - - -@pytest.mark.asyncio -async def test_security_blocker_stops_without_repair_retry(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - repair_calls = 0 - blocker = PreparationBlocker( - kind=BlockerKind.EXTERNAL_CONFIGURATION, - summary="A production account mapping is unavailable.", - user_action="Provide the production account mapping.", - ) - - async def repair( - context: PreparationContext, - checks: list[CheckResult], - ) -> RepairOutcome: - nonlocal repair_calls - repair_calls += 1 - return await _noop_repair(context, checks) - - async def verify( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.INCONCLUSIVE, - summary=blocker.summary, - blocker=blocker, - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=verify, - ) - - assert result.state is PreparationState.BLOCKED - assert result.blocker == blocker - assert repair_calls == 1 - - -@pytest.mark.asyncio -async def test_budget_exhaustion_can_be_approved_by_independent_evidence( - tmp_path: Path, -) -> None: - workspace, commit = _workspace(tmp_path) - verifier_called = False - - async def exhausted( - context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - (context.workspace / "app.py").write_text( - "def result():\n return 'safe'\n", - encoding="utf-8", - ) - return RepairOutcome( - status=RepairStatus.BUDGET_EXHAUSTED, - summary="The repair hit its emergency turn ceiling after editing the patch.", - turns_used=40, - ) - - async def verify( - context: PreparationContext, - checks: list[CheckResult], - ) -> VerifierResult: - nonlocal verifier_called - verifier_called = True - return await _verified(context, checks) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=exhausted, - verify=verify, - ) - - assert result.state is PreparationState.READY - assert verifier_called is True - assert result.attempt_history[0].repair.turns_used == 40 - assert result.changed_files == ["app.py"] - - -@pytest.mark.asyncio -async def test_budget_exhaustion_with_incomplete_evidence_is_blocked(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def exhausted( - context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - (context.workspace / "app.py").write_text( - "def result():\n return 'safe'\n", - encoding="utf-8", - ) - return RepairOutcome( - status=RepairStatus.BUDGET_EXHAUSTED, - summary="The repair hit its emergency turn ceiling after editing the patch.", - turns_used=40, - ) - - async def inconclusive( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.INCONCLUSIVE, - summary="Production configuration could not be verified.", - security_invariant_closed=False, - gaps=["The production environment value is unavailable."], - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=exhausted, - verify=inconclusive, - ) - - assert result.state is PreparationState.BLOCKED - assert result.blocker is not None - assert result.blocker.kind is BlockerKind.SECURITY_EVIDENCE - assert result.attempt_history[0].repair.turns_used == 40 - - -@pytest.mark.asyncio -async def test_prepare_fix_preserves_explicit_blocked_outcome(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - blocker = PreparationBlocker( - kind=BlockerKind.CREDENTIAL, - summary="A required credential is unavailable.", - user_action="Provide a repository-scoped test credential.", - ) - - async def blocked( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - return RepairOutcome( - status=RepairStatus.BLOCKED, - summary=blocker.summary, - blocker=blocker, - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=blocked, - verify=_verified, - ) - - assert result.state is PreparationState.BLOCKED - assert result.blocker == blocker - assert result.verifier is None - - -@pytest.mark.asyncio -async def test_prepare_fix_requires_verifier_security_test(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def verify_without_test( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.VERIFIED, - summary="The verifier did not execute the issue-specific test.", - security_invariant_closed=True, - reproduction_executed=False, - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=verify_without_test, - ) - - assert result.state is PreparationState.BLOCKED - assert "paired functional regression" in " ".join(result.gaps) - - -@pytest.mark.asyncio -async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def incomplete( - context: PreparationContext, - checks: list[CheckResult], - ) -> VerifierResult: - verified = await _verified(context, checks) - return verified.model_copy(update={"security_invariant_closed": False}) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=incomplete, - ) - - assert result.state is PreparationState.BLOCKED - assert "invariant" in " ".join(result.gaps) - - -@pytest.mark.asyncio -async def test_prepare_fix_requires_a_source_change(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def no_change( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - return RepairOutcome(status=RepairStatus.COMPLETE, summary="No change.") - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=no_change, - verify=_verified, - ) - - assert result.state is PreparationState.FAILED - assert result.final_file_manifest == [] - assert "change repository source" in result.stop_reason - - @pytest.mark.asyncio async def test_prepare_fix_rejects_wrong_source_commit(tmp_path: Path) -> None: workspace, _commit = _workspace(tmp_path) @@ -819,40 +312,6 @@ async def test_prepare_fix_rejects_wrong_source_commit(tmp_path: Path) -> None: assert result.blocker.kind is BlockerKind.SOURCE -@pytest.mark.asyncio -async def test_prepare_fix_does_not_apply_escaping_draft_edit(tmp_path: Path) -> None: - workspace, _commit = _workspace(tmp_path) - outside = tmp_path / "outside.txt" - outside.write_text("target\n", encoding="utf-8") - (workspace / "link.txt").symlink_to(outside) - _git(workspace, "add", "link.txt") - _git(workspace, "commit", "-m", "add link") - candidate = _candidate(_git(workspace, "rev-parse", "HEAD")).model_copy( - update={ - "draft_edits": [ - FixEdit( - file="link.txt", - start_line=1, - end_line=1, - before="target", - after="safe", - ) - ] - } - ) - - result = await prepare_fix( - _request(candidate), - workspace, - repair=_noop_repair, - verify=_verified, - ) - - assert result.state is PreparationState.READY - assert outside.read_text(encoding="utf-8") == "target\n" - assert "link.txt" not in result.changed_files - - @pytest.mark.asyncio async def test_prepare_fix_rejects_dirty_workspace(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) @@ -947,68 +406,6 @@ async def test_manifest_patch_includes_untracked_companion_file(tmp_path: Path) _git(workspace, "diff", "--check") -def test_baseline_comparison_normalizes_only_known_checkout_roots() -> None: - result = CheckResult( - name="typecheck", - argv=["tsc"], - status=CheckStatus.FAILED, - exit_code=2, - duration_seconds=0, - output="/workspace/source/a.ts: TS100", - workspace_root="/workspace/source", - baseline_status=CheckStatus.FAILED, - baseline_exit_code=2, - baseline_output="/workspace/base/a.ts: TS100", - baseline_workspace_root="/workspace/base", - ) - assert _matches_baseline_failure(result) - assert not _matches_baseline_failure( - result.model_copy(update={"output": "/workspace/source/a.ts: TS200"}) - ) - - -@pytest.mark.asyncio -async def test_check_planner_sees_companion_files_and_preserves_partial_evidence( - tmp_path: Path, -) -> None: - workspace, commit = _workspace(tmp_path) - seen: list[str] = [] - - async def repair(context: PreparationContext, checks: list[CheckResult]) -> RepairOutcome: - (workspace / "companion.py").write_text("guard = True\n") - return await _noop_repair(context, checks) - - async def planner( - _context: PreparationContext, - manifest: list[FileManifestEntry], - ) -> list[CommandSpec]: - seen.extend(item.path for item in manifest) - return [CommandSpec(name="native tests", argv=["missing-runtime", "test"])] - - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - return CheckResult( - name=command.name, - argv=command.argv, - status=CheckStatus.UNAVAILABLE, - exit_code=127, - duration_seconds=0, - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=repair, - verify=_verified, - check_planner=planner, - command_runner=runner, - ) - assert set(seen) == {"app.py", "companion.py"} - assert result.state is PreparationState.BLOCKED - assert result.verifier and result.verifier.regression_tests[0].passed() - assert len(result.final_file_manifest) == 2 - assert result.attempts == 1 - - def test_skipped_check_and_missing_runtime_are_not_passing_evidence() -> None: assert command_status(0, "Skipping to avoid parser lock")[0] is CheckStatus.SKIPPED assert command_status(127, "bun: command not found")[0] is CheckStatus.UNAVAILABLE @@ -1069,73 +466,212 @@ def test_regression_requires_consistent_execution_provenance() -> None: @pytest.mark.asyncio -async def test_late_setup_refreshes_earlier_checks_before_verification(tmp_path: Path) -> None: +async def test_native_flow_runs_regression_before_review_and_saves_test(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - request = _request(_candidate(commit)) - request.checks.append(CommandSpec(name="native tests", argv=["tests"])) - epoch = 0 - calls: list[str] = [] - async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: - nonlocal epoch - calls.append(command.name) - if command.name == "native tests": - epoch = 1 + 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 = [] + + async def repair(context, checks): + calls.append(context.attempt) + 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 + + +@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: + 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, - environment_id=f"execution:{epoch}", - source_digest="a" * 64, ) - async def verify(context: PreparationContext, checks: list[CheckResult]) -> VerifierResult: - assert {item.environment_id for item in checks} == {"execution:1"} - result = await _verified(context, checks) - for leg in ( - result.regression_tests[0].base, - result.regression_tests[0].patched, - result.regression_tests[0].behavior, - ): - leg.environment_id = "execution:1" - leg.source_digest = "a" * 64 - return result + async def review(*_args): + pytest.fail("Review must not run without passing tests") result = await prepare_fix( - request, workspace, repair=_noop_repair, verify=verify, command_runner=runner + _request(_candidate(commit)), + workspace, + repair=_noop_repair, + verify=review, + command_runner=runner, ) - assert result.state is PreparationState.READY - assert calls == ["compile", "native tests", "compile"] + 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_harness_failure_retains_patch_and_does_not_blame_customer(tmp_path: Path) -> None: +async def test_reviewer_concrete_defect_retries_and_reruns_tests(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - async def broken_verifier( - _context: PreparationContext, _checks: list[CheckResult] - ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.INCONCLUSIVE, - summary="Cannot execute verifier test.", - security_tests=[ - CheckResult( - name="test", - argv=["python", "test.py"], - status=CheckStatus.FAILED, - exit_code=1, - duration_seconds=0, - failure_kind="unknown", - ) - ], - ) + 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=broken_verifier + _request(_candidate(commit)), workspace, repair=_noop_repair, verify=review ) - assert result.state is PreparationState.BLOCKED - assert result.blocker.kind is BlockerKind.VERIFICATION_RUNTIME + 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) + + +@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( + "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 + ) + assert result.state in {PreparationState.BLOCKED, PreparationState.FAILED} assert result.final_file_manifest - assert "prerequisite" not in result.blocker.user_action + + +@pytest.mark.parametrize( + ("output", "expected"), + [ + ("Tests: 2 passed, 2 total", 2), + ("2 pass\n0 fail", 2), + ("# tests 2\n# pass 2\n# fail 0", 2), + ("Ran 3 tests in 0.1s\n\nOK (skipped=1)", 2), + ("3 examples, 0 failures, 1 pending", 2), + ("--- PASS: TestSafe (0.00s)", 1), + ("Ran 2 tests in 0.1s\n\nOK (skipped=2)", 0), + ("No tests found", None), + ], +) +def test_runner_summaries_record_actual_execution(output: str, expected: int | None) -> None: + + assert passed_test_count(output) == expected + + +def test_new_command_metadata_does_not_change_existing_finding_digest(tmp_path: Path) -> None: + + _root, commit = _workspace(tmp_path) + candidate = _candidate(commit) + payload = candidate.model_dump(mode="json") + payload.pop("finding") + payload["reproduction"]["command"].pop("purpose") + previous = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + assert candidate.digest() == previous + + +def test_unknown_runner_format_preserves_exit_result_for_independent_review() -> None: + + command = CommandSpec(name="custom runner", argv=["./tests/run"], purpose="unit") + result = record_test_execution( + CheckResult( + name=command.name, + argv=command.argv, + status=CheckStatus.PASSED, + exit_code=0, + duration_seconds=1, + output="All project assertions completed successfully.", + ), + command, + ) + assert result.status is CheckStatus.PASSED + assert result.tests_passed is None