Require an explicit fix handoff for source-backed findings

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.

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",

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,

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,

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:

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()):

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()

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: