From c4639256de01cc4b0c62cec840f0d0a273d6bb3a Mon Sep 17 00:00:00 2001 From: Sean Turner Date: Wed, 15 Jul 2026 15:07:58 +0100 Subject: [PATCH 1/2] feat(resume): opt-in grounded retraction of fixed findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On --resume, prior findings are rehydrated so the cumulative report/SARIF carries forward across runs. That set is append-only today: if a later run sees a finding has been FIXED, there's no way to drop it, so a resolved finding re-emits into every subsequent SARIF forever — a scanner alert that can never auto-resolve (fail-CLOSED). This adds the inverse of report creation, behind an explicit opt-in. - `ReportState.retract_vulnerability_report(id, reason)` — inverse of add_vulnerability_report: drops the report, deletes its stale MD, and lets save_run_data() rewrite CSV/JSON/SARIF from the reduced set. Idempotent. - `retract_vulnerability_report` tool, exposed on the root agent ONLY on a resumed run AND only when `STRIX_RESUME_RETRACT` is set. Default OFF — resume stays append-only (unchanged behaviour) unless opted in. - Groundedness guard: before dropping, re-verify against the live workspace tree (via an injected sandbox reader) that the vulnerable sink is actually gone. If a fix_before sink is still present verbatim → REFUSE; if no reader is wired → REFUSE (fail-safe). A confident-but-wrong agent can't silently drop a real finding. - id stability: allocate the next `vuln-NNNN` from the max ordinal ever seen (incl. retracted ids retained in _saved_vuln_ids), not the list length. Length-based ids recycle once a report can be removed — a later finding would reuse a retracted id and collide with its MD / SARIF fingerprint. No-op for append-only runs (max-seen == len there). Tests: retract removal/idempotency/reason-required/MD-deletion, id stability under removal, and the guard's refuse/allow paths. Existing report + state tests unchanged (46 pass). --- strix/agents/factory.py | 16 +++ strix/config/settings.py | 10 ++ strix/core/runner.py | 41 ++++++++ strix/report/state.py | 96 +++++++++++++++++- strix/tools/reporting/retract_tool.py | 140 ++++++++++++++++++++++++++ tests/test_retract_vulnerability.py | 127 +++++++++++++++++++++++ 6 files changed, 429 insertions(+), 1 deletion(-) create mode 100644 strix/tools/reporting/retract_tool.py create mode 100644 tests/test_retract_vulnerability.py diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 977d01f19..e89237844 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -411,6 +411,7 @@ def build_strix_agent( system_prompt_context: dict[str, Any] | None = None, extra_tools: Sequence[Tool] | None = None, instructions_override: str | None = None, + is_resume: bool = False, ) -> SandboxAgent[Any]: """Build a SandboxAgent for either root or child use. @@ -421,6 +422,11 @@ def build_strix_agent( registered via ``register_agent_tools``. instructions_override: Use this verbatim as the system prompt instead of rendering the built-in scan prompt. + is_resume: On a resumed root run, and only when opted in via + ``RuntimeSettings.resume_retract_enabled`` (``STRIX_RESUME_RETRACT``), + expose ``retract_vulnerability_report`` so the agent can drop a + rehydrated finding it re-verifies as fixed. Off by default → resume + stays append-only. A fresh scan never needs it. """ if instructions_override is not None: instructions = instructions_override @@ -437,6 +443,16 @@ def build_strix_agent( agent_tools = [*_EXTRA_TOOLS, *(extra_tools or [])] if is_root: tools: list[Tool] = [*_BASE_TOOLS, *agent_tools, finish_scan] + # retract_vulnerability_report is the inverse of create — meaningful + # only on a RESUMED root run (fresh scans have nothing to retract) and + # only when explicitly opted in. Guarded again at runtime: the tool + # refuses unless a source-tree reader is wired (whitebox resume). + from strix.config import load_settings + + if is_resume and load_settings().runtime.resume_retract_enabled: + from strix.tools.reporting.retract_tool import retract_vulnerability_report + + tools.append(retract_vulnerability_report) else: tools = [*_BASE_TOOLS, *agent_tools, agent_finish] _ensure_unique_tool_names(tools) diff --git a/strix/config/settings.py b/strix/config/settings.py index 9d16e0836..ba7a55f68 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -56,6 +56,16 @@ class RuntimeSettings(BaseSettings): # on large repos). Above this, the user must bind-mount via ``--mount``. # Set to 0 (or less) to disable the pre-flight check entirely. max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB") + # Opt-in: on a resumed scan, expose the retract_vulnerability_report tool so + # the agent can DROP a rehydrated finding it re-verifies as fixed. Default + # OFF — resume stays append-only (today's behaviour) unless enabled. Only has + # any effect on a resumed whitebox run (the guard needs a source tree to + # re-verify against); a no-op otherwise. Useful when driving resume as a + # PR-lifecycle loop where cumulative findings must shrink as fixes land. + resume_retract_enabled: bool = Field( + default=False, + alias="STRIX_RESUME_RETRACT", + ) class TelemetrySettings(BaseSettings): diff --git a/strix/core/runner.py b/strix/core/runner.py index 0ebcc48b4..deb6d11df 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -236,6 +236,46 @@ async def run_strix_scan( system_prompt_context=root_context, ) + # Opt-in grounded retraction on resume: wire a sandbox-backed reader so + # the retract tool's guard can re-verify a "fixed" claim against the live + # /workspace tree before dropping a finding. Whitebox only — the guard + # checks source sinks; without a tree it stays fail-safe (refuse). Gated + # behind resume_retract_enabled so this is a no-op unless opted in. + if is_resume and is_whitebox and settings.runtime.resume_retract_enabled: + from strix.tools.reporting.retract_tool import set_target_file_reader + + _session = bundle["session"] + _ws_paths = [ + f"/workspace/{sub}" if sub else "/workspace" + for sub in ( + (t.get("details") or {}).get("workspace_subdir") + for t in targets + if t.get("type") == "local_code" + ) + ] or ["/workspace"] + + async def _read_target_file(rel_path: str) -> str | None: + rel = rel_path.lstrip("/") + for base in _ws_paths: + try: + result = await _session.exec("cat", f"{base}/{rel}", timeout=15) + except Exception as exc: # noqa: BLE001 — unreadable ≠ proof of absence + logger.debug("retract reader: %s/%s unreadable: %r", base, rel, exc) + continue + if getattr(result, "exit_code", 1) == 0: + # ExecResult.stdout is bytes; the guard does a str + # substring check, so decode here (a raw bytes return + # would raise TypeError on every retract). + stdout = result.stdout + return ( + stdout.decode("utf-8", errors="replace") + if isinstance(stdout, bytes | bytearray) + else stdout + ) + return None # not found under any workspace root → treat as gone + + set_target_file_reader(_read_target_file) + root_agent = build_strix_agent( name="strix", skills=skills, @@ -246,6 +286,7 @@ async def run_strix_scan( chat_completions_tools=chat_completions_tools, system_prompt_context=root_context, instructions_override=root_instructions, + is_resume=is_resume, ) if not is_resume: diff --git a/strix/report/state.py b/strix/report/state.py index 178c4047c..f3f91b6a5 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -1,3 +1,4 @@ +import contextlib import json import logging import subprocess @@ -205,6 +206,33 @@ def hydrate_from_run_dir(self) -> None: len(self.vulnerability_reports), ) + def _next_vuln_ordinal(self) -> int: + """Next ``vuln-NNNN`` ordinal, derived from the max ordinal EVER seen + rather than the current list length. + + List-length allocation (``len(reports) + 1``) is only correct while the + report set is append-only. Once a report can be *removed* + (:meth:`retract_vulnerability_report` on resume), length-based ids + recycle: retract ``vuln-0002`` from ``[0001,0002,0003]`` and the next + add computes ``len+1 = 3`` -> ``vuln-0003``, colliding with a live id and + overwriting its on-disk MD, and letting a retracted id later re-emit + under a different finding (SARIF fingerprint collision). Deriving from + the max ordinal across both the live reports and every id ever saved + (``_saved_vuln_ids`` retains retracted ids) keeps ids monotonic and + stable under removal. + """ + max_ordinal = 0 + for rid in ( + *(r.get("id") for r in self.vulnerability_reports), + *self._saved_vuln_ids, + ): + if isinstance(rid, str) and rid.startswith("vuln-"): + try: + max_ordinal = max(max_ordinal, int(rid.removeprefix("vuln-"))) + except ValueError: + continue + return max_ordinal + 1 + def add_vulnerability_report( self, title: str, @@ -232,7 +260,7 @@ def add_vulnerability_report( agent_id: str | None = None, agent_name: str | None = None, ) -> str: - report_id = f"vuln-{len(self.vulnerability_reports) + 1:04d}" + report_id = f"vuln-{self._next_vuln_ordinal():04d}" report: dict[str, Any] = { "id": report_id, @@ -296,6 +324,72 @@ def add_vulnerability_report( self.save_run_data() return report_id + def retract_vulnerability_report(self, report_id: str, reason: str) -> dict[str, Any]: + """Remove a previously-reported finding from the cumulative set. + + On resume, prior findings are rehydrated from ``vulnerabilities.json``. + When a later push FIXES one, the agent re-validates and must be able to + DROP it — otherwise rehydration is append-only and the fixed finding + re-emits into every subsequent SARIF forever, producing a scanner alert + that can never auto-resolve (fail-CLOSED). This is the inverse of + :meth:`add_vulnerability_report`. + + Reverses add's side effects: + 1. drop the report from the in-memory ``vulnerability_reports`` list + 2. delete the stale per-finding ``{id}.md`` (the one append-only + artefact; CSV/JSON/SARIF are rewritten wholesale by + :meth:`save_run_data` below from the reduced set) + + The id is retained in ``_saved_vuln_ids`` (NOT recycled): combined with + :meth:`_next_vuln_ordinal` deriving from the max ordinal ever seen, a + physical removal can never cause a later finding to reuse a retracted + id. Idempotent: retracting an unknown/already-gone id is a no-op success. + + NOTE: the caller (the retract tool) applies the groundedness guard — + re-verifying against the current tree that the vulnerable sink is + actually gone — BEFORE invoking this. This method performs the state + mutation only; it does not decide whether the retraction is justified. + """ + reason = (reason or "").strip() + if not reason: + raise ValueError( + "retract_vulnerability_report requires a non-empty reason " + "(cite what was verified fixed, and where)", + ) + retracted = next( + (r for r in self.vulnerability_reports if r.get("id") == report_id), + None, + ) + if retracted is None: + logger.info("retract: %s not present (no-op)", report_id) + return {"success": True, "retracted": False, "reason": "id not present"} + + before = len(self.vulnerability_reports) + self.vulnerability_reports = [ + r for r in self.vulnerability_reports if r.get("id") != report_id + ] + # Deliberately keep report_id in _saved_vuln_ids so the ordinal counter + # never re-allocates it; delete the stale MD so a future rehydration + # doesn't reload the retracted finding from disk. + with contextlib.suppress(OSError): + md_path = self.get_run_dir() / "vulnerabilities" / f"{report_id}.md" + md_path.unlink(missing_ok=True) + + logger.info( + "Retracted vulnerability report %s (%s) — reason: %s", + report_id, + retracted.get("title", ""), + reason, + ) + self.save_run_data() + return { + "success": True, + "retracted": True, + "report_id": report_id, + "title": retracted.get("title"), + "remaining": before - 1, + } + def get_existing_vulnerabilities(self) -> list[dict[str, Any]]: return list(self.vulnerability_reports) diff --git a/strix/tools/reporting/retract_tool.py b/strix/tools/reporting/retract_tool.py new file mode 100644 index 000000000..07f99a2f0 --- /dev/null +++ b/strix/tools/reporting/retract_tool.py @@ -0,0 +1,140 @@ +"""``retract_vulnerability_report`` — the inverse of create, for resumed scans. + +On ``--resume`` the agent rehydrates every prior finding. When a later push +FIXES one, the agent must be able to DROP it, or rehydration is append-only and +the fixed finding re-emits into every subsequent SARIF forever — an alert that +can never auto-resolve (fail-CLOSED). This tool closes that gap. + +GROUNDEDNESS GUARD (the load-bearing safety property): the tool does not blindly +trust the agent's "it's fixed" claim. Before retracting, it re-verifies against +the current tree via an injected ``read_target_file`` — recovering the finding's +fix-site (sink) locations and checking whether the vulnerable code is STILL +present. If a sink is still there verbatim, the retraction is REFUSED (a +confident-but-wrong agent must not be able to silently drop a real finding = +fail-open). Missing/deleted locations allow the retract (don't fabricate a block +from absent data). + +The reader is injected rather than hard-wired: the target lives inside the +sandbox in v1, so the concrete sandbox-backed reader is supplied by the caller +(the agent factory, which holds the session). When no reader is available the +tool FAILS SAFE — it refuses to retract rather than allowing an unverified drop +(a retract we can't ground is more dangerous than a finding that survives). + +Only exposed to the agent on resumed runs (see factory gating) — a fresh scan +can't need it and shouldn't be handed a findings-drop tool. +""" + +from __future__ import annotations + +import logging +from collections.abc import Awaitable, Callable +from typing import Any + +from agents import RunContextWrapper, function_tool + + +logger = logging.getLogger(__name__) + +# read_target_file(repo_relative_path) -> awaitable file contents, or None if +# absent/unreadable. Injected by the caller (holds the sandbox session), so the +# concrete reader can `session.exec("cat", ...)` against the live /workspace +# tree. None = no reader wired → the guard fails safe (refuse). +ReadTargetFileFn = Callable[[str], Awaitable["str | None"]] + +# Single-element list rather than a module global + ``global`` statement, so the +# reader can be swapped in place without PLW0603. +_reader: list[ReadTargetFileFn | None] = [None] + + +def set_target_file_reader(reader: ReadTargetFileFn | None) -> None: + """Wire the sandbox-backed target-file reader (called at agent build time on + resume). Passing None disables grounded retraction (fail-safe refuse).""" + _reader[0] = reader + + +def _sink_locations(report: dict[str, Any]) -> list[dict[str, Any]]: + """Fix-site (sink) locations to re-verify: only those carrying a + ``fix_before`` (the vulnerable line to change). A plain context ``snippet`` + frequently survives a genuine fix unchanged, so matching on it would + false-refuse a real fix (real-canary lesson from the 0.8 impl).""" + return [ + loc for loc in (report.get("code_locations") or []) + if isinstance(loc, dict) and loc.get("fix_before") and loc.get("file") + ] + + +async def _guard(report: dict[str, Any]) -> tuple[bool, str]: + """Return (allow, detail). allow=False → REFUSE the retract. + + - no reader wired → REFUSE (fail-safe: can't ground → don't drop). + - no fix_before sinks → ALLOW (nothing verifiable; don't fabricate a block). + - any sink's fix_before still present verbatim → REFUSE. + - otherwise → ALLOW. + """ + reader = _reader[0] + if reader is None: + return False, "no target-file reader wired — refusing ungrounded retract" + sinks = _sink_locations(report) + if not sinks: + return True, "no fix-site (fix_before) locations to verify — allowing" + for loc in sinks: + try: + content = await reader(str(loc["file"])) + except Exception: # noqa: BLE001 — reader failure ≠ proof of absence + return False, f"could not read {loc['file']} to verify — refusing" + if content is None: + continue # file gone → this sink can't be present + needle = str(loc["fix_before"]).strip() + if needle and needle in content: + return False, (f"vulnerable code still present at {loc['file']}: " + f"{needle[:80]!r}") + return True, "vulnerable code no longer present at any verified sink" + + +@function_tool(timeout=120, strict_mode=False) +async def retract_vulnerability_report( + ctx: RunContextWrapper, + report_id: str, + reason: str, +) -> dict[str, Any]: + """Retract a finding you have RE-VERIFIED as fixed in the current code. + + Use ONLY when a prior finding (rehydrated on resume) has been genuinely + fixed since it was reported. Cite what changed and where in ``reason``. The + retraction is re-verified against the current tree; if the vulnerable code + is still present, it is refused. + + Args: + report_id: the finding id to retract (e.g. ``vuln-0003``). + reason: code-grounded justification — what was fixed, and where. + """ + reason = (reason or "").strip() + if not reason: + return {"success": False, + "error": "reason is required — cite what was fixed and where"} + + from strix.report.state import get_global_report_state + state = get_global_report_state() + if state is None: + return {"success": False, "error": "no active report state"} + + report = next( + (r for r in state.get_existing_vulnerabilities() if r.get("id") == report_id), + None, + ) + if report is None: + return {"success": True, "retracted": False, "reason": "id not present"} + + allow, detail = await _guard(report) + if not allow: + logger.warning("retract REFUSED for %s: %s", report_id, detail) + return {"success": False, "retracted": False, "refused": True, + "error": f"retraction refused — {detail}"} + + try: + result = state.retract_vulnerability_report(report_id, reason) + except (ValueError, AttributeError) as e: + return {"success": False, "error": f"retract failed: {e!s}"} + logger.info("retract honoured for %s (guard: %s)", report_id, detail) + result["guard"] = detail + return result diff --git a/tests/test_retract_vulnerability.py b/tests/test_retract_vulnerability.py new file mode 100644 index 000000000..519a89a44 --- /dev/null +++ b/tests/test_retract_vulnerability.py @@ -0,0 +1,127 @@ +"""Tests for resume-time vulnerability retraction + id stability under removal.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from strix.report.state import ReportState, set_global_report_state +from strix.tools.reporting import retract_tool + + +if TYPE_CHECKING: + from pathlib import Path + + +@pytest.fixture +def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState: + monkeypatch.chdir(tmp_path) + state = ReportState(run_name="test-run") + set_global_report_state(state) + return state + + +def _add(state: ReportState, title: str) -> str: + return state.add_vulnerability_report(title=title, severity="high") + + +# --- ReportState.retract_vulnerability_report + id stability ----------------- + + +def test_retract_removes_from_cumulative_set(report_state: ReportState) -> None: + a, b, c = (_add(report_state, t) for t in ("SQLi", "CmdInj", "PathTrav")) + result = report_state.retract_vulnerability_report(b, reason="parameterised, verified fixed") + assert result["retracted"] is True + assert result["remaining"] == 2 + ids = [r["id"] for r in report_state.vulnerability_reports] + assert ids == [a, c] + assert b not in ids + + +def test_retract_is_idempotent_on_missing_id(report_state: ReportState) -> None: + _add(report_state, "SQLi") + result = report_state.retract_vulnerability_report("vuln-9999", reason="already gone") + assert result["success"] is True + assert result["retracted"] is False + + +def test_retract_requires_reason(report_state: ReportState) -> None: + vid = _add(report_state, "SQLi") + with pytest.raises(ValueError, match="non-empty reason"): + report_state.retract_vulnerability_report(vid, reason=" ") + + +def test_ids_are_stable_under_removal(report_state: ReportState) -> None: + """A retracted ordinal must never be recycled — otherwise a later finding + reuses a retracted id and collides with its on-disk MD / SARIF fingerprint. + """ + _add(report_state, "SQLi") # vuln-0001 + b = _add(report_state, "CmdInj") # vuln-0002 + c = _add(report_state, "PathTrav") # vuln-0003 + assert (b, c) == ("vuln-0002", "vuln-0003") + + report_state.retract_vulnerability_report(b, reason="fixed") + + # Next add must be vuln-0004, NOT a recycled 0002/0003. + assert _add(report_state, "SSRF") == "vuln-0004" + + +def test_retracted_md_is_deleted(report_state: ReportState) -> None: + vid = _add(report_state, "SQLi") + md = report_state.get_run_dir() / "vulnerabilities" / f"{vid}.md" + assert md.exists() + report_state.retract_vulnerability_report(vid, reason="fixed") + assert not md.exists() + + +# --- retract tool groundedness guard ----------------------------------------- +# NOTE: the ``os.system(...)`` / ``subprocess`` strings below are INERT test +# data — literal file-content and fix_before needles fed to the guard's text +# substring check. Nothing is executed; no shell is invoked. + + +async def _guard(report: dict[str, object]) -> tuple[bool, str]: + return await retract_tool._guard(report) + + +async def test_guard_refuses_without_reader() -> None: + retract_tool.set_target_file_reader(None) + allow, detail = await _guard( + {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, + ) + assert allow is False + assert "no target-file reader" in detail + + +async def test_guard_refuses_when_sink_still_present() -> None: + async def reader(_path: str) -> str: + return "def f(x):\n os.system(x)\n" # vulnerable line still there + + retract_tool.set_target_file_reader(reader) + allow, detail = await _guard( + {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, + ) + assert allow is False + assert "still present" in detail + + +async def test_guard_allows_when_sink_gone() -> None: + async def reader(_path: str) -> str: + return "def f(x):\n subprocess.run(['echo', x])\n" # fixed + + retract_tool.set_target_file_reader(reader) + allow, _ = await _guard( + {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, + ) + assert allow is True + + +async def test_guard_allows_when_no_verifiable_sinks() -> None: + async def reader(_path: str) -> str: + return "irrelevant" + + retract_tool.set_target_file_reader(reader) + # No fix_before → nothing to verify → don't fabricate a block. + allow, _ = await _guard({"code_locations": [{"file": "a.py", "snippet": "context only"}]}) + assert allow is True From 986c974d3d90ddf98c67d1753b5b0f4674832b8b Mon Sep 17 00:00:00 2001 From: Sean Turner Date: Wed, 15 Jul 2026 16:25:23 +0100 Subject: [PATCH 2/2] review: trust-model guard, id high-water persistence, reader hygiene, docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address Greptile review on the retract PR: - Trust model (the guard never vetoes): Strix trusts the LLM to find a vuln, so it trusts the LLM's judgement that it's fixed. The retract now ALWAYS proceeds on the agent's cited reason; the guard is an audit signal, not a gate. It returns a 3-way verdict — gone / present / inconclusive — recorded on the result as `grounded` + `guard_verdict`. This fixes the previous fail-OPEN paths (no fix_before sink, failed read → silently allowed) AND the opposite fail-CLOSED risk (stubbornly refusing a genuine fix on a moved file or a finding with no structured sink). A wrong drop is self-correcting: findings are re-derived every scan, so it re-appears next run. - id high-water mark now PERSISTS (run_record["max_vuln_ordinal"], restored by hydrate_from_run_dir). Without it a retracted top id was reused after a resume in a fresh process — _saved_vuln_ids is rebuilt only from live reports on rehydration. (Greptile issue 4.) - reader hygiene: the sandbox reader now returns None ONLY on positively- confirmed absence ("No such file") and RAISES on any other failed/ambiguous read, so "couldn't read" is scored inconclusive, never mistaken for "gone" (issue 2). Lifted to a module-level `_make_workspace_file_reader` (testable, and no longer a 50-line closure in the scan body). The reader is reset to None in the run's finally block so a torn-down session can't leak into a later scan's grounding (issue 3). - docs: STRIX_RESUME_RETRACT in configuration.mdx; a "Resuming a scan" + "Retracting fixed findings" section in usage/cli.mdx documenting the opt-in and the trust model. Tests updated to the 3-way verdict + tool-always-retracts behaviour; 60 tests pass (14 retract + 46 existing), ruff clean. --- docs/advanced/configuration.mdx | 4 + docs/usage/cli.mdx | 51 +++++++ strix/core/runner.py | 89 +++++++++---- strix/report/state.py | 46 ++++--- strix/tools/reporting/retract_tool.py | 185 +++++++++++++++++--------- tests/test_retract_vulnerability.py | 105 ++++++++++++--- 6 files changed, 359 insertions(+), 121 deletions(-) diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index 5d56c9a16..3cf580dea 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -45,6 +45,10 @@ Configure Strix using environment variables or a config file. Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL). + + Enable retraction of fixed findings on a resumed scan. Off by default, so `--resume` is append-only (findings only carry forward). When set to `1`/`true`, a resumed whitebox scan gains the `retract_vulnerability_report` tool: if the agent determines a rehydrated finding has been fixed, it is dropped from the cumulative report and SARIF so the resolved finding stops re-appearing. See [Resuming scans](/usage/cli#resuming-a-scan). + + OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only. diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 68f7d5edd..562c4105b 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -84,6 +84,15 @@ strix (--target | --target-list | --mount ) [options] counts. + + Resume a prior scan by its run name (the directory under `./strix_runs/`). + Restores the root agent plus every non-terminal subagent's full LLM history + and topology, and rehydrates prior findings so the report and SARIF are + cumulative across runs. Cannot be combined with `--target`/`--target-list`/`--mount` + (the original targets are restored from the prior run). See + [Resuming a scan](#resuming-a-scan). + + ## Examples ```bash @@ -112,6 +121,48 @@ strix --target-list ./targets.txt strix --mount ./huge-monorepo ``` +## Resuming a scan + +`--resume ` continues a prior scan: it restores the full agent history +and rehydrates the findings from the previous run, so the report and SARIF are +**cumulative** across runs. This suits re-scanning a target as it changes — for +example re-running against each push of a pull request. + +```bash +# First scan of the code +strix --run-name pr-42 -t ./repo + +# Later, after new commits land — pick up where it left off +strix --resume pr-42 +``` + +By default a resumed scan is **append-only**: prior findings carry forward, and +new issues are added, but nothing is removed. If the target has changed such that +a previously-reported finding is now **fixed**, it still re-appears in the +cumulative SARIF — which a code-scanning integration cannot auto-resolve. + +### Retracting fixed findings + +Set `STRIX_RESUME_RETRACT=1` to let a resumed **whitebox** scan drop findings it +determines are fixed. This exposes a `retract_vulnerability_report` tool: when +the agent judges a rehydrated finding resolved (the vulnerable code was fixed or +removed), it is removed from the cumulative report and SARIF, so a code-scanning +alert can auto-resolve. + +```bash +STRIX_RESUME_RETRACT=1 strix --resume pr-42 +``` + + + Retraction trusts the agent's judgement — the same judgement used to find the + vulnerability in the first place. Each retraction is re-checked against the + current source where possible and the result is recorded (`grounded` on the + tool result) for auditability, but the agent's decision is not overridden. + Because findings are re-derived on every scan, a mistaken retraction is + self-correcting: it re-appears on the next scan of the same code. The feature + is off by default; `--resume` stays append-only unless you opt in. + + ## Exit Codes | Code | Meaning | diff --git a/strix/core/runner.py b/strix/core/runner.py index deb6d11df..c74664714 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -98,6 +98,62 @@ def _compose_root_instructions_override( ) +def _make_workspace_file_reader( + session: Any, + ws_paths: list[str], +) -> Callable[[str], Any]: + """Build the target-file reader the retract guard uses to re-verify a fix. + + Fail-honest contract (the guard scores the result, see retract_tool): + - returns the file text when a candidate path is read; + - returns ``None`` ONLY when the file is POSITIVELY confirmed absent under + EVERY workspace root (``cat`` -> "No such file"); the sink is gone; + - RAISES when a read neither returns content nor confirms absence (exec + error, or an ambiguous nonzero like permission-denied). "Couldn't read" + is scored inconclusive by the guard, never mistaken for "confirmed gone" + — so a misdirected read can't masquerade as a verified fix. + """ + + async def _read_target_file(rel_path: str) -> str | None: + rel = rel_path.lstrip("/") + confirmed_absent_everywhere = True + errors: list[str] = [] + for base in ws_paths: + path = f"{base}/{rel}" + try: + result = await session.exec("cat", path, timeout=15) + except Exception as exc: # noqa: BLE001 — read failure ≠ proof of absence + confirmed_absent_everywhere = False + errors.append(f"{path}: exec error {exc!r}") + continue + if getattr(result, "exit_code", 1) == 0: + # ExecResult.stdout is bytes; the guard does a str substring + # check, so decode here. + stdout = result.stdout + return ( + stdout.decode("utf-8", errors="replace") + if isinstance(stdout, bytes | bytearray) + else stdout + ) + stderr = getattr(result, "stderr", b"") or b"" + stderr_txt = ( + stderr.decode("utf-8", errors="replace") + if isinstance(stderr, bytes | bytearray) + else str(stderr) + ) + # Only "No such file" is a POSITIVE absence signal. Any other nonzero + # (permission denied, is-a-directory) is NOT proof of a fix. + if "No such file" not in stderr_txt: + confirmed_absent_everywhere = False + errors.append(f"{path}: exit {result.exit_code}: {stderr_txt.strip()[:120]}") + if confirmed_absent_everywhere: + return None # genuinely absent under every root → sink is gone + msg = f"could not read {rel} for retract verification: {'; '.join(errors)}" + raise OSError(msg) + + return _read_target_file + + async def run_strix_scan( *, scan_config: dict[str, Any], @@ -244,8 +300,7 @@ async def run_strix_scan( if is_resume and is_whitebox and settings.runtime.resume_retract_enabled: from strix.tools.reporting.retract_tool import set_target_file_reader - _session = bundle["session"] - _ws_paths = [ + ws_paths = [ f"/workspace/{sub}" if sub else "/workspace" for sub in ( (t.get("details") or {}).get("workspace_subdir") @@ -253,28 +308,7 @@ async def run_strix_scan( if t.get("type") == "local_code" ) ] or ["/workspace"] - - async def _read_target_file(rel_path: str) -> str | None: - rel = rel_path.lstrip("/") - for base in _ws_paths: - try: - result = await _session.exec("cat", f"{base}/{rel}", timeout=15) - except Exception as exc: # noqa: BLE001 — unreadable ≠ proof of absence - logger.debug("retract reader: %s/%s unreadable: %r", base, rel, exc) - continue - if getattr(result, "exit_code", 1) == 0: - # ExecResult.stdout is bytes; the guard does a str - # substring check, so decode here (a raw bytes return - # would raise TypeError on every retract). - stdout = result.stdout - return ( - stdout.decode("utf-8", errors="replace") - if isinstance(stdout, bytes | bytearray) - else stdout - ) - return None # not found under any workspace root → treat as gone - - set_target_file_reader(_read_target_file) + set_target_file_reader(_make_workspace_file_reader(bundle["session"], ws_paths)) root_agent = build_strix_agent( name="strix", @@ -438,6 +472,13 @@ async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]: await coordinator.set_status(root_id, "failed") raise finally: + # Clear the retract reader so it can't bind to a torn-down session and + # verify a later scan against the wrong (or dead) sandbox. The guard + # then fails safe (no reader → refuse) until the next resume re-wires it. + with contextlib.suppress(Exception): + from strix.tools.reporting.retract_tool import set_target_file_reader + + set_target_file_reader(None) for s in sessions_to_close: with contextlib.suppress(Exception): s.close() diff --git a/strix/report/state.py b/strix/report/state.py index f3f91b6a5..a31f9c788 100644 --- a/strix/report/state.py +++ b/strix/report/state.py @@ -206,6 +206,15 @@ def hydrate_from_run_dir(self) -> None: len(self.vulnerability_reports), ) + @staticmethod + def _vuln_ordinal(report_id: object) -> int: + if isinstance(report_id, str) and report_id.startswith("vuln-"): + try: + return int(report_id.removeprefix("vuln-")) + except ValueError: + return 0 + return 0 + def _next_vuln_ordinal(self) -> int: """Next ``vuln-NNNN`` ordinal, derived from the max ordinal EVER seen rather than the current list length. @@ -216,22 +225,23 @@ def _next_vuln_ordinal(self) -> int: recycle: retract ``vuln-0002`` from ``[0001,0002,0003]`` and the next add computes ``len+1 = 3`` -> ``vuln-0003``, colliding with a live id and overwriting its on-disk MD, and letting a retracted id later re-emit - under a different finding (SARIF fingerprint collision). Deriving from - the max ordinal across both the live reports and every id ever saved - (``_saved_vuln_ids`` retains retracted ids) keeps ids monotonic and - stable under removal. + under a different finding (SARIF fingerprint collision). + + The high-water mark is persisted in ``run_record["max_vuln_ordinal"]`` + and restored by :meth:`hydrate_from_run_dir`, so it survives across the + process boundary of a resume. Without that, a NEW process rebuilds + ``_saved_vuln_ids`` only from the live ``vulnerabilities.json`` — the + retracted id is gone from disk, so the next allocation would reuse it. + We take the max of the persisted mark and every ordinal currently in + memory (live reports + retained ``_saved_vuln_ids``), so it stays correct + even if the record is missing/stale. """ - max_ordinal = 0 - for rid in ( - *(r.get("id") for r in self.vulnerability_reports), - *self._saved_vuln_ids, - ): - if isinstance(rid, str) and rid.startswith("vuln-"): - try: - max_ordinal = max(max_ordinal, int(rid.removeprefix("vuln-"))) - except ValueError: - continue - return max_ordinal + 1 + candidates = [ + int(self.run_record.get("max_vuln_ordinal") or 0), + *(self._vuln_ordinal(r.get("id")) for r in self.vulnerability_reports), + *(self._vuln_ordinal(rid) for rid in self._saved_vuln_ids), + ] + return max(candidates) + 1 def add_vulnerability_report( self, @@ -260,7 +270,11 @@ def add_vulnerability_report( agent_id: str | None = None, agent_name: str | None = None, ) -> str: - report_id = f"vuln-{self._next_vuln_ordinal():04d}" + ordinal = self._next_vuln_ordinal() + report_id = f"vuln-{ordinal:04d}" + # Persist the high-water mark so a retracted top id is never reused after + # a resume in a fresh process (run_record survives via run.json). + self.run_record["max_vuln_ordinal"] = ordinal report: dict[str, Any] = { "id": report_id, diff --git a/strix/tools/reporting/retract_tool.py b/strix/tools/reporting/retract_tool.py index 07f99a2f0..837dff38a 100644 --- a/strix/tools/reporting/retract_tool.py +++ b/strix/tools/reporting/retract_tool.py @@ -3,25 +3,44 @@ On ``--resume`` the agent rehydrates every prior finding. When a later push FIXES one, the agent must be able to DROP it, or rehydration is append-only and the fixed finding re-emits into every subsequent SARIF forever — an alert that -can never auto-resolve (fail-CLOSED). This tool closes that gap. - -GROUNDEDNESS GUARD (the load-bearing safety property): the tool does not blindly -trust the agent's "it's fixed" claim. Before retracting, it re-verifies against -the current tree via an injected ``read_target_file`` — recovering the finding's -fix-site (sink) locations and checking whether the vulnerable code is STILL -present. If a sink is still there verbatim, the retraction is REFUSED (a -confident-but-wrong agent must not be able to silently drop a real finding = -fail-open). Missing/deleted locations allow the retract (don't fabricate a block -from absent data). +can never auto-resolve. This tool closes that gap. + +TRUST MODEL — the agent decides, the guard only annotates. Strix trusts the LLM +to FIND vulnerabilities, so it trusts the LLM's judgement that one is FIXED. The +retraction therefore ALWAYS proceeds on the agent's cited reason; the guard +never vetoes it. What the guard adds is an audit signal: it re-reads the +finding's fix-site (``fix_before``) sinks against the current tree and records a +verdict on the result, so a drop is explainable and recoverable — not a +decision-maker that can stubbornly block a genuine fix it merely can't +re-locate (moved/renamed file, dependency finding with no sink, transient read +failure). Recorded on the result: + +- ``grounded=True`` (``guard_verdict="gone"``): the sink was read and the + vulnerable line is gone (or the file is confirmed absent). The fix is + independently corroborated. +- ``grounded=False`` (``guard_verdict="inconclusive"``): couldn't corroborate — + no reader wired, no ``fix_before`` sink on the finding, or the sink file + couldn't be read. Retract still honoured on the agent's word. +- ``grounded=False`` (``guard_verdict="present"``): the ``fix_before`` line is + STILL present verbatim, which contradicts the "fixed" claim — logged as a loud + warning, but the agent may have legitimately re-reasoned (line now unreachable, + guarded, or the fixed form contains the substring), so the retract is honoured + and flagged rather than blocked. + +A wrong retraction is self-correcting: findings are re-derived on every scan of +the code, so a mistakenly-dropped-but-still-live finding re-appears next run. +That recoverability is what makes trusting the agent safe here. The reader is injected rather than hard-wired: the target lives inside the sandbox in v1, so the concrete sandbox-backed reader is supplied by the caller -(the agent factory, which holds the session). When no reader is available the -tool FAILS SAFE — it refuses to retract rather than allowing an unverified drop -(a retract we can't ground is more dangerous than a finding that survives). - -Only exposed to the agent on resumed runs (see factory gating) — a fresh scan -can't need it and shouldn't be handed a findings-drop tool. +(the agent factory, which holds the session). Its contract: return file text if +read, ``None`` ONLY when the file is positively confirmed absent, and RAISE when +a read neither returns content nor confirms absence — so "couldn't read" is +scored ``inconclusive``, never mistaken for "confirmed gone". + +Only exposed to the agent on resumed runs, and only when +``STRIX_RESUME_RETRACT`` is set (see factory gating) — a fresh scan can't need +it, and resume stays append-only by default. """ from __future__ import annotations @@ -48,7 +67,9 @@ def set_target_file_reader(reader: ReadTargetFileFn | None) -> None: """Wire the sandbox-backed target-file reader (called at agent build time on - resume). Passing None disables grounded retraction (fail-safe refuse).""" + resume). Passing None disables source grounding — retracts still proceed on + the agent's reason, scored ``inconclusive`` (ungrounded) rather than + ``gone``.""" _reader[0] = reader @@ -63,51 +84,61 @@ def _sink_locations(report: dict[str, Any]) -> list[dict[str, Any]]: ] -async def _guard(report: dict[str, Any]) -> tuple[bool, str]: - """Return (allow, detail). allow=False → REFUSE the retract. - - - no reader wired → REFUSE (fail-safe: can't ground → don't drop). - - no fix_before sinks → ALLOW (nothing verifiable; don't fabricate a block). - - any sink's fix_before still present verbatim → REFUSE. - - otherwise → ALLOW. +# Three-way guard verdict. The point is to separate "the agent is provably +# WRONG" (hard block) from "we can't independently confirm" (defer to the +# agent's judgement but flag it) — so the tool neither silently drops live +# findings NOR stubbornly refuses a genuine fix it just can't re-locate (moved +# file, no structured sink, transient read failure). +GUARD_PRESENT = "present" # sink confirmed still in source -> hard REFUSE +GUARD_GONE = "gone" # sink/file confirmed absent -> grounded ALLOW +GUARD_INCONCLUSIVE = "inconclusive" # couldn't verify -> allow on agent's word, flagged + + +async def _guard(report: dict[str, Any]) -> tuple[str, str]: + """Re-verify a retract against current source. Returns ``(verdict, detail)``. + + - ``GUARD_PRESENT``: a ``fix_before`` sink is still present verbatim — the + "it's fixed" claim is provably false. Hard refuse. + - ``GUARD_GONE``: every verifiable sink's file was read and the vulnerable + line is gone (or the file is confirmed absent). Grounded — allow. + - ``GUARD_INCONCLUSIVE``: nothing could be positively checked — no reader, + no ``fix_before`` sink (older/dependency findings), or the file couldn't + be read (moved/renamed/permission). NOT proof either way; the tool allows + on the agent's cited reason but records it as ungrounded so a wrong drop + is auditable and a later re-scan re-adds it. """ reader = _reader[0] if reader is None: - return False, "no target-file reader wired — refusing ungrounded retract" + return GUARD_INCONCLUSIVE, "no target-file reader wired — cannot ground the retract" sinks = _sink_locations(report) if not sinks: - return True, "no fix-site (fix_before) locations to verify — allowing" + return GUARD_INCONCLUSIVE, "no fix_before sink on this finding — nothing to verify" + + verified_gone = False + unread: list[str] = [] for loc in sinks: try: - content = await reader(str(loc["file"])) - except Exception: # noqa: BLE001 — reader failure ≠ proof of absence - return False, f"could not read {loc['file']} to verify — refusing" - if content is None: - continue # file gone → this sink can't be present + result = await reader(str(loc["file"])) + except Exception as exc: # noqa: BLE001 — a failed read is not proof either way + unread.append(f"{loc['file']} ({exc!r})") + continue needle = str(loc["fix_before"]).strip() - if needle and needle in content: - return False, (f"vulnerable code still present at {loc['file']}: " - f"{needle[:80]!r}") - return True, "vulnerable code no longer present at any verified sink" - - -@function_tool(timeout=120, strict_mode=False) -async def retract_vulnerability_report( - ctx: RunContextWrapper, - report_id: str, - reason: str, -) -> dict[str, Any]: - """Retract a finding you have RE-VERIFIED as fixed in the current code. - - Use ONLY when a prior finding (rehydrated on resume) has been genuinely - fixed since it was reported. Cite what changed and where in ``reason``. The - retraction is re-verified against the current tree; if the vulnerable code - is still present, it is refused. - - Args: - report_id: the finding id to retract (e.g. ``vuln-0003``). - reason: code-grounded justification — what was fixed, and where. - """ + if result is None: + verified_gone = True # file confirmed absent → sink can't be present + continue + if needle and needle in result: + return GUARD_PRESENT, (f"vulnerable code still present at {loc['file']}: " + f"{needle[:80]!r}") + verified_gone = True # read the file, vulnerable line is gone + + if unread and not verified_gone: + # Every sink was unreadable (e.g. the file moved) — can't confirm a fix, + # but can't confirm it's still vulnerable either. + return GUARD_INCONCLUSIVE, f"could not read any sink to verify: {'; '.join(unread)}" + return GUARD_GONE, "vulnerable code no longer present at any readable sink" + + +async def _do_retract(report_id: str, reason: str) -> dict[str, Any]: reason = (reason or "").strip() if not reason: return {"success": False, @@ -125,16 +156,50 @@ async def retract_vulnerability_report( if report is None: return {"success": True, "retracted": False, "reason": "id not present"} - allow, detail = await _guard(report) - if not allow: - logger.warning("retract REFUSED for %s: %s", report_id, detail) - return {"success": False, "retracted": False, "refused": True, - "error": f"retraction refused — {detail}"} + # The guard NEVER overrides the agent: Strix trusts the LLM to find the + # vulnerability, so it trusts the LLM's judgement that it's fixed. The guard + # is an audit signal, not a veto — it records whether the drop could be + # grounded in source and warns loudly when the agent's "fixed" claim looks + # contradicted, but the retract always proceeds on the agent's cited reason. + # A wrong drop re-appears on the next scan of the same code (findings are + # re-derived every run), so this is recoverable, not silent data loss. + verdict, detail = await _guard(report) + grounded = verdict == GUARD_GONE + if verdict == GUARD_PRESENT: + logger.warning( + "retract for %s: guard says vulnerable sink still present (%s), but " + "honouring agent reason: %s", + report_id, detail, reason, + ) + elif not grounded: + logger.info("retract for %s ungrounded (%s) — honouring agent reason", report_id, detail) try: result = state.retract_vulnerability_report(report_id, reason) except (ValueError, AttributeError) as e: return {"success": False, "error": f"retract failed: {e!s}"} - logger.info("retract honoured for %s (guard: %s)", report_id, detail) + logger.info("retract honoured for %s (grounded=%s, guard: %s)", report_id, grounded, detail) result["guard"] = detail + result["grounded"] = grounded + result["guard_verdict"] = verdict return result + + +@function_tool(timeout=120, strict_mode=False) +async def retract_vulnerability_report( + ctx: RunContextWrapper, + report_id: str, + reason: str, +) -> dict[str, Any]: + """Retract a prior finding you have determined is FIXED in the current code. + + Use on a resumed scan when a finding rehydrated from a previous run is no + longer valid — the vulnerable code has been fixed or removed since it was + reported. It is dropped from the cumulative report and SARIF so the resolved + finding stops re-appearing. Cite what changed and where in ``reason``. + + Args: + report_id: the finding id to retract (e.g. ``vuln-0003``). + reason: what was fixed, and where. + """ + return await _do_retract(report_id, reason) diff --git a/tests/test_retract_vulnerability.py b/tests/test_retract_vulnerability.py index 519a89a44..19413172c 100644 --- a/tests/test_retract_vulnerability.py +++ b/tests/test_retract_vulnerability.py @@ -81,47 +81,110 @@ def test_retracted_md_is_deleted(report_state: ReportState) -> None: # substring check. Nothing is executed; no shell is invoked. -async def _guard(report: dict[str, object]) -> tuple[bool, str]: +async def _guard(report: dict[str, object]) -> tuple[str, str]: return await retract_tool._guard(report) -async def test_guard_refuses_without_reader() -> None: +_SINK = {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]} + + +async def test_guard_inconclusive_without_reader() -> None: retract_tool.set_target_file_reader(None) - allow, detail = await _guard( - {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, - ) - assert allow is False + verdict, detail = await _guard(_SINK) + assert verdict == retract_tool.GUARD_INCONCLUSIVE assert "no target-file reader" in detail -async def test_guard_refuses_when_sink_still_present() -> None: +async def test_guard_present_when_sink_still_there() -> None: async def reader(_path: str) -> str: - return "def f(x):\n os.system(x)\n" # vulnerable line still there + return "def f(x):\n os.system(x)\n" # vulnerable line still present retract_tool.set_target_file_reader(reader) - allow, detail = await _guard( - {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, - ) - assert allow is False + verdict, detail = await _guard(_SINK) + assert verdict == retract_tool.GUARD_PRESENT assert "still present" in detail -async def test_guard_allows_when_sink_gone() -> None: +async def test_guard_gone_when_sink_removed() -> None: async def reader(_path: str) -> str: return "def f(x):\n subprocess.run(['echo', x])\n" # fixed retract_tool.set_target_file_reader(reader) - allow, _ = await _guard( - {"code_locations": [{"file": "a.py", "fix_before": "os.system(x)"}]}, - ) - assert allow is True + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_GONE + + +async def test_guard_gone_when_file_confirmed_absent() -> None: + async def reader(_path: str) -> None: + return None # reader confirms the file is gone + + retract_tool.set_target_file_reader(reader) + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_GONE -async def test_guard_allows_when_no_verifiable_sinks() -> None: +async def test_guard_inconclusive_when_no_fix_before() -> None: async def reader(_path: str) -> str: return "irrelevant" retract_tool.set_target_file_reader(reader) - # No fix_before → nothing to verify → don't fabricate a block. - allow, _ = await _guard({"code_locations": [{"file": "a.py", "snippet": "context only"}]}) - assert allow is True + verdict, _ = await _guard({"code_locations": [{"file": "a.py", "snippet": "context only"}]}) + assert verdict == retract_tool.GUARD_INCONCLUSIVE + + +async def test_guard_inconclusive_when_read_raises() -> None: + async def reader(_path: str) -> str: + raise OSError("could not read — file moved") + + retract_tool.set_target_file_reader(reader) + # A failed read (e.g. moved/renamed file) is NOT proof of a fix, but also + # not proof it's still vulnerable → inconclusive, not present. + verdict, _ = await _guard(_SINK) + assert verdict == retract_tool.GUARD_INCONCLUSIVE + + +# --- tool always retracts (trust model) + records grounding ------------------ + + +async def test_tool_retracts_and_marks_grounded_when_gone(report_state: ReportState) -> None: + vid = report_state.add_vulnerability_report( + title="CmdInj", severity="high", + code_locations=[{"file": "a.py", "fix_before": "os.system(x)"}], + ) + + async def reader(_path: str) -> str: + return "subprocess.run(['echo', x])\n" # fixed + + retract_tool.set_target_file_reader(reader) + res = await retract_tool._do_retract(vid, reason="rewrote to subprocess argv") + assert res["retracted"] is True + assert res["grounded"] is True + assert vid not in [r["id"] for r in report_state.vulnerability_reports] + + +async def test_tool_still_retracts_when_sink_present_but_flags_ungrounded( + report_state: ReportState, +) -> None: + # Guard NEVER vetoes: even when the fix_before line is still present, the + # agent's judgement wins — the drop proceeds but is flagged grounded=False. + vid = report_state.add_vulnerability_report( + title="CmdInj", severity="high", + code_locations=[{"file": "a.py", "fix_before": "os.system(x)"}], + ) + + async def reader(_path: str) -> str: + return "os.system(x) # now behind an unreachable guard\n" + + retract_tool.set_target_file_reader(reader) + res = await retract_tool._do_retract(vid, reason="line now unreachable — dead branch") + assert res["retracted"] is True + assert res["grounded"] is False + assert res["guard_verdict"] == retract_tool.GUARD_PRESENT + assert vid not in [r["id"] for r in report_state.vulnerability_reports] + + +async def test_tool_requires_reason(report_state: ReportState) -> None: + vid = report_state.add_vulnerability_report(title="X", severity="high") + res = await retract_tool._do_retract(vid, reason=" ") + assert res["success"] is False + assert vid in [r["id"] for r in report_state.vulnerability_reports]