Preserve partial fixes and require consistent execution evidence

strix/fix/contracts.py

@@ -65,6 +65,7 @@ class BlockerKind(StrEnum):
     CREDENTIAL = "credential"
     EXTERNAL_CONFIGURATION = "external_configuration"
     SECURITY_EVIDENCE = "security_evidence"
+    VERIFICATION_RUNTIME = "verification_runtime"
 
 
 class PreparationBlocker(ContractModel):
@@ -211,9 +212,13 @@ class CheckResult(ContractModel):
     baseline_output: str | None = None
     cwd: str = "."
     baseline_exit_code: int | None = None
-    failure_kind: Literal["environment", "timeout", "check", "source_changed"] | None = None
+    failure_kind: (
+        Literal["environment", "timeout", "check", "source_changed", "harness", "unknown"] | None
+    ) = None
     workspace_root: str | None = None
     baseline_workspace_root: str | None = None
+    source_digest: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+    environment_id: str | None = None
 
 
 class RegressionTestResult(ContractModel):
@@ -227,6 +232,15 @@ class RegressionTestResult(ContractModel):
     behavior: CheckResult
 
     def passed(self) -> bool:
+        results = (self.base, self.patched, self.behavior)
+        # Legacy records remain readable. Once provenance is supplied, every leg
+        # must have it and the environment must be unchanged across the pair.
+        if any(item.source_digest or item.environment_id for item in results) and (
+            not all(item.source_digest and item.environment_id for item in results)
+            or len({item.environment_id for item in results}) != 1
+            or self.patched.source_digest != self.behavior.source_digest
+        ):
+            return False
         return (
             self.base.target is VerificationTarget.BASE
             and self.patched.target is VerificationTarget.PATCHED

strix/fix/evidence.py

@@ -10,18 +10,30 @@ from strix.fix.contracts import CheckStatus
 
 def command_status(
     exit_code: int, output: str
-) -> tuple[CheckStatus, Literal["environment", "check"] | None]:
+) -> tuple[CheckStatus, Literal["environment", "check", "harness", "unknown"] | 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)",
+            r"(?:command not found|exec: \S+: not found)",
             output,
             re.IGNORECASE,
         )
     ):
         return CheckStatus.UNAVAILABLE, "environment"
+    if exit_code != 0 and re.search(
+        r"(?:SyntaxError|ReferenceError: require is not defined)", output
+    ):
+        return CheckStatus.FAILED, "unknown"
+    if exit_code != 0 and re.search(
+        r"(?:No module named |Cannot find module |ERR_MODULE_NOT_FOUND|"
+        r"ECONNREFUSED|Could not connect to server)",
+        output,
+        re.IGNORECASE,
+    ):
+        # An incorrect import or a failed service is not evidence that installing
+        # dependencies is appropriate. Return it to the caller for diagnosis.
+        return CheckStatus.FAILED, "unknown"
     if re.search(
         r"(?:Skipping to avoid parser lock|collected 0 items|"
         r"No test files found|No tests found, exiting with code 0)",

strix/fix/prepare.py

@@ -29,6 +29,7 @@ from strix.fix.contracts import (
     RepairOutcome,
     RepairStatus,
     VerificationDecision,
+    VerificationTarget,
     VerifierResult,
 )
 from strix.fix.evidence import command_status
@@ -414,10 +415,27 @@ 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
+    }
     return (
         _required_checks_pass(checks)
+        and len(environments) <= 1
+        and len(patched_sources) <= 1
         and verifier.decision is VerificationDecision.VERIFIED
         and verifier.security_invariant_closed
+        and verifier.reproduction_executed
         and not verifier.gaps
         and verifier.blocker is None
         and bool(verifier.regression_tests)
@@ -516,13 +534,16 @@ async def prepare_fix(  # noqa: PLR0915
         gaps: list[str] | None = None,
     ) -> FixPreparationResultV1:
         manifest, summary, artifact = await manifest_builder(workspace)
+        retained_verifier = verifier or (
+            context.feedback[-1].verifier if context.feedback else None
+        )
         return _result(
             context,
             state=state,
             reason=reason,
             started=started,
             checks=checks,
-            verifier=verifier,
+            verifier=retained_verifier,
             reproduction=reproduction,
             manifest=manifest,
             diff_summary=summary,
@@ -592,6 +613,7 @@ async def prepare_fix(  # noqa: PLR0915
                 return await finish(
                     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
@@ -599,11 +621,29 @@ async def prepare_fix(  # noqa: PLR0915
                 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 not _matches_baseline_failure(item)
+                if item.status is CheckStatus.FAILED
+                and item.failure_kind != "source_changed"
+                and not _matches_baseline_failure(item)
             ]
             unchanged_retry = (
                 len(context.feedback) > 1
@@ -668,7 +708,19 @@ async def prepare_fix(  # noqa: PLR0915
             baseline = [item for item in required if _matches_baseline_failure(item)]
             blocker = verifier.blocker
             if blocker is None:
-                if unavailable or not required:
+                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=(

tests/test_fix_preparation.py

@@ -1041,3 +1041,101 @@ def test_candidate_keeps_full_finding_without_inventing_reproduction() -> None:
     assert candidate and candidate.finding
     assert candidate.finding.description == "Critical exploit context."
     assert candidate.reproduction is None
+
+
+def test_ambiguous_imports_and_syntax_never_authorize_environment_recovery() -> None:
+    for output in [
+        "Cannot find module '/tmp/test/skills/policy.js'",
+        "No module named 'wrong_repo_path'",
+        "SyntaxError: invalid syntax",
+    ]:
+        status, kind = command_status(1, output)
+        assert status is CheckStatus.FAILED
+        assert kind == "unknown"
+    assert command_status(127, "bun: command not found") == (CheckStatus.UNAVAILABLE, "environment")
+
+
+def test_regression_requires_consistent_execution_provenance() -> None:
+    regression = _regression()
+    for leg in (regression.base, regression.patched, regression.behavior):
+        leg.source_digest = "a" * 64
+        leg.environment_id = "execution:0"
+    assert regression.passed()
+    regression.base.environment_id = "execution:1"
+    assert not regression.passed()
+    regression.base.environment_id = "execution:0"
+    regression.behavior = regression.behavior.model_copy(update={"source_digest": "b" * 64})
+    assert not regression.passed()
+
+
+@pytest.mark.asyncio
+async def test_late_setup_refreshes_earlier_checks_before_verification(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
+        return CheckResult(
+            name=command.name,
+            argv=command.argv,
+            status=CheckStatus.PASSED,
+            exit_code=0,
+            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
+
+    result = await prepare_fix(
+        request, workspace, repair=_noop_repair, verify=verify, command_runner=runner
+    )
+    assert result.state is PreparationState.READY
+    assert calls == ["compile", "native tests", "compile"]
+
+
+@pytest.mark.asyncio
+async def test_harness_failure_retains_patch_and_does_not_blame_customer(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",
+                )
+            ],
+        )
+
+    result = await prepare_fix(
+        _request(_candidate(commit)), workspace, repair=_noop_repair, verify=broken_verifier
+    )
+    assert result.state is PreparationState.BLOCKED
+    assert result.blocker.kind is BlockerKind.VERIFICATION_RUNTIME
+    assert result.final_file_manifest
+    assert "prerequisite" not in result.blocker.user_action