Let agents delete a vulnerability report they filed (#1354)

This commit is contained in:
alex s 2026-09-23 20:25:23 -04:00 • committed by GitHub
parent 56f7d45388
commit 4c1be22150
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 688 additions and 59 deletions

View file

@ -56,6 +56,7 @@ from strix.tools.proxy.tools import (
from strix.tools.reporting.tool import (
create_dependency_report,
create_vulnerability_report,
delete_vulnerability_report,
get_report,
list_reports,
update_vulnerability_report,
@ -589,6 +590,7 @@ _BASE_TOOLS: tuple[Tool, ...] = (
create_vulnerability_report,
create_dependency_report,
update_vulnerability_report,
delete_vulnerability_report,
list_reports,
get_report,
list_requests,

View file

@ -244,6 +244,7 @@ VALIDATION REQUIREMENTS:
- DEDUPLICATION: The create_vulnerability_report tool uses LLM-based deduplication. If it rejects your report as a duplicate, DO NOT attempt to re-submit the same vulnerability. Accept the rejection and move on to testing other areas. The vulnerability has already been reported by another agent. If your evidence proves more than the finding it matched (a working exploit where that one had only a static trace, a chain that raises the impact), revise that finding with update_vulnerability_report using the duplicate_of id — never re-file it.
- HTTP EVIDENCE: a finding you validated through the proxy is not fully filed until `http_exchange_ids` carries the proxy request ids of the exchanges that prove it — the request that triggers the vulnerability plus the baseline/control request it differs from (an unauthenticated success next to the authenticated one, the payload response next to the benign one). Copy the ids exactly as `list_requests`/`view_request` show them, never invent or guess one, and never omit the field to bypass validation. Leave it out only when there is no captured HTTP exchange at all (static-only code findings, dependency CVEs). If you filed before the proving exchanges existed, attach them afterwards with update_vulnerability_report. Without the ids, the finding ships as prose nobody can replay.
- REVISING A FINDING: use update_vulnerability_report (report id + the fields you want to replace + update_reason) when you learn something a finding already on file does not carry — you built the PoC after filing it, a chain raised its impact, further testing weakened it, or its counterevidence/remediation/code locations were wrong. Editing a finding needs no duplicate verdict, and it is always better than filing a second report for the same issue. Read the finding first with get_report, and pass only the fields that change.
- WITHDRAWING A FINDING: when re-testing proves a finding you filed is not a vulnerability at all — the exploit only worked because of your own test setup (a mixed-up session, a self-inflicted state change, a misread response), the behaviour is intended, or the control is in fact enforced — delete it with delete_vulnerability_report (report id + delete_reason). Never leave a disproved finding on file as a zero-CVSS, "retracted", or "false positive" report, and never rewrite its text into a note saying it was wrong: a report that exists is read as a finding. Delete is for a finding that does not exist; a finding that is real but weaker than filed is revised, not deleted.
- REVIEWING FILED FINDINGS (orchestrator/root agent): use list_reports to see every vulnerability filed so far in this scan (by any agent, root or child) — metadata-first with per-severity counts — and get_report to read one finding in full by its id. These are read-only orchestration tools: the root agent uses them to track coverage, avoid dispatching work on already-covered ground, assemble the finish_scan executive summary, and reason about attack-chaining across confirmed findings. Leaf/specialist agents should NOT call them — just do your assigned testing and file findings. Each entry shows which agent filed it (agent_name), and your own entries are flagged by_you. list_notes/get_note do the same for notes.
STATE & COORDINATION TOOLS (when and how):

View file

@ -122,10 +122,32 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
console.print(vuln_panel)
console.print()
def display_vulnerability_deleted(report: dict[str, Any]) -> None:
report_id = str(report.get("id", "unknown"))
deletion = report.get("deletion")
deletion = deletion if isinstance(deletion, dict) else {}
deleted_by = deletion.get("agent_name") or deletion.get("agent_id") or "agent"
text = Text()
text.append("Withdrawn: ", style="bold")
text.append(f"{report.get('title', '')}\n\n")
text.append(f"Deleted by {deleted_by}. ", style="dim")
text.append(str(deletion.get("reason") or ""))
console.print(
Panel(
text,
title=f"[bold yellow]{report_id.upper()} (withdrawn)",
title_align="left",
border_style="yellow",
padding=(1, 2),
)
)
console.print()
report_state.vulnerability_found_callback = display_vulnerability
report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
report, updated=True
)
report_state.vulnerability_deleted_callback = display_vulnerability_deleted
def cleanup_on_exit() -> None:
report_state.cleanup()

View file

@ -86,6 +86,8 @@ func Tool(data map[string]any) string {
return renderVulnerabilityReport(args, result)
case "update_vulnerability_report":
return renderVulnerabilityReportUpdate(args, result)
case "delete_vulnerability_report":
return renderVulnerabilityReportDelete(args, result)
case "create_dependency_report":
return renderDependencyReport(args, result)
case "list_reports":

View file

@ -21,6 +21,36 @@ func renderVulnerabilityReportUpdate(args map[string]any, result any) string {
return renderReport(args, result, "Vulnerability Report Updated", "Updating report...")
}
// A deletion names the report it withdraws and the reason; the result carries
// the title the report had, so the reader knows what left the scan.
func renderVulnerabilityReportDelete(args map[string]any, result any) string {
resultMap, _ := result.(map[string]any)
var b strings.Builder
b.WriteString("🐞 " + Bold(ReportHdr).Render("Vulnerability Report Deleted"))
reportID := StringValue(args["report_id"])
if reportID != "" {
b.WriteString("\n\n" + Bold(Field).Render("Report: ") + reportID)
}
if title := StringValue(resultMap["title"]); title != "" {
b.WriteString("\n\n" + Bold(Field).Render("Title: ") + title)
}
if sev := StringValue(resultMap["severity"]); sev != "" {
b.WriteString("\n\n" + Bold(Field).Render("Severity: ") +
lipgloss.NewStyle().Bold(true).Foreground(SeverityColor(sev)).Render(strings.ToUpper(sev)))
}
if reason := StringValue(args["delete_reason"]); reason != "" {
b.WriteString("\n\n" + Bold(Field).Render("Reason") + "\n" + reason)
}
if errMsg := StringValue(resultMap["error"]); errMsg != "" {
b.WriteString("\n\n" + Col(SevHigh).Render(errMsg))
}
if reportID == "" {
b.WriteString("\n " + Dim().Render("Deleting report..."))
}
return "\n\n" + b.String() + "\n\n"
}
func renderReport(args map[string]any, result any, heading, pending string) string {
resultMap, _ := result.(map[string]any)
var b strings.Builder

View file

@ -114,6 +114,9 @@ class GoTuiRuntime:
self.report_state.vulnerability_updated_callback = lambda _report: (
self.controller.notify_changed()
)
self.report_state.vulnerability_deleted_callback = lambda _report: (
self.controller.notify_changed()
)
self.controller.notify_changed()
async def check_setup_model(self) -> None:

View file

@ -0,0 +1,49 @@
"use client";
import type { ToolRendererProps } from "@/types/events";
import { TruncatedText } from "./ToolCard";
const SEVERITY_COLORS: Record<string, string> = {
critical: "text-red-400", high: "text-orange-400", medium: "text-yellow-400",
low: "text-blue-400", info: "text-cyan-400",
};
/** A withdrawn finding: which report went, and what disproved it. */
export default function VulnReportDeleteRenderer({ args, result, status }: ToolRendererProps) {
const reportId = (args.report_id as string) ?? "";
const reason = (args.delete_reason as string) ?? "";
const res = result && typeof result === "object" ? (result as Record<string, unknown>) : null;
const ok = res?.success === true;
const error = typeof res?.error === "string" ? res.error : "";
const title = typeof res?.title === "string" ? res.title : "";
const severity = typeof res?.severity === "string" ? res.severity.toLowerCase() : "";
const failed = res != null ? !ok : status === "failed" || status === "error";
const pending = res == null && !failed;
const label = pending ? "withdrawing report\u2026" : failed ? "report not withdrawn" : "report withdrawn";
const labelColor = pending ? "text-[#888]" : failed ? "text-red-400/80" : "text-yellow-400/80";
return (
<div className="space-y-2">
<div className="flex items-center gap-2 flex-wrap">
<span className={`font-semibold text-sm ${labelColor}`} aria-live="polite">
{label}
</span>
{reportId && <span className="text-[#555] font-mono text-[13px]">{reportId}</span>}
{severity && (
<span className={`text-[13px] line-through ${SEVERITY_COLORS[severity] ?? "text-[#888]"}`}>
{severity.toUpperCase()}
</span>
)}
</div>
{title && <div className="text-[15px] text-white/60 line-through">{title}</div>}
{reason && (
<div>
<span className="text-emerald-400/60 text-sm font-semibold">Why</span>
<div className="mt-1"><TruncatedText text={reason} maxLines={10} /></div>
</div>
)}
{error && <div className="text-red-400/80 text-[13px]">{error}</div>}
</div>
);
}

View file

@ -12,6 +12,7 @@ import FileEditRenderer from "./FileEditRenderer";
import ApplyPatchRenderer from "./ApplyPatchRenderer";
import ViewImageRenderer from "./ViewImageRenderer";
import VulnReportRenderer from "./VulnReportRenderer";
import VulnReportDeleteRenderer from "./VulnReportDeleteRenderer";
import ReportListRenderer from "./ReportListRenderer";
import ProxyRenderer from "./ProxyRenderer";
import ThinkRenderer from "./ThinkRenderer";
@ -116,7 +117,7 @@ const CATEGORY_TOOLS: Record<ToolCategory, readonly string[]> = {
filesystem: ["apply_patch", "view_image", "str_replace_editor", "list_files", "search_files"],
// Caido proxy tools (legacy: send_request)
proxy: ["list_requests", "view_request", "repeat_request", "list_sitemap", "view_sitemap_entry", "scope_rules", "send_request"],
reporting: ["create_vulnerability_report", "update_vulnerability_report", "list_reports", "get_report"],
reporting: ["create_vulnerability_report", "update_vulnerability_report", "delete_vulnerability_report", "list_reports", "get_report"],
thinking: ["think"],
agents: ["create_agent", "agent_finish", "send_message_to_agent", "wait_for_agents", "view_agent_graph", "stop_agent"],
search: ["web_search"],
@ -151,6 +152,7 @@ const RENDERER_OVERRIDES: Partial<Record<string, ComponentType<ToolRendererProps
view_image: ViewImageRenderer,
list_reports: ReportListRenderer,
get_report: ReportListRenderer,
delete_vulnerability_report: VulnReportDeleteRenderer,
};
/**

View file

@ -6,8 +6,8 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" />
<title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-B94ANU8d.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-DN__rVv3.css">
<script type="module" crossorigin src="./assets/index-DbPDzbcJ.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-CrLE9rrV.css">
</head>
<body>
<div id="root"></div>

View file

@ -36,6 +36,23 @@ _global_report_state: Optional["ReportState"] = None
_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]+")
class ReportRollbackError(RuntimeError):
"""Raised when a failed deletion could not rewrite the on-disk indexes.
The report is still on file in memory; the indexes are rewritten from
memory on the next save.
"""
def __init__(self, report_id: str, *, cause: BaseException) -> None:
self.report_id = report_id
self.cause = cause
super().__init__(
f"Deletion of report '{report_id}' failed ({cause}) and the on-disk indexes "
"could not be restored; the report is still on file and the indexes are "
"rewritten on the next save"
)
def _strix_version() -> str | None:
"""Best-effort package version for the SARIF tool.driver.version field."""
try:
@ -221,6 +238,7 @@ class ReportState:
self.caido_url: str | None = None
self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None
self.vulnerability_updated_callback: Callable[[dict[str, Any]], None] | None = None
self.vulnerability_deleted_callback: Callable[[dict[str, Any]], None] | None = None
self._sarif_repo_ctx: dict[str, Any] | None = None
self._sarif_repo_ctx_ready: bool = False
@ -342,7 +360,7 @@ class ReportState:
agent_id: str | None = None,
agent_name: str | None = None,
) -> str:
report_id = f"vuln-{len(self.vulnerability_reports) + 1:04d}"
report_id = self._next_report_id()
report: dict[str, Any] = {
"id": report_id,
@ -418,6 +436,24 @@ class ReportState:
self.save_run_data()
return report_id
def _deleted_vulnerability_reports(self) -> list[dict[str, Any]]:
raw = self.run_record.get("deleted_vulnerability_reports")
return [e for e in raw if isinstance(e, dict)] if isinstance(raw, list) else []
def _next_report_id(self) -> str:
"""Allocate the id after every id this run has ever handed out.
A deleted report leaves the list, so counting entries would hand its id
to the next finding and let that finding overwrite the deleted MD on disk
and inherit its history in every consumer that keys on the id.
"""
used = 0
for entry in [*self.vulnerability_reports, *self._deleted_vulnerability_reports()]:
match = re.fullmatch(r"vuln-(\d+)", str(entry.get("id", "")))
if match:
used = max(used, int(match.group(1)))
return f"vuln-{used + 1:04d}"
def update_vulnerability_report(
self,
report_id: str,
@ -516,6 +552,88 @@ class ReportState:
self.save_run_data()
return report
def delete_vulnerability_report(
self,
report_id: str,
*,
delete_reason: str,
deleted_by_agent_id: str | None = None,
deleted_by_agent_name: str | None = None,
) -> dict[str, Any] | None:
"""Withdraw a report from the run, keeping a record of the withdrawal.
Returns the removed report, or ``None`` when the id is unknown. Every
agent in a run shares one trust boundary, so any of them may withdraw
any report (just as any of them may revise one); the run record keeps
who filed it, who withdrew it and why. The report leaves
``vulnerability_reports`` and its rendered artifacts, and its id is
never reissued.
The rewritten artifacts and the ``vulnerability_deleted_callback`` must
both accept the deletion first: if either fails the report is put back
and the error propagates, so the deletion can be retried
(:class:`ReportRollbackError` when the indexes could not be put back).
"""
report = next((r for r in self.vulnerability_reports if r.get("id") == report_id), None)
if report is None:
logger.warning("cannot delete unknown vulnerability report %s", report_id)
return None
entry: dict[str, Any] = {
"id": report_id,
"title": report.get("title"),
"severity": report.get("severity"),
"filed_at": report.get("timestamp"),
"deleted_at": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC"),
"reason": delete_reason.strip()[:500],
}
if report.get("agent_id"):
entry["filed_by_agent_id"] = report["agent_id"]
if deleted_by_agent_id:
entry["agent_id"] = deleted_by_agent_id
if deleted_by_agent_name:
entry["agent_name"] = deleted_by_agent_name
position = self.vulnerability_reports.index(report)
was_saved = report_id in self._saved_vuln_ids
history = self._deleted_vulnerability_reports()
self.vulnerability_reports.remove(report)
self._saved_vuln_ids.discard(report_id)
self.run_record["deleted_vulnerability_reports"] = [*history, entry]
try:
# Local artifacts first: they can be put back if persistence then
# refuses, whereas a row deleted elsewhere cannot.
self._sync_llm_usage_record()
self._write_artifacts()
if self.vulnerability_deleted_callback:
self.vulnerability_deleted_callback({**report, "deletion": entry})
except Exception as exc:
self.vulnerability_reports.insert(position, report)
if was_saved:
self._saved_vuln_ids.add(report_id)
if history:
self.run_record["deleted_vulnerability_reports"] = history
else:
self.run_record.pop("deleted_vulnerability_reports", None)
try:
self._write_artifacts()
except Exception as rollback_exc:
logger.exception(
"could not restore artifacts after failed deletion of %s", report_id
)
raise ReportRollbackError(report_id, cause=exc) from rollback_exc
raise
md_path = self.get_run_dir() / "vulnerabilities" / f"{report_id}.md"
try:
md_path.unlink(missing_ok=True)
except OSError:
logger.exception("could not remove %s", md_path)
logger.info("Deleted vulnerability report %s - %s", report_id, report.get("title"))
return report
def get_existing_vulnerabilities(self) -> list[dict[str, Any]]:
return list(self.vulnerability_reports)
@ -699,47 +817,54 @@ class ReportState:
return None
def _save_artifacts(self) -> None:
"""Write scan artifacts under ``run_dir``."""
run_dir = self.get_run_dir()
"""Write scan artifacts under ``run_dir``; a write failure is logged."""
try:
run_dir.mkdir(parents=True, exist_ok=True)
coverage = self._coverage_document()
if coverage is not None:
try:
write_coverage(run_dir, coverage)
except OSError:
logger.exception("coverage.json write failed (non-fatal)")
if self.final_scan_result:
write_executive_report(run_dir, self.final_scan_result)
if self.vulnerability_reports:
write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids)
# SARIF 2.1.0 emitter for CI / ASPM integration. Always emit (even
# empty) so a clean run overwrites a prior findings.sarif rather than
# leaving a stale one — codeql-action's "absent from new submission →
# fixed" needs the fresh empty doc to auto-resolve alerts. Isolated
# in its own try: a SARIF-build error must NEVER break the CSV/MD/
# run-record path (the emitter's own contract).
try:
write_sarif(
run_dir,
self.vulnerability_reports,
tool_version=_strix_version(),
repository_context=self._sarif_repository_context(),
coverage=coverage,
)
except Exception:
logger.exception("SARIF emit failed (non-fatal; CSV/MD unaffected)")
write_run_record(run_dir, self.run_record)
logger.info("Essential scan data saved to: %s", run_dir)
self._write_artifacts()
except (OSError, RuntimeError):
logger.exception("Failed to save scan data")
def _write_artifacts(self) -> None:
"""Write scan artifacts under ``run_dir``, raising when the index or
run record cannot be written."""
run_dir = self.get_run_dir()
run_dir.mkdir(parents=True, exist_ok=True)
coverage = self._coverage_document()
if coverage is not None:
try:
write_coverage(run_dir, coverage)
except OSError:
logger.exception("coverage.json write failed (non-fatal)")
if self.final_scan_result:
write_executive_report(run_dir, self.final_scan_result)
# An index is written for an empty list too once a report was
# deleted, or the CSV/JSON on disk would still list it.
if self.vulnerability_reports or self._deleted_vulnerability_reports():
write_vulnerabilities(run_dir, self.vulnerability_reports, self._saved_vuln_ids)
# SARIF 2.1.0 emitter for CI / ASPM integration. Always emit (even
# empty) so a clean run overwrites a prior findings.sarif rather than
# leaving a stale one — codeql-action's "absent from new submission →
# fixed" needs the fresh empty doc to auto-resolve alerts. Isolated
# in its own try: a SARIF-build error must NEVER break the CSV/MD/
# run-record path (the emitter's own contract).
try:
write_sarif(
run_dir,
self.vulnerability_reports,
tool_version=_strix_version(),
repository_context=self._sarif_repository_context(),
coverage=coverage,
)
except Exception:
logger.exception("SARIF emit failed (non-fatal; CSV/MD unaffected)")
write_run_record(run_dir, self.run_record)
logger.info("Essential scan data saved to: %s", run_dir)
def _sarif_repository_context(self) -> dict[str, Any] | None:
"""Repo/commit/branch context for SARIF provenance (repo scans only).

View file

@ -696,6 +696,87 @@ def _do_update(
}
def _do_delete(
*,
report_id: str,
delete_reason: str,
agent_id: str | None = None,
agent_name: str | None = None,
) -> dict[str, Any]:
"""Withdraw a report an agent filed and has since disproved.
Deletion is for a finding that is not a vulnerability at all. A finding
that is real but weaker than filed is revised, not deleted.
"""
report_id = (report_id or "").strip()
delete_reason = (delete_reason or "").strip()
if not report_id or not delete_reason:
missing = "report_id" if not report_id else "delete_reason"
return {
"success": False,
"error": (
f"{missing} cannot be empty - name the report you are withdrawing and state "
"what disproved it"
),
}
from strix.report.state import ReportRollbackError, get_global_report_state
report_state = get_global_report_state()
if report_state is None:
return {
"success": False,
"error": "Report state unavailable - no reports have been filed yet",
}
try:
deleted = report_state.delete_vulnerability_report(
report_id,
delete_reason=delete_reason,
deleted_by_agent_id=agent_id,
deleted_by_agent_name=agent_name,
)
except Exception as e:
logger.exception("delete_vulnerability_report persistence failed")
if isinstance(e, ReportRollbackError):
outcome = (
f"{e.cause!s}. The report is still on file, but its on-disk indexes could "
"not be rewritten and may be stale until the next report is saved"
)
else:
outcome = f"{e!s}. The report is still on file"
return {
"success": False,
"error": f"Failed to delete report '{report_id}': {outcome}; retry the deletion.",
"report_id": report_id,
}
if deleted is None:
return {
"success": False,
"error": f"Report with id '{report_id}' not found",
"report_id": report_id,
}
logger.info(
"Vulnerability report %s deleted by %s: %s",
report_id,
agent_name or agent_id or "an agent",
delete_reason[:200],
)
return {
"success": True,
"action": "deleted",
"message": (
f"Report '{report_id}' ({deleted.get('title')}) is withdrawn and no longer "
"counts as a finding of this scan. Do not file it again unless new evidence "
"proves it."
),
"report_id": report_id,
"title": deleted.get("title"),
"severity": deleted.get("severity"),
}
async def _do_create(
*,
title: str,
@ -1548,6 +1629,48 @@ async def update_vulnerability_report(
return json.dumps(_with_warning(result, http_exchange_warning), ensure_ascii=False, default=str)
@function_tool(timeout=60)
async def delete_vulnerability_report(
ctx: RunContextWrapper,
report_id: str,
delete_reason: str,
) -> str:
"""Withdraw a vulnerability report that later testing disproved.
Use this when a finding you filed turns out not to be a vulnerability
at all: the exploit only worked because of a mistake in your own test
setup (a mixed-up session, a self-inflicted state change, a misread
response), the behaviour is documented and intended, or the control
you believed was missing is in fact enforced. The report is removed
from the scan; it does not stay behind as a zero-severity entry.
Do NOT use this for a finding that is real but weaker than filed —
revise it with ``update_vulnerability_report`` so the severity, the
impact and the counterevidence come down together. Do NOT rewrite a
report into a "retracted" or "false positive" note either: delete it.
Only withdraw a report you have re-tested yourself, whoever filed it.
Call ``get_report`` first to read what it claims, then state in
``delete_reason`` exactly what disproved it, so the scan history shows
who withdrew the finding and why. A withdrawn report cannot be restored; if new
evidence later proves the issue, file it again.
Args:
report_id: Id of the report to withdraw (format ``vuln-NNNN``).
delete_reason: What disproved the finding, in one or two
sentences.
"""
agent_id, agent_name = _caller_identity(ctx)
result = await asyncio.to_thread(
_do_delete,
report_id=report_id,
delete_reason=delete_reason,
agent_id=agent_id,
agent_name=agent_name,
)
return json.dumps(result, ensure_ascii=False, default=str)
_DEP_SEVERITY_FROM_CVSS = {
(9.0, 10.0): "critical",
(7.0, 9.0): "high",

View file

@ -13,17 +13,20 @@ from strix.report.dedupe import (
_prepare_report_for_comparison,
check_duplicate,
)
from strix.report.state import ReportState, set_global_report_state
from strix.report.state import ReportRollbackError, ReportState, set_global_report_state
from strix.report.writer import write_run_record
from strix.tools.finish.tool import finish_scan
from strix.tools.reporting import tool as reporting_tool
from strix.tools.reporting.tool import (
_do_create,
_do_create_dependency,
_do_delete,
_do_update,
_normalize_http_exchange_ids,
_verify_http_exchange_ids,
create_dependency_report,
create_vulnerability_report,
delete_vulnerability_report,
update_vulnerability_report,
)
@ -1976,3 +1979,270 @@ def test_update_refuses_code_locations_it_cannot_use(report_state: ReportState)
assert result["success"] is False
assert any("start_line" in error for error in result["errors"])
def _seed_saved_report(report_state: ReportState) -> Path:
"""The weak report, written to disk the way a filed finding is."""
_seed_weak_report(report_state)
report_state._saved_vuln_ids.clear()
report_state.save_run_data()
run_dir = report_state._run_dir
assert run_dir is not None
md_path = run_dir / "vulnerabilities" / "vuln-0009.md"
assert md_path.exists()
return run_dir
def test_delete_withdraws_a_disproved_report_and_its_artifacts(
report_state: ReportState,
) -> None:
"""A deleted finding leaves the run's state and every rendered artifact."""
run_dir = _seed_saved_report(report_state)
persisted: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = persisted.append
result = _do_delete(
report_id="vuln-0009",
delete_reason="The cross-tenant read came from the harness reusing the victim's session.",
agent_id="aaaa1111",
agent_name="Recon Agent",
)
assert result["success"] is True
assert result["action"] == "deleted"
assert result["report_id"] == "vuln-0009"
assert result["title"].startswith("Directus 11.5.1")
assert report_state.vulnerability_reports == []
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
assert json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8")) == []
assert "vuln-0009" not in (run_dir / "vulnerabilities.csv").read_text(encoding="utf-8")
assert len(persisted) == 1
assert persisted[0]["id"] == "vuln-0009"
assert persisted[0]["agent_id"] == "aaaa1111", "the callback sees the finding as filed"
deletion = persisted[0]["deletion"]
assert deletion["agent_id"] == "aaaa1111"
assert deletion["agent_name"] == "Recon Agent"
assert deletion["filed_by_agent_id"] == "aaaa1111"
assert deletion["reason"].startswith("The cross-tenant read")
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
history = run["deleted_vulnerability_reports"]
assert [entry["id"] for entry in history] == ["vuln-0009"]
assert history[0]["severity"] == "medium"
assert history[0]["deleted_at"]
def test_delete_never_reissues_the_withdrawn_id(report_state: ReportState) -> None:
"""The next finding must not inherit a deleted report's id, file or history."""
_seed_saved_report(report_state)
assert report_state.delete_vulnerability_report(
"vuln-0009", delete_reason="Disproved.", deleted_by_agent_id="aaaa1111"
)
new_id = report_state.add_vulnerability_report(title="A real finding", severity="high")
assert new_id == "vuln-0010"
assert [r["id"] for r in report_state.vulnerability_reports] == ["vuln-0010"]
def _assert_report_still_filed(
report_state: ReportState, run_dir: Path, original: dict[str, Any]
) -> None:
"""The report is in memory and in every artifact exactly as it was filed."""
assert report_state.vulnerability_reports == [original]
assert "vuln-0009" in report_state._saved_vuln_ids
assert (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
indexed = json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8"))
assert [r["id"] for r in indexed] == ["vuln-0009"]
assert "vuln-0009" in (run_dir / "vulnerabilities.csv").read_text(encoding="utf-8")
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
assert "deleted_vulnerability_reports" not in run
assert "deleted_vulnerability_reports" not in report_state.run_record
assert report_state._next_report_id() == "vuln-0010"
def test_any_agent_in_the_run_may_delete_a_report(report_state: ReportState) -> None:
"""Agents in a run share one trust boundary: a report another agent filed can be
withdrawn once disproved, and the history keeps both identities apart."""
run_dir = _seed_saved_report(report_state)
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
result = _do_delete(
report_id="vuln-0009",
delete_reason="Disproved.",
agent_id="834f79fb",
agent_name="Validation Agent",
)
assert result["success"] is True
assert report_state.vulnerability_reports == []
assert [c["id"] for c in calls] == ["vuln-0009"]
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
(deletion,) = report_state.run_record["deleted_vulnerability_reports"]
assert deletion["filed_by_agent_id"] == "aaaa1111"
assert deletion["agent_id"] == "834f79fb"
assert deletion["agent_name"] == "Validation Agent"
def test_delete_reports_persistence_failure_and_keeps_the_report(
report_state: ReportState,
) -> None:
"""Persistence refused after the local indexes were rewritten: they are rewritten back."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
def fail_persistence(_report: dict[str, Any]) -> None:
assert report_state.vulnerability_reports == [], "indexes are rewritten first"
raise RuntimeError("persistence failed")
report_state.vulnerability_deleted_callback = fail_persistence
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "persistence failed" in result["error"]
assert "still on file" in result["error"]
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_reports_an_artifact_write_failure_and_keeps_the_report(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A disk that cannot take the rewritten run record is a failed deletion, not a
success with stale artifacts; persistence is never told about it."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
attempts = 0
def write_run_record_once_failing(run_dir_: Path, record: dict[str, Any]) -> None:
nonlocal attempts
attempts += 1
if attempts == 1:
raise OSError(28, "No space left on device")
(run_dir_ / "run.json").write_text(json.dumps(record), encoding="utf-8")
monkeypatch.setattr("strix.report.state.write_run_record", write_run_record_once_failing)
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "No space left on device" in result["error"]
assert "still on file" in result["error"]
assert calls == []
assert attempts == 2, "the run record is written back after the failed rewrite"
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_reports_when_the_rollback_itself_cannot_be_written(
report_state: ReportState, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A disk that stays full: memory is restored, the caller is told the indexes
may be stale, and the next save rewrites them."""
run_dir = _seed_saved_report(report_state)
original = dict(report_state.vulnerability_reports[0])
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
disk_full = True
def write_run_record_while_disk_full(run_dir_: Path, record: dict[str, Any]) -> None:
if disk_full:
raise OSError(28, "No space left on device")
write_run_record(run_dir_, record)
monkeypatch.setattr("strix.report.state.write_run_record", write_run_record_while_disk_full)
with pytest.raises(ReportRollbackError) as excinfo:
report_state.delete_vulnerability_report(
"vuln-0009", delete_reason="Disproved.", deleted_by_agent_id="aaaa1111"
)
assert isinstance(excinfo.value.cause, OSError)
assert calls == []
assert report_state.vulnerability_reports == [original]
assert "vuln-0009" in report_state._saved_vuln_ids
assert "deleted_vulnerability_reports" not in report_state.run_record
result = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert result["success"] is False
assert "No space left on device" in result["error"]
assert "may be stale" in result["error"]
disk_full = False
report_state.save_run_data()
_assert_report_still_filed(report_state, run_dir, original)
def test_delete_can_be_retried_after_persistence_recovers(report_state: ReportState) -> None:
run_dir = _seed_saved_report(report_state)
outcomes = iter([RuntimeError("persistence failed"), None])
def flaky_persistence(_report: dict[str, Any]) -> None:
outcome = next(outcomes)
if outcome is not None:
raise outcome
report_state.vulnerability_deleted_callback = flaky_persistence
first = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
second = _do_delete(report_id="vuln-0009", delete_reason="Disproved.", agent_id="aaaa1111")
assert first["success"] is False
assert second["success"] is True
assert report_state.vulnerability_reports == []
assert not (run_dir / "vulnerabilities" / "vuln-0009.md").exists()
assert json.loads((run_dir / "vulnerabilities.json").read_text(encoding="utf-8")) == []
run = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
assert [e["id"] for e in run["deleted_vulnerability_reports"]] == ["vuln-0009"]
@pytest.mark.parametrize(
("report_id", "delete_reason", "expected"),
[
(" ", "reason", "report_id cannot be empty"),
("vuln-0009", " ", "delete_reason cannot be empty"),
("vuln-0404", "reason", "not found"),
],
)
def test_delete_rejects_a_call_it_cannot_act_on(
report_state: ReportState,
report_id: str,
delete_reason: str,
expected: str,
) -> None:
_seed_weak_report(report_state)
calls: list[dict[str, Any]] = []
report_state.vulnerability_deleted_callback = calls.append
result = _do_delete(report_id=report_id, delete_reason=delete_reason, agent_id="aaaa1111")
assert result["success"] is False
assert expected in result["error"]
assert len(report_state.vulnerability_reports) == 1
assert calls == [], "nothing reaches persistence for a call that is refused"
async def test_delete_tool_carries_the_caller_identity(report_state: ReportState) -> None:
_seed_saved_report(report_state)
coordinator = type("Coordinator", (), {"names": {"aaaa1111": "Recon Agent"}})()
ctx = ToolContext(
context={"agent_id": "aaaa1111", "coordinator": coordinator},
tool_name="delete_vulnerability_report",
tool_call_id="call-1",
tool_arguments="{}",
)
raw = await delete_vulnerability_report.on_invoke_tool(
ctx,
json.dumps({"report_id": "vuln-0009", "delete_reason": "Disproved on re-test."}),
)
result = json.loads(raw)
assert result["success"] is True
history = report_state.run_record["deleted_vulnerability_reports"]
assert history[0]["agent_id"] == "aaaa1111"
assert history[0]["agent_name"] == "Recon Agent"
assert "update_vulnerability_report" in delete_vulnerability_report.description