From df313c14386947af0a760e3d8f8281fe175d1b29 Mon Sep 17 00:00:00 2001 From: yoni Date: Mon, 28 Sep 2026 12:13:19 +0000 Subject: [PATCH] refactor fix preparation gates --- strix/fix/__init__.py | 6 + strix/fix/contracts.py | 35 ++- strix/fix/prepare.py | 398 ++++++++++++------------ tests/test_fix_preparation.py | 558 +++++++++++++++------------------- 4 files changed, 497 insertions(+), 500 deletions(-) diff --git a/strix/fix/__init__.py b/strix/fix/__init__.py index d368b155e..4c2936e9a 100644 --- a/strix/fix/__init__.py +++ b/strix/fix/__init__.py @@ -1,6 +1,7 @@ """Verified fix preparation contracts and runtime.""" from strix.fix.contracts import ( + BlockerKind, CandidateLocation, CheckResult, CheckStatus, @@ -11,6 +12,7 @@ from strix.fix.contracts import ( FixPreparationAttempt, FixPreparationRequestV1, FixPreparationResultV1, + PreparationBlocker, PreparationState, RepairOutcome, RepairStatus, @@ -18,6 +20,7 @@ from strix.fix.contracts import ( SourceIdentity, SourceIdentityKind, VerificationDecision, + VerificationTarget, VerifierResult, candidate_from_legacy_report, ) @@ -31,6 +34,7 @@ from strix.fix.prepare import ( __all__ = [ + "BlockerKind", "CandidateLocation", "CheckResult", "CheckStatus", @@ -41,6 +45,7 @@ __all__ = [ "FixPreparationAttempt", "FixPreparationRequestV1", "FixPreparationResultV1", + "PreparationBlocker", "PreparationCancelledError", "PreparationContext", "PreparationPolicy", @@ -51,6 +56,7 @@ __all__ = [ "SourceIdentity", "SourceIdentityKind", "VerificationDecision", + "VerificationTarget", "VerifierResult", "build_git_manifest", "candidate_from_legacy_report", diff --git a/strix/fix/contracts.py b/strix/fix/contracts.py index 950e87db2..b8f1f4e90 100644 --- a/strix/fix/contracts.py +++ b/strix/fix/contracts.py @@ -27,8 +27,6 @@ class SourceIdentityKind(StrEnum): class PreparationState(StrEnum): PREPARING = "preparing" READY = "ready" - READY_WITH_GAPS = "ready_with_gaps" - NEEDS_REVIEW = "needs_review" BLOCKED = "blocked" FAILED = "failed" STALE = "stale" @@ -41,6 +39,11 @@ class CheckStatus(StrEnum): CANCELLED = "cancelled" +class VerificationTarget(StrEnum): + BASE = "base" + PATCHED = "patched" + + class VerificationDecision(StrEnum): VERIFIED = "verified" REJECTED = "rejected" @@ -54,6 +57,22 @@ class RepairStatus(StrEnum): INCOMPLETE = "incomplete" +class BlockerKind(StrEnum): + SOURCE = "source" + ENVIRONMENT = "environment" + REPOSITORY_BASELINE = "repository_baseline" + CREDENTIAL = "credential" + EXTERNAL_CONFIGURATION = "external_configuration" + SECURITY_EVIDENCE = "security_evidence" + + +class PreparationBlocker(ContractModel): + kind: BlockerKind + summary: str = Field(min_length=1) + user_action: str = Field(min_length=1) + details: list[str] = [] + + class SourceIdentity(ContractModel): kind: SourceIdentityKind value: str = Field(min_length=1) @@ -161,7 +180,7 @@ class FixPreparationRequestV1(ContractModel): repository_id: str | None = None candidate: FixCandidateV1 checks: list[CommandSpec] = [] - max_repair_attempts: int = Field(default=4, ge=1, le=4) + max_repair_attempts: int = Field(default=2, ge=1, le=2) timeout_seconds: int = Field(default=1800, ge=30, le=14400) network_allowed: bool = False credentials_allowed: list[str] = [] @@ -175,6 +194,9 @@ class CheckResult(ContractModel): duration_seconds: float = Field(ge=0) output: str = "" required: bool = True + target: VerificationTarget | None = None + baseline_status: CheckStatus | None = None + baseline_output: str | None = None class VerifierResult(ContractModel): @@ -186,6 +208,9 @@ class VerifierResult(ContractModel): sibling_paths_reviewed: list[str] = [] preserved_behaviors: list[str] = [] gaps: list[str] = [] + security_tests: list[CheckResult] = [] + repairable: bool = False + blocker: PreparationBlocker | None = None class RepairOutcome(ContractModel): @@ -194,6 +219,7 @@ class RepairOutcome(ContractModel): gaps: list[str] = [] reproduction_command: CommandSpec | None = None turns_used: int = Field(default=0, ge=0) + blocker: PreparationBlocker | None = None class FixPreparationAttempt(ContractModel): @@ -201,7 +227,7 @@ class FixPreparationAttempt(ContractModel): repair: RepairOutcome checks: list[CheckResult] = [] security_reproduction: CheckResult | None = None - verifier: VerifierResult + verifier: VerifierResult | None = None workspace_digest: str = Field(pattern=r"^[0-9a-f]{64}$") @@ -231,6 +257,7 @@ class FixPreparationResultV1(ContractModel): verifier: VerifierResult | None = None attempt_history: list[FixPreparationAttempt] = [] gaps: list[str] = [] + blocker: PreparationBlocker | None = None attempts: int = Field(default=0, ge=0) elapsed_seconds: float = Field(default=0, ge=0) cost_usd: float | None = Field(default=None, ge=0) diff --git a/strix/fix/prepare.py b/strix/fix/prepare.py index ab4039743..ecbe1562e 100644 --- a/strix/fix/prepare.py +++ b/strix/fix/prepare.py @@ -15,6 +15,7 @@ from pathlib import Path from typing import Literal from strix.fix.contracts import ( + BlockerKind, CheckResult, CheckStatus, CommandSpec, @@ -23,13 +24,15 @@ from strix.fix.contracts import ( FixPreparationAttempt, FixPreparationRequestV1, FixPreparationResultV1, + PreparationBlocker, PreparationState, RepairOutcome, RepairStatus, VerificationDecision, + VerificationTarget, VerifierResult, ) -from strix.fix.locations import AnchorStatus, anchor_candidate +from strix.fix.locations import AnchorStatus, anchor_location class PreparationCancelledError(RuntimeError): @@ -43,13 +46,13 @@ CancellationCheck = Callable[[], bool] @dataclass(slots=True) class PreparationPolicy: - max_repair_attempts: int = 4 + max_repair_attempts: int = 2 timeout_seconds: int = 1800 max_output_chars: int = 20000 def __post_init__(self) -> None: - if not 1 <= self.max_repair_attempts <= 4: - raise ValueError("max_repair_attempts must be between 1 and 4") + if not 1 <= self.max_repair_attempts <= 2: + raise ValueError("max_repair_attempts must be between 1 and 2") @dataclass(slots=True) @@ -66,7 +69,7 @@ RepairAgent = Callable[ Awaitable[RepairOutcome | None], ] IndependentVerifier = Callable[ - [PreparationContext, list[CheckResult], CheckResult | None], + [PreparationContext, list[CheckResult]], Awaitable[VerifierResult], ] SourceVerifier = Callable[[PreparationContext], Awaitable[bool]] @@ -189,34 +192,6 @@ async def run_command( ) -def _apply_edits(workspace: Path, candidate: FixCandidateV1) -> None: - by_file: dict[str, list[tuple[int, int, str, str]]] = {} - for edit in candidate.draft_edits: - by_file.setdefault(edit.file, []).append( - (edit.start_line, edit.end_line, edit.before, edit.after) - ) - - workspace_resolved = workspace.resolve() - for file_path, edits in by_file.items(): - path = (workspace / file_path).resolve() - if not path.is_relative_to(workspace_resolved): - raise ValueError(f"Draft edit path escapes the workspace: {file_path}") - content = path.read_text(encoding="utf-8") - lines = content.splitlines(keepends=True) - newline = "\r\n" if "\r\n" in content else "\n" - for start, end, before, after in sorted(edits, reverse=True): - original_segment = "".join(lines[start - 1 : end]) - actual = original_segment.rstrip("\r\n") - if actual != before.rstrip("\r\n"): - raise ValueError(f"Draft edit source changed at {file_path}:{start}-{end}") - replacement = after.splitlines(keepends=True) - preserve_newline = original_segment.endswith(("\n", "\r")) or end < len(lines) - if replacement and not replacement[-1].endswith(("\n", "\r")) and preserve_newline: - replacement[-1] += newline - lines[start - 1 : end] = replacement - path.write_text("".join(lines), encoding="utf-8", newline="") - - async def build_git_manifest( workspace: Path, ) -> tuple[list[FileManifestEntry], str, str | None]: @@ -368,6 +343,7 @@ def _result( diff_summary: str = "", artifact_ref: str | None = None, attempt_history: list[FixPreparationAttempt] | None = None, + blocker: PreparationBlocker | None = None, ) -> FixPreparationResultV1: return FixPreparationResultV1( state=state, @@ -384,6 +360,7 @@ def _result( verifier=verifier, attempt_history=attempt_history or [], gaps=gaps or [], + blocker=blocker, attempts=context.attempt, elapsed_seconds=time.monotonic() - started, ) @@ -405,23 +382,23 @@ def _required_checks_pass(checks: list[CheckResult]) -> bool: def _verification_passes( checks: list[CheckResult], - reproduction: CheckResult | None, verifier: VerifierResult, ) -> bool: return ( _required_checks_pass(checks) - and reproduction is not None - and reproduction.status is CheckStatus.PASSED and verifier.decision is VerificationDecision.VERIFIED and verifier.security_invariant_closed and verifier.reproduction_executed + and any( + result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED + for result in verifier.security_tests + ) ) def _verification_gaps( repair_outcome: RepairOutcome, checks: list[CheckResult], - reproduction: CheckResult | None, verifier: VerifierResult, ) -> list[str]: gaps = list(repair_outcome.gaps) @@ -438,58 +415,21 @@ def _verification_gaps( for result in checks if not result.required and result.status is not CheckStatus.PASSED ) - if reproduction is None: - gaps.append("No executable security reproduction was available.") - elif reproduction.status is not CheckStatus.PASSED: - gaps.append(f"{reproduction.name}: security reproduction {reproduction.status}") + if not verifier.security_tests: + gaps.append("The independent verifier did not execute a security test.") + elif not any( + result.target is VerificationTarget.PATCHED and result.status is CheckStatus.PASSED + for result in verifier.security_tests + ): + gaps.append("The independent verifier did not pass a security test on the patch.") + 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 _verification_fingerprint( - repair_outcome: RepairOutcome, - checks: list[CheckResult], - reproduction: CheckResult | None, - verifier: VerifierResult, -) -> str: - payload = { - "repair": { - "status": repair_outcome.status, - "gaps": repair_outcome.gaps, - }, - "checks": [ - { - "name": result.name, - "argv": result.argv, - "status": result.status, - "exit_code": result.exit_code, - "output": result.output, - "required": result.required, - } - for result in checks - ], - "reproduction": ( - { - "name": reproduction.name, - "argv": reproduction.argv, - "status": reproduction.status, - "exit_code": reproduction.exit_code, - "output": reproduction.output, - } - if reproduction is not None - else None - ), - "verifier": { - "decision": verifier.decision, - "security_invariant_closed": verifier.security_invariant_closed, - "reproduction_executed": verifier.reproduction_executed, - "gaps": verifier.gaps, - }, - } - encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() - return hashlib.sha256(encoded).hexdigest() - - async def prepare_fix( # noqa: PLR0915 request: FixPreparationRequestV1, workspace: Path, @@ -516,139 +456,237 @@ async def prepare_fix( # noqa: PLR0915 network_allowed=request.network_allowed, ) - async def execute() -> FixPreparationResultV1: # noqa: PLR0911 + async def execute() -> FixPreparationResultV1: # noqa: PLR0911, PLR0912, PLR0915 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 _result( context, state=PreparationState.STALE, - reason="The workspace does not match the recorded source identity.", + reason=blocker.summary, + blocker=blocker, started=started, ) - anchored, anchors = anchor_candidate(workspace, context.candidate) - edit_results = anchors[len(context.candidate.finding_locations) :] - if any(result.status is AnchorStatus.STALE for result in edit_results): - return _result( - context, - state=PreparationState.STALE, - reason="A draft edit does not match the recorded source.", - started=started, - ) + anchors = [ + anchor_location(workspace, location) for location in context.candidate.finding_locations + ] if any(result.status is not AnchorStatus.UNIQUE for result in anchors): - gaps = [ + details = [ f"{result.location.file}: {result.status}" for result in anchors if result.status is not AnchorStatus.UNIQUE ] + 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=details, + ) return _result( context, - state=PreparationState.NEEDS_REVIEW, - reason="One or more candidate locations could not be anchored uniquely.", - gaps=gaps, + state=PreparationState.BLOCKED, + reason=blocker.summary, + gaps=details, + blocker=blocker, started=started, ) - context.candidate = anchored - _apply_edits(workspace, context.candidate) + context.candidate = context.candidate.model_copy( + update={"finding_locations": [result.location for result in anchors]} + ) checks: list[CheckResult] = [] - reproduction: CheckResult | None = None verifier: VerifierResult | None = None attempt_history: list[FixPreparationAttempt] = [] - previous_workspace_digest: str | None = None - previous_verification_fingerprint: str | None = None - reproduction_command = ( - context.candidate.reproduction.command - if context.candidate.reproduction is not None - else None - ) for attempt in range(1, resolved_policy.max_repair_attempts + 1): context.attempt = attempt if cancelled(): raise PreparationCancelledError repair_outcome = _repair_outcome(await repair(context, checks)) - if repair_outcome.reproduction_command is not None: - reproduction_command = repair_outcome.reproduction_command + + if repair_outcome.status is RepairStatus.BLOCKED: + manifest, summary, artifact_ref = await manifest_builder(workspace) + blocker = repair_outcome.blocker or PreparationBlocker( + kind=BlockerKind.EXTERNAL_CONFIGURATION, + summary=repair_outcome.summary, + user_action=( + "Provide the missing repository, credential, or production " + "configuration prerequisite." + ), + details=repair_outcome.gaps, + ) + return _result( + context, + state=PreparationState.BLOCKED, + reason=blocker.summary, + gaps=repair_outcome.gaps, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + blocker=blocker, + started=started, + ) + + if repair_outcome.status is not RepairStatus.COMPLETE: + manifest, summary, artifact_ref = await manifest_builder(workspace) + return _result( + context, + state=PreparationState.FAILED, + reason=repair_outcome.summary, + gaps=repair_outcome.gaps, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + started=started, + ) + checks = [await runner(workspace, check) for check in request.checks] - reproduction = ( - await runner(workspace, reproduction_command) - if reproduction_command is not None - else None - ) - verifier = await verify(context, checks, reproduction) workspace_digest = await _workspace_digest(workspace) attempt_record = FixPreparationAttempt( attempt=attempt, repair=repair_outcome, checks=checks, - security_reproduction=reproduction, - verifier=verifier, workspace_digest=workspace_digest, ) attempt_history.append(attempt_record) context.feedback = list(attempt_history) - gaps = _verification_gaps(repair_outcome, checks, reproduction, verifier) - gaps = list(dict.fromkeys(gaps)) - verification_fingerprint = _verification_fingerprint( - repair_outcome, - checks, - reproduction, - verifier, - ) - if repair_outcome.status is RepairStatus.BLOCKED: + required = [result for result in checks if result.required] + unavailable = [ + result for result in required if result.status is CheckStatus.UNAVAILABLE + ] + if not required or unavailable: + details = ( + ["No required repository quality check was configured."] + if not required + else [f"{result.name}: {result.output}" for result in unavailable] + ) + blocker = PreparationBlocker( + kind=BlockerKind.ENVIRONMENT, + summary="The repository quality gate could not run.", + user_action=( + "Provide the missing tool or repository setup needed to run " + "the required checks." + ), + details=details, + ) manifest, summary, artifact_ref = await manifest_builder(workspace) return _result( context, state=PreparationState.BLOCKED, - reason=repair_outcome.summary, + reason=blocker.summary, checks=checks, - reproduction=reproduction, + gaps=blocker.details, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + blocker=blocker, + started=started, + ) + + failed = [result for result in required if result.status is CheckStatus.FAILED] + baseline_failures = [ + result for result in failed if result.baseline_status is CheckStatus.FAILED + ] + if baseline_failures: + blocker = PreparationBlocker( + kind=BlockerKind.REPOSITORY_BASELINE, + summary="Required checks also fail on the unchanged repository.", + user_action=( + "Repair the repository baseline or identify authoritative " + "replacement checks." + ), + details=[result.name for result in baseline_failures], + ) + manifest, summary, artifact_ref = await manifest_builder(workspace) + return _result( + context, + state=PreparationState.BLOCKED, + reason=blocker.summary, + checks=checks, + gaps=blocker.details, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + blocker=blocker, + started=started, + ) + if failed: + if attempt < resolved_policy.max_repair_attempts: + continue + manifest, summary, artifact_ref = await manifest_builder(workspace) + return _result( + context, + state=PreparationState.FAILED, + reason="The repair did not pass the repository quality gate.", + checks=checks, + gaps=[f"{result.name}: {result.output}" for result in failed], + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + started=started, + ) + + verifier = await verify(context, checks) + attempt_record.verifier = verifier + attempt_record.security_reproduction = next( + ( + result + for result in verifier.security_tests + if result.target is VerificationTarget.PATCHED + ), + None, + ) + context.feedback = list(attempt_history) + gaps = _verification_gaps(repair_outcome, checks, verifier) + + if verifier.blocker is not None: + manifest, summary, artifact_ref = await manifest_builder(workspace) + return _result( + context, + state=PreparationState.BLOCKED, + reason=verifier.blocker.summary, + checks=checks, + reproduction=attempt_record.security_reproduction, verifier=verifier, gaps=gaps, manifest=manifest, diff_summary=summary, artifact_ref=artifact_ref, attempt_history=attempt_history, + blocker=verifier.blocker, started=started, ) - if _verification_passes(checks, reproduction, verifier): + if _verification_passes(checks, verifier): manifest, summary, artifact_ref = await manifest_builder(workspace) if not manifest: return _result( context, - state=PreparationState.NEEDS_REVIEW, - reason="The prepared workspace does not contain a source change.", + state=PreparationState.FAILED, + reason="The repair did not change repository source.", checks=checks, - reproduction=reproduction, verifier=verifier, - gaps=[*gaps, "No prepared source change was produced."], - attempt_history=attempt_history, - started=started, - ) - if gaps: - return _result( - context, - state=PreparationState.READY_WITH_GAPS, - reason=("The fix passed available checks, but verification gaps remain."), - checks=checks, - reproduction=reproduction, - verifier=verifier, - gaps=gaps, - manifest=manifest, - diff_summary=summary, - artifact_ref=artifact_ref, + gaps=["No prepared source change was produced."], attempt_history=attempt_history, started=started, ) return _result( context, state=PreparationState.READY, - reason="The fix passed required checks and independent verification.", + reason="The fix passed repository checks and security verification.", checks=checks, - reproduction=reproduction, + reproduction=attempt_record.security_reproduction, verifier=verifier, manifest=manifest, diff_summary=summary, @@ -657,46 +695,32 @@ async def prepare_fix( # noqa: PLR0915 started=started, ) - if ( - previous_workspace_digest == workspace_digest - and previous_verification_fingerprint == verification_fingerprint - and verifier.decision is not VerificationDecision.INCONCLUSIVE - ): - manifest, summary, artifact_ref = await manifest_builder(workspace) - return _result( - context, - state=PreparationState.NEEDS_REVIEW, - reason="The repair made no repository progress after verification feedback.", - checks=checks, - reproduction=reproduction, - verifier=verifier, - gaps=[*gaps, "Two repair cycles produced the same repository state."], - manifest=manifest, - diff_summary=summary, - artifact_ref=artifact_ref, - attempt_history=attempt_history, - started=started, - ) - previous_workspace_digest = workspace_digest - previous_verification_fingerprint = verification_fingerprint + if verifier.repairable and attempt < resolved_policy.max_repair_attempts: + continue + + manifest, summary, artifact_ref = await manifest_builder(workspace) + return _result( + context, + state=PreparationState.FAILED, + reason="The independent security verifier did not approve the repair.", + checks=checks, + reproduction=attempt_record.security_reproduction, + verifier=verifier, + gaps=gaps, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact_ref, + attempt_history=attempt_history, + started=started, + ) - assert verifier is not None manifest, summary, artifact_ref = await manifest_builder(workspace) return _result( context, - state=PreparationState.NEEDS_REVIEW, - reason="The fix did not satisfy verification within the repair cycle limit.", + state=PreparationState.FAILED, + reason="The repair did not satisfy the required gates.", checks=checks, - reproduction=reproduction, verifier=verifier, - gaps=[ - *_verification_gaps( - attempt_history[-1].repair, - checks, - reproduction, - verifier, - ), - ], manifest=manifest, diff_summary=summary, artifact_ref=artifact_ref, diff --git a/tests/test_fix_preparation.py b/tests/test_fix_preparation.py index 9aa66a2a7..5b9fac767 100644 --- a/tests/test_fix_preparation.py +++ b/tests/test_fix_preparation.py @@ -10,14 +10,15 @@ from typing import TYPE_CHECKING import pytest from strix.fix.contracts import ( + BlockerKind, CandidateLocation, CheckResult, CheckStatus, CommandSpec, - FileManifestEntry, FixCandidateV1, FixEdit, FixPreparationRequestV1, + PreparationBlocker, PreparationState, RepairOutcome, RepairStatus, @@ -25,6 +26,7 @@ from strix.fix.contracts import ( SourceIdentity, SourceIdentityKind, VerificationDecision, + VerificationTarget, VerifierResult, candidate_from_legacy_report, ) @@ -105,7 +107,7 @@ def _candidate( ) -def _request(candidate: FixCandidateV1, *, attempts: int = 4) -> FixPreparationRequestV1: +def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationRequestV1: return FixPreparationRequestV1( scan_id="scan-1", finding_id="finding-1", @@ -125,19 +127,24 @@ def _request(candidate: FixCandidateV1, *, attempts: int = 4) -> FixPreparationR async def _noop_repair( - _context: PreparationContext, + context: PreparationContext, _checks: list[CheckResult], ) -> RepairOutcome: + path = context.workspace / "app.py" + if path.exists(): + path.write_text( + path.read_text(encoding="utf-8").replace("'unsafe'", "'safe'"), + encoding="utf-8", + ) return RepairOutcome( status=RepairStatus.COMPLETE, - summary="The draft is ready for independent evaluation.", + summary="The repository fix is ready for independent evaluation.", ) async def _verified( _context: PreparationContext, _checks: list[CheckResult], - _reproduction: CheckResult | None, ) -> VerifierResult: return VerifierResult( decision=VerificationDecision.VERIFIED, @@ -147,6 +154,16 @@ async def _verified( reproduction_summary="The vulnerable input is rejected.", sibling_paths_reviewed=["app.py"], preserved_behaviors=["The module compiles."], + security_tests=[ + CheckResult( + name="security reproduction", + argv=[sys.executable, "-c", "assert True"], + status=CheckStatus.PASSED, + exit_code=0, + duration_seconds=0, + target=VerificationTarget.PATCHED, + ) + ], ) @@ -234,74 +251,62 @@ def test_anchor_location_detects_stale_file_digest(tmp_path: Path) -> None: @pytest.mark.asyncio async def test_prepare_fix_returns_ready_with_manifest(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - reproduction = ReproductionSpec( - instructions="Confirm that result returns safe.", - command=CommandSpec( - name="security reproduction", - argv=[ - sys.executable, - "-c", - "from app import result; assert result() == 'safe'", - ], - ), - ) result = await prepare_fix( - _request(_candidate(commit, reproduction=reproduction)), + _request(_candidate(commit)), workspace, repair=_noop_repair, verify=_verified, ) assert result.state is PreparationState.READY - assert result.changed_files == ["app.py"] - assert result.final_file_manifest[0].operation == "modify" + 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.status is CheckStatus.PASSED - assert (workspace / "app.py").read_text(encoding="utf-8").endswith("return 'safe'\n") + assert result.security_reproduction.target is VerificationTarget.PATCHED @pytest.mark.asyncio -async def test_prepare_fix_allows_verified_multi_file_repairs( - tmp_path: Path, -) -> None: +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 widening_repair( - _context: PreparationContext, + async def repair( + context: PreparationContext, _checks: list[CheckResult], ) -> RepairOutcome: - (workspace / "hardening.py").write_text("HELPER = True\n", encoding="utf-8") + 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="Added the companion hardening module.", + summary="Implemented the repair from repository context.", ) result = await prepare_fix( _request(_candidate(commit)), workspace, - repair=widening_repair, + repair=repair, verify=_verified, ) assert result.state is PreparationState.READY - assert {entry.path for entry in result.final_file_manifest} == { - "app.py", - "hardening.py", - } - assert result.candidate.digest() == result.candidate_digest + assert observed_source == ["def result():\n return 'unsafe'\n"] @pytest.mark.asyncio -async def test_prepare_fix_retries_failed_checks(tmp_path: Path) -> None: +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 - if command.name == "compile": - compile_calls += 1 - failed = command.name == "compile" and compile_calls == 1 + compile_calls += 1 + failed = compile_calls == 1 return CheckResult( name=command.name, argv=command.argv, @@ -311,55 +316,45 @@ async def test_prepare_fix_retries_failed_checks(tmp_path: Path) -> None: required=command.required, ) - repair_calls = 0 - async def repair( - _context: PreparationContext, - _checks: list[CheckResult], + context: PreparationContext, + checks: list[CheckResult], ) -> RepairOutcome: nonlocal repair_calls repair_calls += 1 - if repair_calls == 2: - (workspace / "app.py").write_text( - "def result():\n return 'safe'\n# retry\n", - encoding="utf-8", - ) - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="Repair cycle complete.", + 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=_verified, + verify=verify, command_runner=runner, ) assert result.state is PreparationState.READY - assert result.attempts == 2 + assert repair_calls == 2 assert compile_calls == 2 - assert len(result.attempt_history) == 2 + assert verification_calls == 1 @pytest.mark.asyncio -async def test_prepare_fix_stops_at_repair_limit(tmp_path: Path) -> None: +async def test_failed_quality_gate_never_runs_security_verifier(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - repair_calls = 0 - - async def repair( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - nonlocal repair_calls - repair_calls += 1 - with (workspace / "app.py").open("a", encoding="utf-8") as handle: - handle.write(f"# attempt {repair_calls}\n") - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="Repair cycle complete.", - ) + verifier_called = False async def runner(_workspace: Path, command: CommandSpec) -> CheckResult: return CheckResult( @@ -371,46 +366,115 @@ async def test_prepare_fix_stops_at_repair_limit(tmp_path: Path) -> None: 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=4), + _request(_candidate(commit), attempts=1), workspace, - repair=repair, + 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_output="existing failure", + ) + + result = await prepare_fix( + _request(_candidate(commit)), + workspace, + repair=_noop_repair, verify=_verified, command_runner=runner, ) - assert result.state is PreparationState.NEEDS_REVIEW - assert result.attempts == 4 - assert len(result.attempt_history) == 4 - assert "cycle limit" in result.stop_reason + 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_prepare_fix_feeds_verifier_rejection_into_next_repair( - tmp_path: Path, -) -> None: +async def test_unavailable_required_check_is_a_typed_blocker(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - repair_calls = 0 + + 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: - nonlocal repair_calls - repair_calls += 1 - if repair_calls == 2: - assert context.feedback[0].verifier.gaps == ["Harden the sibling path."] - (workspace / "sibling.py").write_text("SAFE = True\n", encoding="utf-8") - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="Repair cycle complete.", + 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], - _reproduction: CheckResult | None, + context: PreparationContext, + checks: list[CheckResult], ) -> VerifierResult: nonlocal verifier_calls verifier_calls += 1 @@ -419,8 +483,9 @@ async def test_prepare_fix_feeds_verifier_rejection_into_next_repair( decision=VerificationDecision.REJECTED, summary="A sibling path remains vulnerable.", gaps=["Harden the sibling path."], + repairable=True, ) - return await _verified(_context, _checks, _reproduction) + return await _verified(context, checks) result = await prepare_fix( _request(_candidate(commit)), @@ -431,156 +496,96 @@ async def test_prepare_fix_feeds_verifier_rejection_into_next_repair( assert result.state is PreparationState.READY assert result.attempts == 2 - assert result.attempt_history[0].verifier.decision is VerificationDecision.REJECTED - assert result.attempt_history[1].verifier.decision is VerificationDecision.VERIFIED + assert repair_feedback == [[], ["Harden the sibling path."]] @pytest.mark.asyncio -async def test_prepare_fix_stops_after_repeated_repository_state(tmp_path: Path) -> None: +async def test_security_blocker_stops_without_repair_retry(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - - async def rejected( - _context: PreparationContext, - _checks: list[CheckResult], - _reproduction: CheckResult | None, - ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.REJECTED, - summary="The fix remains incomplete.", - gaps=["Change the implementation."], - ) - - result = await prepare_fix( - _request(_candidate(commit)), - workspace, - repair=_noop_repair, - verify=rejected, + repair_calls = 0 + blocker = PreparationBlocker( + kind=BlockerKind.EXTERNAL_CONFIGURATION, + summary="A production account mapping is unavailable.", + user_action="Provide the production account mapping.", ) - assert result.state is PreparationState.NEEDS_REVIEW - assert result.attempts == 2 - assert "no repository progress" in result.stop_reason - - -@pytest.mark.asyncio -async def test_prepare_fix_retries_when_unchanged_state_has_new_feedback( - tmp_path: Path, -) -> None: - workspace, commit = _workspace(tmp_path) - verifier_calls = 0 - async def repair( context: PreparationContext, - _checks: list[CheckResult], + checks: list[CheckResult], ) -> RepairOutcome: - if context.attempt == 3: - with (workspace / "app.py").open("a", encoding="utf-8") as handle: - handle.write("# deployment invariant\n") - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="Repair cycle complete.", - ) + nonlocal repair_calls + repair_calls += 1 + return await _noop_repair(context, checks) async def verify( _context: PreparationContext, _checks: list[CheckResult], - _reproduction: CheckResult | None, ) -> VerifierResult: - nonlocal verifier_calls - verifier_calls += 1 - if verifier_calls == 1: - return VerifierResult( - decision=VerificationDecision.REJECTED, - summary="The source guard is incomplete.", - gaps=["Inspect the deployment configuration."], - ) - if verifier_calls == 2: - return VerifierResult( - decision=VerificationDecision.REJECTED, - summary="The deployment invariant is not enforced.", - gaps=["Set and enforce the production environment marker."], - ) - return await _verified(_context, _checks, _reproduction) + return VerifierResult( + decision=VerificationDecision.INCONCLUSIVE, + summary=blocker.summary, + blocker=blocker, + ) result = await prepare_fix( - _request(_candidate(commit), attempts=4), + _request(_candidate(commit)), workspace, repair=repair, verify=verify, ) - assert result.state is PreparationState.READY - assert result.attempts == 3 - assert len(result.attempt_history) == 3 + assert result.state is PreparationState.BLOCKED + assert result.blocker == blocker + assert repair_calls == 1 @pytest.mark.asyncio -async def test_prepare_fix_retries_transient_inconclusive_verification( - tmp_path: Path, -) -> None: +async def test_budget_exhaustion_fails_without_verification(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - verifier_calls = 0 + 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], - reproduction: CheckResult | None, ) -> VerifierResult: - nonlocal verifier_calls - verifier_calls += 1 - if verifier_calls < 3: - return VerifierResult( - decision=VerificationDecision.INCONCLUSIVE, - summary="The independent verifier reached its turn limit.", - gaps=["Independent verification did not complete within 30 turns."], - ) - return await _verified(context, checks, reproduction) - - result = await prepare_fix( - _request(_candidate(commit), attempts=4), - workspace, - repair=_noop_repair, - verify=verify, - ) - - assert result.state is PreparationState.READY - assert result.attempts == 3 - assert [attempt.verifier.decision for attempt in result.attempt_history] == [ - VerificationDecision.INCONCLUSIVE, - VerificationDecision.INCONCLUSIVE, - VerificationDecision.VERIFIED, - ] - - -@pytest.mark.asyncio -async def test_prepare_fix_evaluates_budget_exhausted_patch(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - - async def exhausted( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - return RepairOutcome( - status=RepairStatus.BUDGET_EXHAUSTED, - summary="The repair agent reached its turn limit.", - turns_used=40, - ) + nonlocal verifier_called + verifier_called = True + return await _verified(context, checks) result = await prepare_fix( _request(_candidate(commit)), workspace, repair=exhausted, - verify=_verified, + verify=verify, ) - assert result.state is PreparationState.READY - assert result.attempt_history[0].repair.status is RepairStatus.BUDGET_EXHAUSTED - assert result.attempt_history[0].repair.turns_used == 40 + assert result.state is PreparationState.FAILED + assert verifier_called is False + assert result.changed_files == ["app.py"] @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, @@ -588,7 +593,8 @@ async def test_prepare_fix_preserves_explicit_blocked_outcome(tmp_path: Path) -> ) -> RepairOutcome: return RepairOutcome( status=RepairStatus.BLOCKED, - summary="The repair requires an unavailable generated source file.", + summary=blocker.summary, + blocker=blocker, ) result = await prepare_fix( @@ -599,69 +605,34 @@ async def test_prepare_fix_preserves_explicit_blocked_outcome(tmp_path: Path) -> ) assert result.state is PreparationState.BLOCKED - assert result.stop_reason == "The repair requires an unavailable generated source file." - assert result.attempts == 1 + assert result.blocker == blocker + assert result.verifier is None @pytest.mark.asyncio -async def test_prepare_fix_runs_repair_proposed_reproduction(tmp_path: Path) -> None: - workspace, commit = _workspace(tmp_path) - candidate = _candidate(commit).model_copy(update={"reproduction": None}) - - async def repair( - _context: PreparationContext, - _checks: list[CheckResult], - ) -> RepairOutcome: - return RepairOutcome( - status=RepairStatus.COMPLETE, - summary="The patch and reproduction are ready.", - reproduction_command=CommandSpec( - name="repair-proposed security reproduction", - argv=[ - sys.executable, - "-c", - "from app import result; assert result() == 'safe'", - ], - ), - ) - - result = await prepare_fix( - _request(candidate), - workspace, - repair=repair, - verify=_verified, - ) - - assert result.state is PreparationState.READY - assert result.security_reproduction is not None - assert result.security_reproduction.name == "repair-proposed security reproduction" - - -@pytest.mark.asyncio -async def test_prepare_fix_requires_independent_verifier_approval(tmp_path: Path) -> None: +async def test_prepare_fix_requires_verifier_security_test(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) - async def rejected( + async def verify_without_test( _context: PreparationContext, _checks: list[CheckResult], - _reproduction: CheckResult | None, ) -> VerifierResult: return VerifierResult( - decision=VerificationDecision.REJECTED, - summary="A sibling path remains vulnerable.", - gaps=["Review the sibling handler."], + 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=rejected, + verify=verify_without_test, ) - assert result.state is PreparationState.NEEDS_REVIEW - assert result.verifier is not None - assert result.verifier.decision is VerificationDecision.REJECTED + assert result.state is PreparationState.FAILED + assert "did not execute a security test" in " ".join(result.gaps) @pytest.mark.asyncio @@ -669,15 +640,11 @@ async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> workspace, commit = _workspace(tmp_path) async def incomplete( - _context: PreparationContext, - _checks: list[CheckResult], - _reproduction: CheckResult | None, + context: PreparationContext, + checks: list[CheckResult], ) -> VerifierResult: - return VerifierResult( - decision=VerificationDecision.VERIFIED, - summary="The local edit works, but the invariant is not closed.", - reproduction_executed=True, - ) + verified = await _verified(context, checks) + return verified.model_copy(update={"security_invariant_closed": False}) result = await prepare_fix( _request(_candidate(commit)), @@ -686,95 +653,68 @@ async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> verify=incomplete, ) - assert result.state is PreparationState.NEEDS_REVIEW + assert result.state is PreparationState.FAILED + 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 empty_manifest( - _workspace: Path, - ) -> tuple[list[FileManifestEntry], str, str | None]: - return [], "No changes.", None + 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=_noop_repair, + repair=no_change, verify=_verified, - manifest_builder=empty_manifest, ) - assert result.state is PreparationState.NEEDS_REVIEW - assert "source change" in result.stop_reason + 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) - candidate = _candidate("0" * 40) result = await prepare_fix( - _request(candidate), + _request(_candidate("0" * 40)), workspace, repair=_noop_repair, verify=_verified, ) assert result.state is PreparationState.STALE - assert "source identity" in result.stop_reason - - -def test_anchor_location_rejects_symlink_escape(tmp_path: Path) -> None: - outside = tmp_path / "outside.txt" - outside.write_text("target\n", encoding="utf-8") - root = tmp_path / "repo" - root.mkdir() - (root / "link.txt").symlink_to(outside) - location = CandidateLocation( - file="link.txt", - start_line=1, - end_line=1, - snippet="target", - ) - - assert anchor_location(root, location).status is AnchorStatus.MISSING - - -def test_anchor_location_treats_unreadable_file_as_missing(tmp_path: Path) -> None: - (tmp_path / "blob.bin").write_bytes(b"\x89PNG\r\n\x1a\n\x00\xff\xfe") - location = CandidateLocation( - file="blob.bin", - start_line=1, - end_line=1, - snippet="blob", - ) - - assert anchor_location(tmp_path, location).status is AnchorStatus.MISSING + assert result.blocker is not None + assert result.blocker.kind is BlockerKind.SOURCE @pytest.mark.asyncio -async def test_prepare_fix_rejects_edit_escaping_workspace(tmp_path: Path) -> None: +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") - commit = _git(workspace, "rev-parse", "HEAD") - candidate = FixCandidateV1( - source_identity=SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit), - security_invariant="Replace target.", - draft_edits=[ - FixEdit( - file="link.txt", - start_line=1, - end_line=1, - before="target", - after="safe", - ) - ], + 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( @@ -784,8 +724,9 @@ async def test_prepare_fix_rejects_edit_escaping_workspace(tmp_path: Path) -> No verify=_verified, ) - assert result.state is PreparationState.NEEDS_REVIEW + 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 @@ -801,7 +742,6 @@ async def test_prepare_fix_rejects_dirty_workspace(tmp_path: Path) -> None: ) assert result.state is PreparationState.STALE - assert "source identity" in result.stop_reason @pytest.mark.asyncio