From f213a7da6aaeb606a73e74c80c53fc9fd9a848db Mon Sep 17 00:00:00 2001 From: Jonathan Singer Date: Tue, 29 Sep 2026 03:53:31 -0400 Subject: [PATCH] Require functional fix evidence and preserve partial preparation work --- strix/fix/__init__.py | 8 + strix/fix/contracts.py | 68 ++++- strix/fix/evidence.py | 32 +++ strix/fix/locations.py | 4 + strix/fix/prepare.py | 512 ++++++++++++++++------------------ tests/test_fix_preparation.py | 154 +++++++++- 6 files changed, 500 insertions(+), 278 deletions(-) create mode 100644 strix/fix/evidence.py diff --git a/strix/fix/__init__.py b/strix/fix/__init__.py index 4c2936e9a..de4ef9321 100644 --- a/strix/fix/__init__.py +++ b/strix/fix/__init__.py @@ -7,6 +7,7 @@ from strix.fix.contracts import ( CheckStatus, CommandSpec, FileManifestEntry, + FindingContext, FixCandidateV1, FixEdit, FixPreparationAttempt, @@ -14,12 +15,14 @@ from strix.fix.contracts import ( FixPreparationResultV1, PreparationBlocker, PreparationState, + RegressionTestResult, RepairOutcome, RepairStatus, ReproductionSpec, SourceIdentity, SourceIdentityKind, VerificationDecision, + VerificationHarness, VerificationTarget, VerifierResult, candidate_from_legacy_report, @@ -29,6 +32,7 @@ from strix.fix.prepare import ( PreparationContext, PreparationPolicy, build_git_manifest, + build_git_patch, prepare_fix, ) @@ -40,6 +44,7 @@ __all__ = [ "CheckStatus", "CommandSpec", "FileManifestEntry", + "FindingContext", "FixCandidateV1", "FixEdit", "FixPreparationAttempt", @@ -50,15 +55,18 @@ __all__ = [ "PreparationContext", "PreparationPolicy", "PreparationState", + "RegressionTestResult", "RepairOutcome", "RepairStatus", "ReproductionSpec", "SourceIdentity", "SourceIdentityKind", "VerificationDecision", + "VerificationHarness", "VerificationTarget", "VerifierResult", "build_git_manifest", + "build_git_patch", "candidate_from_legacy_report", "prepare_fix", ] diff --git a/strix/fix/contracts.py b/strix/fix/contracts.py index b8f1f4e90..5aa292552 100644 --- a/strix/fix/contracts.py +++ b/strix/fix/contracts.py @@ -37,6 +37,7 @@ class CheckStatus(StrEnum): FAILED = "failed" UNAVAILABLE = "unavailable" CANCELLED = "cancelled" + SKIPPED = "skipped" class VerificationTarget(StrEnum): @@ -153,6 +154,13 @@ class ReportedCheck(ContractModel): executed: bool = False +class FindingContext(ContractModel): + title: str = "" + description: str = "" + evidence: str = "" + remediation: str = "" + + class FixCandidateV1(ContractModel): version: Literal["1"] = "1" source_identity: SourceIdentity | None = None @@ -162,10 +170,14 @@ class FixCandidateV1(ContractModel): reproduction: ReproductionSpec | None = None reported_checks: list[ReportedCheck] = [] known_gaps: list[str] = [] + finding: FindingContext | None = None def digest(self) -> str: + data = self.model_dump(mode="json") + if self.finding is None: + data.pop("finding", None) # Preserve digests for stored legacy candidates. payload = json.dumps( - self.model_dump(mode="json"), + data, sort_keys=True, separators=(",", ":"), ).encode() @@ -197,6 +209,45 @@ class CheckResult(ContractModel): target: VerificationTarget | None = None baseline_status: CheckStatus | None = None baseline_output: str | None = None + cwd: str = "." + baseline_exit_code: int | None = None + failure_kind: Literal["environment", "timeout", "check", "source_changed"] | None = None + workspace_root: str | None = None + baseline_workspace_root: str | None = None + + +class RegressionTestResult(ContractModel): + """One unchanged test run on both revisions, plus a legitimate-operation check.""" + + name: str + expected_base_failure: str = Field(min_length=1) + harness_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + base: CheckResult + patched: CheckResult + behavior: CheckResult + + def passed(self) -> bool: + return ( + self.base.target is VerificationTarget.BASE + and self.patched.target is VerificationTarget.PATCHED + and self.behavior.target is VerificationTarget.PATCHED + and self.base.argv == self.patched.argv + and self.base.cwd == self.patched.cwd + and self.base.status is CheckStatus.FAILED + and self.base.exit_code == 1 + and self.base.failure_kind in {None, "check"} + and self.expected_base_failure in self.base.output + and self.patched.status is CheckStatus.PASSED + and self.patched.exit_code == 0 + and self.behavior.status is CheckStatus.PASSED + and self.behavior.exit_code == 0 + ) + + +class VerificationHarness(ContractModel): + path: str + sha256: str = Field(pattern=r"^[0-9a-f]{64}$") + content: str class VerifierResult(ContractModel): @@ -211,6 +262,9 @@ class VerifierResult(ContractModel): security_tests: list[CheckResult] = [] repairable: bool = False blocker: PreparationBlocker | None = None + regression_tests: list[RegressionTestResult] = [] + harnesses: list[VerificationHarness] = [] + notes: list[str] = [] class RepairOutcome(ContractModel): @@ -220,6 +274,8 @@ class RepairOutcome(ContractModel): reproduction_command: CommandSpec | None = None turns_used: int = Field(default=0, ge=0) blocker: PreparationBlocker | None = None + checks: list[CommandSpec] = [] + command_results: list[CheckResult] = [] class FixPreparationAttempt(ContractModel): @@ -258,6 +314,7 @@ class FixPreparationResultV1(ContractModel): attempt_history: list[FixPreparationAttempt] = [] gaps: list[str] = [] blocker: PreparationBlocker | None = None + setup_checks: list[CheckResult] = [] attempts: int = Field(default=0, ge=0) elapsed_seconds: float = Field(default=0, ge=0) cost_usd: float | None = Field(default=None, ge=0) @@ -318,13 +375,18 @@ def candidate_from_legacy_report( ReportedCheck(name="reporting-agent verification", result=verification, executed=False) ) + reproduction = str(report.get("poc_description") or report.get("evidence") or "").strip() return FixCandidateV1( source_identity=source_identity, security_invariant=invariant, finding_locations=locations, draft_edits=edits, - reproduction=ReproductionSpec( - instructions=str(report.get("poc_description") or report.get("evidence") or invariant) + reproduction=ReproductionSpec(instructions=reproduction) if reproduction else None, + finding=FindingContext( + title=str(report.get("title") or ""), + description=str(report.get("description") or report.get("technical_analysis") or ""), + evidence=str(report.get("evidence") or report.get("poc_description") or ""), + remediation=str(report.get("remediation_steps") or ""), ), reported_checks=reported_checks, known_gaps=["The reporting-agent verification is not independent."], diff --git a/strix/fix/evidence.py b/strix/fix/evidence.py new file mode 100644 index 000000000..23f99d3ce --- /dev/null +++ b/strix/fix/evidence.py @@ -0,0 +1,32 @@ +"""Execution facts shared by local and managed fix preparation.""" + +from __future__ import annotations + +import re +from typing import Literal + +from strix.fix.contracts import CheckStatus + + +def command_status( + exit_code: int, output: str +) -> tuple[CheckStatus, Literal["environment", "check"] | None]: + """Classify launch/collection failures before interpreting an assertion result.""" + if exit_code in {126, 127} or ( + exit_code != 0 + and re.search( + r"(?:command not found|exec: \S+: not found|No module named |" + r"Cannot find module |ECONNREFUSED|Could not connect to server)", + output, + re.IGNORECASE, + ) + ): + return CheckStatus.UNAVAILABLE, "environment" + if re.search( + r"(?:Skipping to avoid parser lock|collected 0 items|" + r"No test files found|No tests found, exiting with code 0)", + output, + re.IGNORECASE, + ): + return CheckStatus.SKIPPED, None + return (CheckStatus.PASSED, None) if exit_code == 0 else (CheckStatus.FAILED, "check") diff --git a/strix/fix/locations.py b/strix/fix/locations.py index b401a6b49..d7a9e220e 100644 --- a/strix/fix/locations.py +++ b/strix/fix/locations.py @@ -61,6 +61,8 @@ def _read_anchor_source(root: Path, file: str) -> str | None: def anchor_location( root: Path, location: CandidateLocation | FixEdit, + *, + exact_source: bool = False, ) -> AnchorResult: content = _read_anchor_source(root, location.file) if content is None: @@ -78,6 +80,8 @@ def anchor_location( ): return AnchorResult(AnchorStatus.STALE, location) return AnchorResult(AnchorStatus.MISSING, location) + if len(matches) > 1 and exact_source and location.start_line in matches: + matches = (location.start_line,) if len(matches) > 1: return AnchorResult(AnchorStatus.AMBIGUOUS, location, matches) diff --git a/strix/fix/prepare.py b/strix/fix/prepare.py index 8b2c4d0bb..e609dbeb0 100644 --- a/strix/fix/prepare.py +++ b/strix/fix/prepare.py @@ -29,9 +29,9 @@ from strix.fix.contracts import ( RepairOutcome, RepairStatus, VerificationDecision, - VerificationTarget, VerifierResult, ) +from strix.fix.evidence import command_status from strix.fix.locations import AnchorStatus, anchor_location @@ -73,6 +73,7 @@ IndependentVerifier = Callable[ Awaitable[VerifierResult], ] SourceVerifier = Callable[[PreparationContext], Awaitable[bool]] +CheckPlanner = Callable[[PreparationContext, list[FileManifestEntry]], Awaitable[list[CommandSpec]]] _COMMAND_ENV_ALLOWLIST = frozenset( @@ -181,14 +182,18 @@ async def run_command( output=f"Timed out after {command.timeout_seconds} seconds.", required=command.required, ) + status, failure_kind = command_status(process.returncode or 0, output.decode(errors="replace")) return CheckResult( name=command.name, argv=command.argv, - status=CheckStatus.PASSED if process.returncode == 0 else CheckStatus.FAILED, + status=status, exit_code=process.returncode, duration_seconds=time.monotonic() - started, output=output.decode(errors="replace")[-20000:], required=command.required, + failure_kind=failure_kind, + cwd=command.cwd, + workspace_root=str(workspace.resolve()), ) @@ -277,17 +282,42 @@ async def build_git_manifest( ) index += 1 - summary_process = await asyncio.create_subprocess_exec( + summary = "\n".join(f"{entry.operation}: {entry.path}" for entry in entries) + return entries, summary, None + + +async def build_git_patch(workspace: Path, manifest: list[FileManifestEntry]) -> bytes: + """Include new files in the exact diff the reviewer and artifact consumer receive.""" + chunks: list[bytes] = [] + commands = [["git", "diff", "--binary", "HEAD", "--"]] + for entry in manifest: + if entry.operation == "add" and not await _tracked_in_index(workspace, entry.path): + commands.append( # noqa: PERF401 - requires sequential async index lookup + ["git", "diff", "--no-index", "--binary", "--", "/dev/null", entry.path] + ) + for argv in commands: + process = await asyncio.create_subprocess_exec( + *argv, cwd=workspace, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE + ) + output, error = await process.communicate() + if process.returncode not in {0, 1}: + raise RuntimeError(error.decode(errors="replace")) + chunks.append(output) + return b"".join(chunks) + + +async def _tracked_in_index(workspace: Path, path: str) -> bool: + process = await asyncio.create_subprocess_exec( "git", - "diff", - "--stat", + "ls-files", + "--error-unmatch", "--", + path, cwd=workspace, - stdout=asyncio.subprocess.PIPE, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) - summary, _ = await summary_process.communicate() - return entries, summary.decode(errors="replace").strip(), None + return await process.wait() == 0 async def _workspace_digest(workspace: Path) -> str: @@ -388,11 +418,10 @@ def _verification_passes( _required_checks_pass(checks) 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 - ) + and not verifier.gaps + and verifier.blocker is None + and bool(verifier.regression_tests) + and all(result.passed() for result in verifier.regression_tests) ) @@ -415,13 +444,12 @@ def _verification_gaps( for result in checks if not result.required and result.status is not CheckStatus.PASSED ) - 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 + if not verifier.regression_tests or not all( + item.passed() for item in verifier.regression_tests ): - gaps.append("The independent verifier did not pass a security test on the patch.") + 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: @@ -431,10 +459,22 @@ def _verification_gaps( def _matches_baseline_failure(result: CheckResult) -> bool: - if result.baseline_status is not CheckStatus.FAILED: + if result.baseline_status is not result.status or result.status not in { + CheckStatus.FAILED, + CheckStatus.UNAVAILABLE, + CheckStatus.SKIPPED, + }: return False - candidate_output = " ".join(result.output.split()) - baseline_output = " ".join((result.baseline_output or "").split()) + 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) @@ -447,6 +487,7 @@ async def prepare_fix( # noqa: PLR0915 command_runner: CommandRunner = run_command, manifest_builder: ManifestBuilder = build_git_manifest, source_verifier: SourceVerifier = _verify_source, + check_planner: CheckPlanner | None = None, cancelled: CancellationCheck = lambda: False, policy: PreparationPolicy | None = None, ) -> FixPreparationResultV1: @@ -456,15 +497,44 @@ async def prepare_fix( # noqa: PLR0915 timeout_seconds=request.timeout_seconds, ) context = PreparationContext(request=request, workspace=workspace, candidate=request.candidate) - runner: CommandRunner = command_runner + runner = command_runner if runner is run_command: runner = functools.partial( run_command, credentials_allowed=request.credentials_allowed, network_allowed=request.network_allowed, ) + checks: list[CheckResult] = [] + verifier: VerifierResult | None = None + reproduction: CheckResult | None = None + + async def finish( + state: PreparationState, + reason: str, + *, + blocker: PreparationBlocker | None = None, + gaps: list[str] | None = None, + ) -> FixPreparationResultV1: + manifest, summary, artifact = await manifest_builder(workspace) + return _result( + context, + state=state, + reason=reason, + started=started, + checks=checks, + verifier=verifier, + reproduction=reproduction, + manifest=manifest, + diff_summary=summary, + artifact_ref=artifact, + attempt_history=context.feedback, + blocker=blocker, + gaps=gaps, + ) async def execute() -> FixPreparationResultV1: # noqa: PLR0911, PLR0912, PLR0915 + nonlocal checks, verifier, reproduction + blocker: PreparationBlocker | None if cancelled(): raise PreparationCancelledError if not await source_verifier(context): @@ -480,276 +550,178 @@ async def prepare_fix( # noqa: PLR0915 blocker=blocker, started=started, ) - anchors = [ - anchor_location(workspace, location) for location in context.candidate.finding_locations + anchor_location(workspace, location, exact_source=True) + for location in context.candidate.finding_locations ] - if any(result.status is not AnchorStatus.UNIQUE for result in anchors): - details = [ - f"{result.location.file}: {result.status}" - for result in anchors - if result.status is not AnchorStatus.UNIQUE - ] + unresolved = [item for item in anchors if item.status is not AnchorStatus.UNIQUE] + if unresolved: blocker = PreparationBlocker( kind=BlockerKind.SOURCE, summary="The reported finding locations could not be resolved uniquely.", user_action="Refresh the finding or identify the affected source location.", - details=details, - ) - return _result( - context, - state=PreparationState.BLOCKED, - reason=blocker.summary, - gaps=details, - blocker=blocker, - started=started, + details=[f"{item.location.file}: {item.status}" for item in unresolved], ) + return await finish(PreparationState.BLOCKED, blocker.summary, blocker=blocker) context.candidate = context.candidate.model_copy( - update={"finding_locations": [result.location for result in anchors]} + update={"finding_locations": [item.location for item in anchors]} ) - - checks: list[CheckResult] = [] - verifier: VerifierResult | None = None - attempt_history: list[FixPreparationAttempt] = [] 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.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] - workspace_digest = await _workspace_digest(workspace) - attempt_record = FixPreparationAttempt( + verifier = None + reproduction = None + record = FixPreparationAttempt( attempt=attempt, - repair=repair_outcome, - checks=checks, - workspace_digest=workspace_digest, + repair=RepairOutcome(status=RepairStatus.INCOMPLETE, summary="Repair started."), + workspace_digest=await _workspace_digest(workspace), ) - attempt_history.append(attempt_record) - context.feedback = list(attempt_history) - - 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=blocker.summary, - checks=checks, - 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 _matches_baseline_failure(result)] - 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, verifier): - manifest, summary, artifact_ref = await manifest_builder(workspace) - if not manifest: - return _result( - context, - state=PreparationState.FAILED, - reason="The repair did not change repository source.", - checks=checks, - verifier=verifier, - gaps=["No prepared source change was produced."], - attempt_history=attempt_history, - started=started, + context.feedback.append(record) + record.repair = _repair_outcome(await repair(context, checks)) + record.workspace_digest = await _workspace_digest(workspace) + manifest, _, _ = await build_git_manifest(workspace) + if not manifest: + if record.repair.blocker: + return await finish( + PreparationState.BLOCKED, + record.repair.summary, + blocker=record.repair.blocker, + gaps=record.repair.gaps, ) - return _result( - context, - state=PreparationState.READY, - reason="The fix passed repository checks and security verification.", - checks=checks, - reproduction=attempt_record.security_reproduction, - verifier=verifier, - manifest=manifest, - diff_summary=summary, - artifact_ref=artifact_ref, - attempt_history=attempt_history, - started=started, + return await finish( + PreparationState.FAILED, "The repair did not change repository source." ) - - 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, + 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)) + required = [item for item in checks if item.required] + regressions = [ + item + for item in required + if item.status is CheckStatus.FAILED and not _matches_baseline_failure(item) + ] + unchanged_retry = ( + len(context.feedback) > 1 + and context.feedback[-2].workspace_digest == record.workspace_digest ) - - manifest, summary, artifact_ref = await manifest_builder(workspace) - return _result( - context, - state=PreparationState.FAILED, - reason="The repair did not satisfy the required gates.", - checks=checks, - verifier=verifier, - manifest=manifest, - diff_summary=summary, - artifact_ref=artifact_ref, - attempt_history=attempt_history, - started=started, + can_retry = ( + record.repair.status is RepairStatus.COMPLETE + and attempt < resolved_policy.max_repair_attempts + and not unchanged_retry + ) + if regressions: + 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], + ) + # Missing tools and pre-existing failures do not erase attainable security evidence. + 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.", + ) + 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, + ) + if record.repair.status is not RepairStatus.COMPLETE: + state = ( + PreparationState.BLOCKED + if record.repair.status is RepairStatus.BLOCKED + else PreparationState.FAILED + ) + return await finish( + state, + record.repair.summary, + blocker=record.repair.blocker, + gaps=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, + ) + 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: + if 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, + ) + return await finish( + PreparationState.BLOCKED, blocker.summary, blocker=blocker, gaps=gaps + ) + return await finish( + PreparationState.FAILED, "The repair exhausted its configured attempts." ) try: async with asyncio.timeout(resolved_policy.timeout_seconds): return await execute() except PreparationCancelledError: - return _result( - context, - state=PreparationState.FAILED, - reason="Fix preparation was cancelled.", - attempt_history=context.feedback, - started=started, - ) + return await finish(PreparationState.FAILED, "Fix preparation was cancelled.") except TimeoutError: - return _result( - context, - state=PreparationState.FAILED, - reason="Fix preparation exceeded its time limit.", - attempt_history=context.feedback, - started=started, + return await finish(PreparationState.FAILED, "Fix preparation exceeded its time limit.") + except Exception as error: # noqa: BLE001 + # Preserve partial work without exposing unredacted exception text. + return await finish( + PreparationState.FAILED, + f"Fix preparation stopped after {type(error).__name__}; partial work was retained.", ) diff --git a/tests/test_fix_preparation.py b/tests/test_fix_preparation.py index 4cb781772..eb0918c47 100644 --- a/tests/test_fix_preparation.py +++ b/tests/test_fix_preparation.py @@ -20,6 +20,7 @@ from strix.fix.contracts import ( FixPreparationRequestV1, PreparationBlocker, PreparationState, + RegressionTestResult, RepairOutcome, RepairStatus, ReproductionSpec, @@ -30,11 +31,14 @@ from strix.fix.contracts import ( VerifierResult, candidate_from_legacy_report, ) +from strix.fix.evidence import command_status 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, prepare_fix, run_command, ) @@ -44,6 +48,34 @@ if TYPE_CHECKING: from pathlib import Path +def _regression() -> RegressionTestResult: + base = CheckResult( + name="authorization", + argv=["python", "regression.py"], + status=CheckStatus.FAILED, + exit_code=1, + duration_seconds=0, + target=VerificationTarget.BASE, + output="unauthorized access allowed", + ) + passed = base.model_copy( + update={ + "status": CheckStatus.PASSED, + "exit_code": 0, + "target": VerificationTarget.PATCHED, + "output": "passed", + } + ) + return RegressionTestResult( + name="authorization", + expected_base_failure="unauthorized access allowed", + harness_sha256="a" * 64, + base=base, + patched=passed, + behavior=passed, + ) + + def _git(workspace: Path, *args: str) -> str: return subprocess.run( # noqa: S603 ["/usr/bin/git", *args], @@ -123,6 +155,7 @@ def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationR ) ], max_repair_attempts=attempts, + network_allowed=True, ) @@ -154,6 +187,7 @@ async def _verified( reproduction_summary="The vulnerable input is rejected.", sibling_paths_reviewed=["app.py"], preserved_behaviors=["The module compiles."], + regression_tests=[_regression()], security_tests=[ CheckResult( name="security reproduction", @@ -403,6 +437,7 @@ async def test_repository_baseline_failure_is_a_typed_blocker( output="existing failure", required=command.required, baseline_status=CheckStatus.FAILED, + baseline_exit_code=1, baseline_output="existing failure", ) @@ -445,6 +480,7 @@ async def test_different_candidate_and_baseline_failures_are_repairable( output="candidate-specific failure", required=command.required, baseline_status=CheckStatus.FAILED, + baseline_exit_code=1, baseline_output="different pre-existing failure", ) @@ -582,7 +618,7 @@ async def test_security_blocker_stops_without_repair_retry(tmp_path: Path) -> No @pytest.mark.asyncio -async def test_budget_exhaustion_fails_without_verification(tmp_path: Path) -> None: +async def test_budget_exhaustion_retains_partial_evidence_and_turns(tmp_path: Path) -> None: workspace, commit = _workspace(tmp_path) verifier_called = False @@ -616,7 +652,8 @@ async def test_budget_exhaustion_fails_without_verification(tmp_path: Path) -> N ) assert result.state is PreparationState.FAILED - assert verifier_called is False + assert verifier_called is True + assert result.attempt_history[0].repair.turns_used == 40 assert result.changed_files == ["app.py"] @@ -673,8 +710,8 @@ async def test_prepare_fix_requires_verifier_security_test(tmp_path: Path) -> No verify=verify_without_test, ) - assert result.state is PreparationState.FAILED - assert "did not execute a security test" in " ".join(result.gaps) + assert result.state is PreparationState.BLOCKED + assert "paired functional regression" in " ".join(result.gaps) @pytest.mark.asyncio @@ -695,7 +732,7 @@ async def test_prepare_fix_requires_closed_security_invariant(tmp_path: Path) -> verify=incomplete, ) - assert result.state is PreparationState.FAILED + assert result.state is PreparationState.BLOCKED assert "invariant" in " ".join(result.gaps) @@ -852,3 +889,110 @@ async def test_run_command_blocks_egress_when_network_not_allowed(tmp_path: Path assert "not run" in result.output else: assert result.status is CheckStatus.FAILED + + +@pytest.mark.asyncio +async def test_manifest_patch_includes_untracked_companion_file(tmp_path: Path) -> None: + + workspace, _ = _workspace(tmp_path) + (workspace / "companion.py").write_text("guard = True\n") + manifest, _, _ = await build_git_manifest(workspace) + patch = await build_git_patch(workspace, manifest) + assert b"b/companion.py" in patch + assert b"+guard = True" in patch + _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 = [] + + 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) -> 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 + assert ( + not _regression() + .model_copy( + update={"base": _regression().base.model_copy(update={"failure_kind": "environment"})} + ) + .passed() + ) + + +def test_candidate_keeps_full_finding_without_inventing_reproduction() -> None: + candidate = candidate_from_legacy_report( + { + "title": "Authorization bypass", + "technical_analysis": "Critical exploit context.", + "remediation_steps": "Enforce authorization.", + "code_locations": [ + { + "file": "app.py", + "start_line": 1, + "end_line": 1, + "fix_before": "unsafe", + "fix_after": "safe", + } + ], + } + ) + assert candidate and candidate.finding + assert candidate.finding.description == "Critical exploit context." + assert candidate.reproduction is None