From 77bd5dade5b870d78d74a1ab552af6018c5d8d80 Mon Sep 17 00:00:00 2001 From: Jonathan Singer Date: Wed, 30 Sep 2026 03:40:21 -0400 Subject: [PATCH] Require an explicit fix handoff for source-backed findings --- docs/fix-preparation.md | 14 +++ strix/fix/__init__.py | 2 + strix/fix/contracts.py | 25 ++++- strix/fix/prepare.py | 8 +- strix/report/state.py | 8 +- strix/tools/reporting/tool.py | 161 +++++++++++++++++++++--------- tests/test_fix_preparation.py | 19 ++++ tests/test_reporting_fields.py | 172 ++++++++++++++++++++++++++++++++- 8 files changed, 353 insertions(+), 56 deletions(-) diff --git a/docs/fix-preparation.md b/docs/fix-preparation.md index 8207fcc24..7fbe366a2 100644 --- a/docs/fix-preparation.md +++ b/docs/fix-preparation.md @@ -27,6 +27,20 @@ agent instructions, not a separate controller that selects or interprets tests. ## Completion and handoffs +Before preparation, a source-backed scan report must supply paired `fix_before` / +`fix_after` edits in `code_locations` plus `fix_verification`, or an explicit +`fix_candidate_blocker` with a reason. Prose in `fix_pr_body` alone is not a code +handoff. This also applies to external tests with an attached repository. Black-box +findings without repository source can still be reported without code locations. +The verification note describes draft reasoning and testing gaps; the scanner does +not need to implement or test the fix before reporting the vulnerability. + +The blocker is stored as `fix_candidate.blocker`. It preserves the finding and +explains why preparation cannot start. Updating with a blocker withdraws old edits; +new paired locations and verification replace it. Hosted callers persist the whole +candidate and its digest together, and candidate changes invalidate old preparations +and revoke outstanding callbacks. + Agents finish through Strix's `agent_finish` tool: - Repair: `done` starts review; `blocked` stops and preserves work. diff --git a/strix/fix/__init__.py b/strix/fix/__init__.py index 5b77605a2..65b166cdf 100644 --- a/strix/fix/__init__.py +++ b/strix/fix/__init__.py @@ -8,6 +8,7 @@ from strix.fix.contracts import ( CommandSpec, FileManifestEntry, FindingContext, + FixCandidateBlocker, FixCandidateV1, FixEdit, FixPreparationAttempt, @@ -42,6 +43,7 @@ __all__ = [ "CommandSpec", "FileManifestEntry", "FindingContext", + "FixCandidateBlocker", "FixCandidateV1", "FixEdit", "FixPreparationAttempt", diff --git a/strix/fix/contracts.py b/strix/fix/contracts.py index 835492d72..6964aecd7 100644 --- a/strix/fix/contracts.py +++ b/strix/fix/contracts.py @@ -158,6 +158,19 @@ class FindingContext(ContractModel): remediation: str = "" +class FixCandidateBlocker(ContractModel): + """Why the reporting agent cannot hand off a code fix for this finding.""" + + reason: str = Field(min_length=1) + + @field_validator("reason") + @classmethod + def nonempty_reason(cls, value: str) -> str: + if not value.strip(): + raise ValueError("explain what prevents a code fix") + return value.strip() + + class FixCandidateV1(ContractModel): version: Literal["1"] = "1" source_identity: SourceIdentity | None = None @@ -168,11 +181,14 @@ class FixCandidateV1(ContractModel): reported_checks: list[ReportedCheck] = [] known_gaps: list[str] = [] finding: FindingContext | None = None + blocker: FixCandidateBlocker | 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. + if self.blocker is None: + data.pop("blocker", None) if data.get("reproduction") and data["reproduction"].get("command"): data["reproduction"]["command"].pop("purpose", None) payload = json.dumps( @@ -295,10 +311,12 @@ def candidate_from_legacy_report( *, source_identity: SourceIdentity | None = None, ) -> FixCandidateV1 | None: + raw_blocker = report.get("fix_candidate_blocker") + blocker = FixCandidateBlocker.model_validate(raw_blocker) if raw_blocker is not None else None raw_locations = report.get("code_locations") - if not isinstance(raw_locations, list): + if not isinstance(raw_locations, list) and blocker is None: return None - location_values = cast("list[object]", raw_locations) + location_values = cast("list[object]", raw_locations) if isinstance(raw_locations, list) else [] locations: list[CandidateLocation] = [] edits: list[FixEdit] = [] @@ -329,7 +347,7 @@ def candidate_from_legacy_report( except ValueError: continue - if not locations: + if not locations and blocker is None: return None invariant = str( @@ -348,6 +366,7 @@ def candidate_from_legacy_report( reproduction = str(report.get("poc_description") or report.get("evidence") or "").strip() return FixCandidateV1( source_identity=source_identity, + blocker=blocker, security_invariant=invariant, finding_locations=locations, draft_edits=edits, diff --git a/strix/fix/prepare.py b/strix/fix/prepare.py index c996f4614..be8cb8840 100644 --- a/strix/fix/prepare.py +++ b/strix/fix/prepare.py @@ -308,10 +308,16 @@ async def prepare_fix( # noqa: PLR0915 - thin orchestration and cleanup gaps=gaps, ) - async def execute() -> FixPreparationResultV1: # noqa: PLR0911 - explicit terminal outcomes + async def execute() -> FixPreparationResultV1: # noqa: PLR0911, PLR0912 - terminal outcomes nonlocal checks, verifier, repair_turns, review_turns if cancelled(): raise PreparationCancelledError + if context.candidate.blocker: + return await finish( + PreparationState.BLOCKED, + context.candidate.blocker.reason, + gaps=[context.candidate.blocker.reason], + ) if not await source_verifier(context): return await finish( PreparationState.STALE, diff --git a/strix/report/state.py b/strix/report/state.py index adec8c93f..ddd0c1eb2 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -487,7 +487,7 @@ class ReportState: changed: dict[str, Any] = {} for key, raw_value in fields.items(): - if key not in UPDATABLE_REPORT_FIELDS or raw_value is None: + if key not in UPDATABLE_REPORT_FIELDS or (raw_value is None and key != "fix_candidate"): continue value = raw_value if isinstance(value, str): @@ -892,7 +892,7 @@ class ReportState: repo_targets = [ target for target in targets - if isinstance(target, dict) and target.get("type") == "repository" + if isinstance(target, dict) and target.get("type") in {"repository", "local_code"} ] # Provenance binds the whole run to one repo; with multiple repo targets # that's ambiguous, so omit it rather than mis-attributing later repos' @@ -904,6 +904,8 @@ class ReportState: if not isinstance(details, dict): return None uri = details.get("target_repo") + if target.get("type") == "local_code" and details.get("target_path"): + uri = Path(details["target_path"]).resolve().as_uri() if not isinstance(uri, str) or not uri.strip(): return None @@ -911,7 +913,7 @@ class ReportState: full_name = _parse_repo_full_name(uri) if full_name: context["repositoryFullName"] = full_name - cloned = details.get("cloned_repo_path") + cloned = details.get("cloned_repo_path") or details.get("target_path") if isinstance(cloned, str) and cloned.strip(): commit, branch = _git_head(cloned.strip()) if commit: diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index e5cdb4ad7..c9b1e09b9 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any, cast from agents import RunContextWrapper, function_tool +from strix.fix.contracts import FixCandidateBlocker from strix.tools.nullish import clean_optional from strix.tools.proxy.tools import existing_request_ids @@ -87,7 +88,7 @@ def _normalize_code_locations( if field in ("snippet", "fix_before", "fix_after") else str(value).strip() ) - if text: + if text or field == "fix_after": normalized[field] = text if normalized.get("file") and normalized.get("start_line") is not None: cleaned.append(normalized) @@ -340,7 +341,7 @@ def _validate_fix_verification( fix_verification: str | None, ) -> list[str]: """Require a verification statement whenever an applyable fix is proposed.""" - if not locations or not any(loc.get("fix_after") for loc in locations): + if not locations or not any("fix_after" in loc for loc in locations): return [] if str(fix_verification or "").strip(): return [] @@ -383,13 +384,16 @@ def _fix_source_context( return None, None raw_targets = cast("object", report_state.run_record.get("targets_info")) targets = cast("list[object]", raw_targets) if isinstance(raw_targets, list) else [] - if len(targets) != 1: - return SourceIdentity(kind=SourceIdentityKind.COMMIT, value=commit), None - raw_target = targets[0] + repo_targets = [ + t for t in targets if isinstance(t, dict) and t.get("type") in {"repository", "local_code"} + ] + if len(repo_targets) != 1: + return None, None + raw_target = repo_targets[0] target = cast("dict[str, object]", raw_target) if isinstance(raw_target, dict) else {} raw_details = target.get("details") details = cast("dict[str, object]", raw_details) if isinstance(raw_details, dict) else {} - repo_path = details.get("cloned_repo_path") + repo_path = details.get("cloned_repo_path") or details.get("target_path") repository = context.get("repositoryUri") return ( SourceIdentity( @@ -412,7 +416,7 @@ def _build_fix_candidate( candidate = candidate_from_legacy_report(report_fields, source_identity=source_identity) if candidate is None: return None - if repo_path is not None: + if repo_path is not None and candidate.blocker is None: try: anchored, results = anchor_candidate(repo_path, candidate) except Exception as exc: # noqa: BLE001 @@ -442,17 +446,48 @@ _FIX_CANDIDATE_FIELDS = frozenset( "poc_description", "evidence", "fix_verification", + "fix_pr_body", + "fix_candidate_blocker", } ) +def _fix_handoff( + report_state: ReportState, fields: Mapping[str, object] +) -> tuple[dict[str, object] | None, list[str]]: + """Require edits or an explanation, without requiring the scanner to repair the repo.""" + candidate = _build_fix_candidate(report_state, fields) + targets = report_state.run_record.get("targets_info") or [] + has_source = any( + isinstance(target, dict) and target.get("type") in {"repository", "local_code"} + for target in targets + ) + if candidate and candidate.get("blocker"): + if candidate.get("draft_edits"): + return None, [ + "Choose code edits or fix_candidate_blocker, not both. To withdraw an old fix, " + "replace code_locations with locations without fix_before/fix_after (or [])." + ] + return candidate, [] + if has_source and ( + not candidate or not candidate.get("draft_edits") or not candidate.get("source_identity") + ): + return None, [ + "Repository source is attached. Supply code_locations with relative file paths, " + "line ranges, exact fix_before/fix_after pairs and fix_verification; prose in " + "technical_analysis or fix_pr_body is not a fix handoff. If the finding cannot " + "be mapped or safely fixed, or its source revision is unavailable, supply " + 'fix_candidate_blocker={"reason": "Explain the missing context or external change"}. ' + "The blocker preserves the vulnerability without inventing a patch." + ] + return candidate, [] + + def _refresh_fix_candidate( report_state: ReportState, report_id: str, changes: dict[str, Any], -) -> None: - if not _FIX_CANDIDATE_FIELDS.intersection(changes): - return +) -> list[str]: existing = next( ( report @@ -462,15 +497,37 @@ def _refresh_fix_candidate( None, ) if existing is None: - return - candidate = _build_fix_candidate(report_state, {**existing, **changes}) - if candidate is not None: + return [] + if not _FIX_CANDIDATE_FIELDS.intersection(changes) and existing.get("fix_candidate"): + return [] + merged = {**existing, **changes} + if "fix_candidate_blocker" in changes and "code_locations" not in changes: + # Explicitly withdrawing a fix must not retain its old edits. + merged["code_locations"] = changes["code_locations"] = [ + {key: value for key, value in loc.items() if key not in {"fix_before", "fix_after"}} + for loc in existing.get("code_locations") or [] + ] + # A blocker is kept in the canonical candidate. New locations replace that + # decision; unrelated revisions must keep it until there is a real fix. + if "fix_candidate_blocker" not in changes and "code_locations" not in changes: + merged["fix_candidate_blocker"] = (existing.get("fix_candidate") or {}).get("blocker") + if "code_locations" in changes and "fix_verification" not in changes: + merged.pop("fix_verification", None) + errors = _validate_fix_verification( + merged.get("code_locations"), merged.get("fix_verification") + ) + candidate, handoff_errors = _fix_handoff(report_state, merged) + if errors or handoff_errors: + return errors + handoff_errors + changes.pop("fix_candidate_blocker", None) + if candidate is not None or "fix_candidate" in existing: changes["fix_candidate"] = candidate if candidate is not None or existing.get("fix_candidate") or existing.get("fix_preparation"): changes["fix_preparation"] = { "state": "stale", "stop_reason": "The finding or draft candidate changed after preparation.", } + return [] _UPDATE_TEXT_FIELDS = ( @@ -549,12 +606,26 @@ def _collect_update_changes( # noqa: PLR0912, PLR0915 errors.extend(_validate_code_locations(locations)) errors.extend(_validate_fix_verification(locations, changes.get("fix_verification"))) changes["code_locations"] = locations + elif raw_locations == []: + changes["code_locations"] = [] elif raw_locations: errors.append( "code_locations were dropped as unusable - every location needs a relative " "'file' and an integer 'start_line'" ) + raw_blocker = fields.get("fix_candidate_blocker") + if raw_blocker is not None: + try: + changes["fix_candidate_blocker"] = FixCandidateBlocker.model_validate( + raw_blocker + ).model_dump() + except ValueError: + errors.append( + "fix_candidate_blocker needs a non-empty reason explaining " + "why no code fix is provided" + ) + cve, cwe, identifier_errors = _validate_identifiers( clean_optional(fields.get("cve")), clean_optional(fields.get("cwe")) ) @@ -722,7 +793,7 @@ def _read_revision( return changes, None -def _do_update( +def _do_update( # noqa: PLR0911 - explicit validation and persistence outcomes *, report_id: str, update_reason: str, @@ -753,7 +824,9 @@ def _do_update( class_error = _fit_revision_to_class(report_state, report_id, changes) if class_error is not None: return class_error - _refresh_fix_candidate(report_state, report_id, changes) + fix_errors = _refresh_fix_candidate(report_state, report_id, changes) + if fix_errors: + return {"success": False, "error": "Validation failed", "errors": fix_errors} try: updated = report_state.update_vulnerability_report( @@ -880,7 +953,7 @@ def _do_delete( } -async def _do_create( +async def _do_create( # noqa: PLR0911 - explicit validation and persistence outcomes *, title: str, description: str, @@ -906,6 +979,7 @@ async def _do_create( confidence_rationale: str | None = None, fix_verification: str | None = None, fix_pr_body: str | None = None, + fix_candidate_blocker: FixCandidateBlocker | None = None, agent_id: str | None = None, agent_name: str | None = None, ) -> dict[str, Any]: @@ -1013,9 +1087,13 @@ async def _do_create( "code_locations": parsed_locations, "fix_verification": fix_verification, "fix_pr_body": fix_pr_body, + "fix_candidate_blocker": fix_candidate_blocker, "http_exchange_ids": normalized_http_exchange_ids, } - fix_candidate = _build_fix_candidate(report_state, report_fields) + fix_candidate, fix_errors = _fix_handoff(report_state, report_fields) + if fix_errors: + return {"success": False, "error": "Validation failed", "errors": fix_errors} + report_fields.pop("fix_candidate_blocker", None) if fix_candidate: report_fields["fix_candidate"] = fix_candidate @@ -1111,6 +1189,7 @@ async def create_vulnerability_report( confidence_rationale: str | None = None, fix_verification: str | None = None, fix_pr_body: str | None = None, + fix_candidate_blocker: FixCandidateBlocker | None = None, ) -> str: """File a vulnerability report — one report per fully-verified finding. @@ -1425,37 +1504,17 @@ async def create_vulnerability_report( - Padding ``fix_before`` with surrounding context lines that aren't part of the fix. - Duplicating the same change across multiple locations. - fix_verification: REQUIRED whenever any ``code_locations`` entry - carries a ``fix_after``. This field records the reporting - agent's checks on the draft candidate. It is not an independent - verification result. Before writing this field, work the gates - **in order**: - - 1. **Security closure** — re-trace the source → sink path - through the *patched* code and state why it is now - blocked. Re-run the PoC against the fix if you can. - 2. **Bypass review** — re-read the diff *without* leaning on - the rationale that produced it. Name the sibling call - sites, equivalent sinks, and alternate malicious input - classes you checked, and try at least one. - 3. **Preserved behavior** — name the legitimate inputs, - public APIs, and error semantics that must keep working, - and confirm the patch leaves them intact. A fix that - breaks the feature is not a fix. - 4. **Repository checks** — run the narrowest relevant - syntax / type / lint / test check that covers the - changed lines. - - Then write what you did: the commands you ran and their - results, and every gate you could only reason about rather - than execute, marked explicitly as a gap. Do not claim that a - gate passed because the candidate looks correct. If a gate - fails, revise the candidate or drop ``fix_after``. - - Also use this field to record the narrowest-complete-change - judgement: prefer the smallest repository-native fix that - fully enforces the invariant, using existing helpers, with - no unrelated refactors folded in. + fix_verification: Required with draft edits, including deletions. + Explain why the proposed change addresses the finding and what + you checked. Distinguish executed checks from reasoning and + name testing gaps. Repair and independent testing happen later; + this field does not claim the draft is verified. + fix_candidate_blocker: With repository source attached, provide either + code_locations containing paired fix_before/fix_after edits and + fix_verification, or this object with a concrete reason why you + cannot provide a code fix (unmapped code, missing context, or an + external configuration change). The finding is still reported. + Do not supply a blocker together with draft edits. fix_pr_body: Optional. When source is available and you have a concrete fix, a markdown PR-description body proposing the fix (summary + rationale). Prose/markdown only — the code @@ -1556,6 +1615,7 @@ async def create_vulnerability_report( http_exchange_ids=http_exchange_ids, fix_verification=fix_verification, fix_pr_body=fix_pr_body, + fix_candidate_blocker=fix_candidate_blocker, agent_id=agent_id, agent_name=agent_name, ) @@ -1591,6 +1651,7 @@ async def update_vulnerability_report( http_exchange_ids: list[str] | None = None, fix_verification: str | None = None, fix_pr_body: str | None = None, + fix_candidate_blocker: FixCandidateBlocker | None = None, contextual_cvss_reasoning: str | None = None, ) -> str: """Revise a vulnerability report that is already filed, keeping its id. @@ -1665,6 +1726,9 @@ async def update_vulnerability_report( list to remove all linked exchanges. fix_verification: Verification statement for an applyable fix. fix_pr_body: Replacement fix PR body. + fix_candidate_blocker: Explain why no code fix can be provided; clears + the previous candidate's edits. New paired code_locations resolve + the blocker. Source-backed reports cannot silently drop their fix. contextual_cvss_reasoning: Dependency findings only. What you observed in this codebase that justifies the contextual ``cvss_breakdown``. @@ -1710,6 +1774,7 @@ async def update_vulnerability_report( "http_exchange_ids": http_exchange_ids, "fix_verification": fix_verification, "fix_pr_body": fix_pr_body, + "fix_candidate_blocker": fix_candidate_blocker, "contextual_cvss_reasoning": contextual_cvss_reasoning, } if http_exchange_warning and all(value is None for value in fields.values()): diff --git a/tests/test_fix_preparation.py b/tests/test_fix_preparation.py index f99c9314b..9511191f5 100644 --- a/tests/test_fix_preparation.py +++ b/tests/test_fix_preparation.py @@ -8,6 +8,7 @@ import json import subprocess import sys from typing import TYPE_CHECKING +from unittest.mock import AsyncMock import pytest @@ -132,6 +133,23 @@ def _request(candidate: FixCandidateV1, *, attempts: int = 2) -> FixPreparationR ) +async def test_explicit_candidate_blocker_does_not_start_agents(tmp_path: Path) -> None: + workspace, _commit = _workspace(tmp_path) + candidate = FixCandidateV1.model_validate( + { + "security_invariant": "Guard access", + "blocker": {"reason": "Affected source is unavailable."}, + } + ) + repair, review = AsyncMock(), AsyncMock() + result = await prepare_fix(_request(candidate), workspace, repair=repair, verify=review) + assert result.state is PreparationState.BLOCKED + assert result.stop_reason == candidate.blocker.reason + assert result.attempts == 0 + repair.assert_not_awaited() + review.assert_not_awaited() + + async def _noop_repair( context: PreparationContext, _checks: list[CheckResult], @@ -557,6 +575,7 @@ def test_new_command_metadata_does_not_change_existing_finding_digest(tmp_path: candidate = _candidate(commit) payload = candidate.model_dump(mode="json") payload.pop("finding") + payload.pop("blocker") payload["reproduction"]["command"].pop("purpose") previous = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() diff --git a/tests/test_reporting_fields.py b/tests/test_reporting_fields.py index b07ee39fa..6f7a8fba6 100644 --- a/tests/test_reporting_fields.py +++ b/tests/test_reporting_fields.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import subprocess from typing import TYPE_CHECKING, Any, cast import pytest @@ -1139,6 +1140,175 @@ _INFO_LOCATION = { } +def _attach_source(state: ReportState, tmp_path: Path, *, local: bool = False) -> str: + repo = tmp_path / "source" + (repo / "app").mkdir(parents=True) + (repo / "app/views.py").write_text(_FIX_LOCATION["fix_before"] + "\n") + for args in ( + ["init", "-q"], + ["add", "."], + ["-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "base"], + ): + subprocess.run(["/usr/bin/git", "-C", str(repo), *args], check=True) # noqa: S603 + state.run_record["targets_info"] = [ + {"type": "web_application", "details": {"target_url": "https://example.com"}}, + { + "type": "local_code" if local else "repository", + "details": {"target_path": str(repo)} + if local + else { + "target_repo": "https://github.com/example/app", + "cloned_repo_path": str(repo), + }, + }, + ] + return subprocess.run( # noqa: S603 + ["/usr/bin/git", "-C", str(repo), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +@pytest.mark.parametrize("locations", [None, [], [_INFO_LOCATION]]) +async def test_source_report_requires_edits_or_blocker( + report_state: ReportState, + tmp_path: Path, + locations: Any, +) -> None: + _attach_source(report_state, tmp_path) + result = await _create_with( + report_state, code_locations=locations, fix_pr_body="Use parameters." + ) + assert not result["success"] + assert "fix_candidate_blocker" in " ".join(result["errors"]) + assert not report_state.vulnerability_reports + + +@pytest.mark.parametrize("local", [False, True]) +async def test_mixed_source_report_persists_candidate_identity( + report_state: ReportState, + tmp_path: Path, + local: bool, +) -> None: + commit = _attach_source(report_state, tmp_path, local=local) + result = await _create_with( + report_state, + code_locations=[_FIX_LOCATION], + fix_verification="Traced the draft; tests pending.", + ) + assert result["success"], result + candidate = report_state.vulnerability_reports[0]["fix_candidate"] + assert candidate["source_identity"]["value"] == commit + assert candidate["source_identity"]["repository"] + assert candidate["draft_edits"][0]["before"] == _FIX_LOCATION["fix_before"] + + +async def test_source_without_revision_can_report_explicit_blocker( + report_state: ReportState, +) -> None: + report_state.run_record["targets_info"] = [{"type": "repository", "details": {}}] + rejected = await _create_with( + report_state, code_locations=[_FIX_LOCATION], fix_verification="Traced the draft." + ) + assert not rejected["success"] + result = await _create_with( + report_state, fix_candidate_blocker={"reason": "The checkout revision is unavailable."} + ) + assert result["success"], result + candidate = report_state.vulnerability_reports[0]["fix_candidate"] + assert candidate["blocker"]["reason"] == "The checkout revision is unavailable." + assert candidate["draft_edits"] == [] + + +async def test_update_withdraws_and_replaces_candidate( + report_state: ReportState, + tmp_path: Path, +) -> None: + _attach_source(report_state, tmp_path) + await _create_with( + report_state, code_locations=[_FIX_LOCATION], fix_verification="Draft review." + ) + report_id = report_state.vulnerability_reports[0]["id"] + assert not _do_update( + report_id=report_id, update_reason="withdraw", fields={"code_locations": []} + )["success"] + result = _do_update( + report_id=report_id, + update_reason="external change needed", + fields={ + "fix_candidate_blocker": {"reason": "Requires an upstream service change."}, + }, + ) + assert result["success"], result + report = report_state.vulnerability_reports[0] + assert report["fix_candidate"]["draft_edits"] == [] + assert report["fix_preparation"]["state"] == "stale" + assert "fix_after" not in report["code_locations"][0] + result = _do_update( + report_id=report_id, update_reason="more evidence", fields={"evidence": "New evidence"} + ) + assert result["success"], result + assert report_state.vulnerability_reports[0]["fix_candidate"]["blocker"] + result = _do_update( + report_id=report_id, + update_reason="local fix found", + fields={ + "code_locations": [_FIX_LOCATION], + "fix_verification": "Draft trace; tests pending.", + }, + ) + assert result["success"], result + candidate = report_state.vulnerability_reports[0]["fix_candidate"] + assert candidate["draft_edits"] + assert candidate["blocker"] is None + + +async def test_blackbox_candidate_can_be_explicitly_cleared(report_state: ReportState) -> None: + await _create_with( + report_state, code_locations=[_FIX_LOCATION], fix_verification="Draft review." + ) + report_id = report_state.vulnerability_reports[0]["id"] + result = _do_update( + report_id=report_id, update_reason="source no longer applies", fields={"code_locations": []} + ) + assert result["success"], result + report = report_state.vulnerability_reports[0] + assert report["fix_candidate"] is None + assert report["code_locations"] == [] + assert report["fix_preparation"]["state"] == "stale" + + +async def test_deletion_edit_is_kept_and_requires_verification(report_state: ReportState) -> None: + location = {**_FIX_LOCATION, "fix_after": ""} + rejected = await _create_with(report_state, code_locations=[location]) + assert not rejected["success"] + result = await _create_with( + report_state, code_locations=[location], fix_verification="Remove the unsafe statement." + ) + assert result["success"], result + assert report_state.vulnerability_reports[0]["fix_candidate"]["draft_edits"][0]["after"] == "" + + +def test_reporting_tools_expose_optional_fix_blocker() -> None: + for tool in (create_vulnerability_report, update_vulnerability_report): + assert "fix_candidate_blocker" in tool.params_json_schema["properties"] + + +async def test_legacy_source_report_update_requires_fix_handoff( + report_state: ReportState, + tmp_path: Path, +) -> None: + created = await _create_with(report_state) + _attach_source(report_state, tmp_path) + result = _do_update( + report_id=created["report_id"], update_reason="Correct title", fields={"title": "New title"} + ) + assert not result["success"] + assert "fix_candidate_blocker" in " ".join(result["errors"]) + assert report_state.vulnerability_reports[0]["title"] == "X" + + async def test_fix_after_requires_verification(report_state: ReportState) -> None: result = await _create_with(report_state, code_locations=[_FIX_LOCATION]) assert result["success"] is False @@ -2002,7 +2172,7 @@ def test_update_marks_preparation_stale_when_candidate_cannot_be_rebuilt( assert report["code_locations"] == [{"file": "./files.py", "start_line": 4, "end_line": 9}] assert report["fix_preparation"]["state"] == "stale" assert "changed after preparation" in report["fix_preparation"]["stop_reason"] - assert report["fix_candidate"] == candidate + assert report["fix_candidate"] is None def _seed_saved_report(report_state: ReportState) -> Path: