fix(report): validate deduplication decisions

strix/report/dedupe.py

@@ -300,7 +300,15 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
         raise ValueError(f"No JSON object found in dedupe response: {content[:500]}")
     parsed = json.loads(text[start : end + 1])
 
-    duplicate_id = str(parsed.get("duplicate_id") or "")[:64]
+    is_duplicate = parsed.get("is_duplicate")
+    if not isinstance(is_duplicate, bool):
+        raise TypeError("is_duplicate must be a boolean")
+    duplicate_id = parsed.get("duplicate_id") or ""
+    if is_duplicate and (not isinstance(duplicate_id, str) or not duplicate_id):
+        raise ValueError("A duplicate decision must name an existing report ID")
+    # Keep the ID verbatim: truncating or stripping it could turn an unknown ID
+    # into a different, existing one before the membership check below.
+    duplicate_id = str(duplicate_id)
     reason = str(parsed.get("reason") or "")[:500]
     try:
         confidence = float(parsed.get("confidence", 0.0))
@@ -308,13 +316,22 @@ def _parse_dedupe_response(content: str) -> dict[str, Any]:
         confidence = 0.0
 
     return {
-        "is_duplicate": bool(parsed.get("is_duplicate", False)),
+        "is_duplicate": is_duplicate,
         "duplicate_id": duplicate_id,
         "confidence": confidence,
         "reason": reason,
     }
 
 
+def _validate_duplicate_reference(
+    result: dict[str, Any], existing_reports: list[dict[str, Any]]
+) -> None:
+    if result["is_duplicate"] and not any(
+        report.get("id") == result["duplicate_id"] for report in existing_reports
+    ):
+        raise ValueError("Duplicate decision refers to an unknown report ID")
+
+
 def _extract_text(response: ModelResponse) -> str:
     parts: list[str] = []
     for item in response.output:
@@ -397,6 +414,7 @@ async def check_duplicate(
             }
 
         result = _parse_dedupe_response(content)
+        _validate_duplicate_reference(result, existing_reports)
 
         logger.info(
             "Deduplication check: is_duplicate=%s, confidence=%.2f, reason=%s",

tests/test_dedupe_decisions.py

@@ -0,0 +1,159 @@
+"""Only validated model decisions may suppress a finding."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock
+
+import pytest
+from agents.items import ModelResponse
+from agents.usage import Usage
+from openai.types.responses import ResponseOutputMessage, ResponseOutputText
+
+from strix.config import loader
+from strix.report import dedupe
+from strix.report.state import ReportState, get_global_report_state, set_global_report_state
+from strix.tools.reporting.tool import _do_create
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture
+def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[ReportState]:
+    monkeypatch.chdir(tmp_path)
+    previous = get_global_report_state()
+    state = ReportState(run_name="dedupe-decisions")
+    set_global_report_state(state)
+    state.add_vulnerability_report(title="Existing finding", severity="low")
+    try:
+        yield state
+    finally:
+        set_global_report_state(previous)
+
+
+def _mock_judge(monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]) -> AsyncMock:
+    settings = loader.load_settings().model_copy(deep=True)
+    settings.llm.model = "openai/synthetic-model"
+    model = AsyncMock()
+    model.get_response.return_value = ModelResponse(
+        output=[
+            ResponseOutputMessage(
+                id="msg_synthetic",
+                type="message",
+                role="assistant",
+                status="completed",
+                content=[
+                    ResponseOutputText(
+                        type="output_text",
+                        annotations=[],
+                        text=json.dumps({"confidence": 0.95, "reason": "Synthetic", **decision}),
+                    )
+                ],
+            )
+        ],
+        usage=Usage(),
+        response_id="synthetic",
+    )
+    monkeypatch.setattr(dedupe, "load_settings", lambda: settings)
+    monkeypatch.setattr(dedupe, "configure_sdk_model_defaults", lambda _: None)
+    monkeypatch.setattr(dedupe, "resolve_dedupe_model", lambda *_: model)
+    return model
+
+
+async def _file_finding() -> dict[str, Any]:
+    return await _do_create(
+        title="New finding",
+        description="The endpoint reflects unencoded input.",
+        impact="The reflected script executes in a victim's browser.",
+        target="https://example.invalid",
+        technical_analysis="Input is interpolated into HTML without encoding.",
+        poc_description="Open the endpoint with the reflected parameter.",
+        poc_script_code="print('synthetic reproduction')",
+        remediation_steps="Encode output for its HTML context.",
+        evidence="The synthetic response contains the unencoded parameter.",
+        assumptions="A victim follows the crafted link.",
+        counterevidence="No encoding or content security policy is present.",
+        confidence="high",
+        severity_change_conditions="A strict content security policy would lower the impact.",
+        fix_effort="low",
+        cvss_breakdown={
+            "attack_vector": "N",
+            "attack_complexity": "L",
+            "privileges_required": "N",
+            "user_interaction": "R",
+            "scope": "U",
+            "confidentiality": "L",
+            "integrity": "L",
+            "availability": "N",
+        },
+        endpoint="/search",
+        method="GET",
+        cve=None,
+        cwe="CWE-79",
+        code_locations=None,
+    )
+
+
+@pytest.mark.parametrize("value", ["false", "true", 0, 1, None, [], {}])
+def test_dedupe_parser_requires_a_boolean(value: Any) -> None:
+    with pytest.raises(TypeError, match="is_duplicate"):
+        dedupe._parse_dedupe_response(json.dumps({"is_duplicate": value}))
+
+
+@pytest.mark.parametrize(
+    "decision",
+    [
+        {"is_duplicate": "false", "duplicate_id": "vuln-0001"},
+        {"is_duplicate": "true", "duplicate_id": "vuln-0001"},
+        {"is_duplicate": 1, "duplicate_id": "vuln-0001"},
+        {"is_duplicate": True, "duplicate_id": "vuln-9999"},
+        {"is_duplicate": True, "duplicate_id": ""},
+        {"is_duplicate": True, "duplicate_id": None},
+        {"is_duplicate": True, "duplicate_id": ["vuln-0001"]},
+        {"is_duplicate": True},
+    ],
+)
+async def test_invalid_dedupe_decision_does_not_suppress_finding(
+    report_state: ReportState, monkeypatch: pytest.MonkeyPatch, decision: dict[str, Any]
+) -> None:
+    model = _mock_judge(monkeypatch, decision)
+
+    result = await _file_finding()
+
+    assert result["success"] is True
+    model.get_response.assert_awaited_once()
+    saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
+    assert [report["title"] for report in saved] == ["Existing finding", "New finding"]
+    assert (report_state.get_run_dir() / "vulnerabilities" / "vuln-0002.md").is_file()
+
+
+@pytest.mark.parametrize("is_duplicate", [False, True])
+async def test_valid_dedupe_decisions_keep_their_meaning(
+    report_state: ReportState, monkeypatch: pytest.MonkeyPatch, is_duplicate: bool
+) -> None:
+    # An extraneous ID must not turn a valid false decision into a duplicate.
+    _mock_judge(monkeypatch, {"is_duplicate": is_duplicate, "duplicate_id": "vuln-0001"})
+
+    result = await _file_finding()
+
+    assert result["success"] is not is_duplicate
+    saved = json.loads((report_state.get_run_dir() / "vulnerabilities.json").read_text())
+    assert len(saved) == (1 if is_duplicate else 2)
+    if is_duplicate:
+        assert result["duplicate_of"] == "vuln-0001"
+
+
+async def test_dedupe_id_is_matched_without_truncation(monkeypatch: pytest.MonkeyPatch) -> None:
+    existing_id = "x" * 64
+    _mock_judge(monkeypatch, {"is_duplicate": True, "duplicate_id": existing_id + "-unknown"})
+
+    result = await dedupe.check_duplicate(
+        {"title": "New finding"}, [{"id": existing_id, "title": "Existing finding"}]
+    )
+
+    assert result["is_duplicate"] is False
+    assert "error" in result