mirror of
https://github.com/usestrix/strix.git
synced 2026-10-02 02:13:43 +00:00
refactor fix preparation gates
This commit is contained in:
yoni
parent
c510f58ef3
commit
df313c1438
4 changed files with 497 additions and 500 deletions
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue