refactor(sarif): preserve durable finding identities

src/metis/cli/utils.py

@@ -32,7 +32,7 @@
 )
 
 from .exporters import export_csv, export_html, export_sarif
-from metis.sarif.utils import create_fingerprint
+from metis.sarif.triage import apply_triage_annotations
 from metis.vector_store.retrievers import retriever_query_config
 
 try:
@@ -377,7 +377,7 @@ def save_output(output_files, data, quiet=False, sarif_payload=None):
     else:
         files = list(output_files)
     json_payload = (
-        _merge_triage_annotations(data, sarif_payload)
+        apply_triage_annotations(data, sarif_payload)
         if sarif_payload is not None
         else data
     )
@@ -431,198 +431,6 @@ def _write_payload(path: Path, payload: object, label: str) -> None:
         _write_payload(output_path, json_payload, "Results")
 
 
-def _merge_triage_annotations(report_data, sarif_payload):
-    if not isinstance(report_data, dict):
-        return report_data
-    reviews = report_data.get("reviews")
-    if not isinstance(reviews, list):
-        return report_data
-    runs = sarif_payload.get("runs") if isinstance(sarif_payload, dict) else None
-    if not isinstance(runs, list):
-        return report_data
-
-    sarif_results = []
-    for run in runs:
-        if not isinstance(run, dict):
-            continue
-        results = run.get("results")
-        if not isinstance(results, list):
-            continue
-        sarif_results.extend(results)
-
-    issue_refs = []
-    for file_entry in reviews:
-        if not isinstance(file_entry, dict):
-            continue
-        file_name = str(file_entry.get("file") or file_entry.get("file_path") or "")
-        issues = file_entry.get("reviews")
-        if not isinstance(issues, list):
-            continue
-        for issue in issues:
-            if isinstance(issue, dict):
-                issue_refs.append((issue, file_name))
-
-    if not sarif_results or not issue_refs:
-        return report_data
-
-    indexed = []
-    fp_map: dict[str, list[int]] = {}
-    file_line_issue_map: dict[tuple[str, int, str], list[int]] = {}
-    file_line_map: dict[tuple[str, int], list[int]] = {}
-    file_issue_map: dict[tuple[str, str], list[int]] = {}
-    file_line_rule_map: dict[tuple[str, int, str], list[int]] = {}
-    file_rule_issue_map: dict[tuple[str, str, str], list[int]] = {}
-
-    for idx, result in enumerate(sarif_results):
-        if not isinstance(result, dict):
-            continue
-        properties = result.get("properties")
-        if not isinstance(properties, dict):
-            continue
-
-        file_name, line_number = _extract_sarif_location(result)
-        issue_text = _extract_sarif_issue_text(result)
-        rule_id = _extract_sarif_rule_id(result)
-        fingerprint = _extract_sarif_fingerprint(result)
-        indexed.append((idx, properties))
-        if fingerprint:
-            fp_map.setdefault(fingerprint, []).append(idx)
-        if file_name and line_number > 0 and rule_id:
-            file_line_rule_map.setdefault((file_name, line_number, rule_id), []).append(
-                idx
-            )
-        if file_name and line_number > 0 and issue_text:
-            file_line_issue_map.setdefault(
-                (file_name, line_number, issue_text), []
-            ).append(idx)
-        if file_name and rule_id and issue_text:
-            file_rule_issue_map.setdefault((file_name, rule_id, issue_text), []).append(
-                idx
-            )
-        if file_name and line_number > 0:
-            file_line_map.setdefault((file_name, line_number), []).append(idx)
-        if file_name and issue_text:
-            file_issue_map.setdefault((file_name, issue_text), []).append(idx)
-
-    unused = {idx for idx, _ in indexed}
-
-    def _take_from(mapping, key):
-        entries = mapping.get(key)
-        if not entries:
-            return None
-        while entries:
-            candidate = entries.pop(0)
-            if candidate in unused:
-                return candidate
-        return None
-
-    properties_by_idx = {idx: props for idx, props in indexed}
-
-    for issue, file_name in issue_refs:
-        line_number = _normalize_issue_line(issue.get("line_number"))
-        issue_text = str(issue.get("issue") or issue.get("title") or "").strip()
-        issue_rule = str(issue.get("rule_id") or issue.get("ruleId") or "").strip()
-        fingerprint = ""
-        if file_name and line_number > 0:
-            fingerprint = create_fingerprint(file_name, line_number, "AI001")
-
-        matchers = []
-        if fingerprint:
-            matchers.append((fp_map, fingerprint))
-        if file_name and line_number > 0 and issue_rule:
-            matchers.append((file_line_rule_map, (file_name, line_number, issue_rule)))
-        if file_name and line_number > 0 and issue_text:
-            matchers.append((file_line_issue_map, (file_name, line_number, issue_text)))
-        if file_name and issue_rule and issue_text:
-            matchers.append((file_rule_issue_map, (file_name, issue_rule, issue_text)))
-        if file_name and line_number > 0:
-            matchers.append((file_line_map, (file_name, line_number)))
-        if file_name and issue_text:
-            matchers.append((file_issue_map, (file_name, issue_text)))
-
-        match_idx = None
-        for mapping, key in matchers:
-            match_idx = _take_from(mapping, key)
-            if match_idx is not None:
-                break
-        if match_idx is None:
-            continue
-
-        unused.discard(match_idx)
-        properties = properties_by_idx.get(match_idx)
-        if not properties:
-            continue
-        _apply_triage_properties(issue, properties)
-
-    return report_data
-
-
-def _normalize_issue_line(raw_line) -> int:
-    try:
-        parsed = int(raw_line)
-    except Exception:
-        return 1
-    return parsed if parsed > 0 else 1
-
-
-def _extract_sarif_fingerprint(result: dict) -> str:
-    partial = result.get("partialFingerprints")
-    if not isinstance(partial, dict):
-        return ""
-    return str(partial.get("primaryLocationLineHash") or "").strip()
-
-
-def _extract_sarif_location(result: dict) -> tuple[str, int]:
-    locations = result.get("locations")
-    if not isinstance(locations, list) or not locations:
-        return "", 1
-    first = locations[0]
-    if not isinstance(first, dict):
-        return "", 1
-    physical = first.get("physicalLocation")
-    if not isinstance(physical, dict):
-        return "", 1
-    artifact = physical.get("artifactLocation")
-    file_name = ""
-    if isinstance(artifact, dict):
-        file_name = str(artifact.get("uri") or "")
-    region = physical.get("region")
-    properties = result.get("properties")
-    if isinstance(properties, dict):
-        reported_line = properties.get("reportedLineNumber")
-        if reported_line is not None:
-            return file_name, _normalize_issue_line(reported_line)
-    if not isinstance(region, dict):
-        return file_name, 1
-    return file_name, _normalize_issue_line(region.get("startLine"))
-
-
-def _extract_sarif_rule_id(result: dict) -> str:
-    return str(result.get("ruleId") or "").strip()
-
-
-def _extract_sarif_issue_text(result: dict) -> str:
-    message = result.get("message")
-    if isinstance(message, dict):
-        return str(message.get("text") or "").strip()
-    if isinstance(message, str):
-        return message.strip()
-    return ""
-
-
-def _apply_triage_properties(issue: dict, properties: dict) -> None:
-    if "metisTriaged" in properties:
-        issue["metisTriaged"] = bool(properties.get("metisTriaged"))
-    if "metisTriageStatus" in properties:
-        issue["metisTriageStatus"] = str(properties.get("metisTriageStatus") or "")
-    if "metisTriageReason" in properties:
-        issue["metisTriageReason"] = str(properties.get("metisTriageReason") or "")
-    if "metisTriageTimestamp" in properties:
-        issue["metisTriageTimestamp"] = str(
-            properties.get("metisTriageTimestamp") or ""
-        )
-
-
 def check_file_exists(file_path, quiet=False):
     if not Path(file_path).is_file():
         print_console(f"[red]File not found:[/red] {escape(file_path)}", quiet)

src/metis/engine/nodes/finding_dedup/core.py

@@ -83,7 +83,7 @@ def _unique_candidates(candidates: list[ReviewCandidate]) -> list[ReviewCandidat
     seen: set[tuple[str, str]] = set()
     for candidate in candidates:
         group = candidate.group.model_dump(mode="json", exclude={"reviews"})
-        finding = candidate.finding.model_dump(mode="json")
+        finding = candidate.finding.model_dump(mode="json", exclude={"id"})
         identity = (
             json.dumps(group, sort_keys=True, separators=(",", ":")),
             json.dumps(finding, sort_keys=True, separators=(",", ":")),

src/metis/engine/nodes/reachability/finding_adapter.py

@@ -37,6 +37,7 @@ def finding_to_review_item(
     primary_file = finding.primary_file or finding.sink_file or finding.source_file
     smap = SourceMap.for_file(codebase_path, primary_file) if primary_file else None
     item = {
+        "id": finding.id,
         "issue": issue,
         "line_number": line_number,
         "anchor": dict(finding.primary_anchor) if finding.primary_anchor else None,

src/metis/engine/stages/review/models.py

@@ -9,6 +9,7 @@
 from typing import Literal
 from typing import NotRequired
 from typing import Required
+from uuid import uuid4
 
 from pydantic import BaseModel
 from pydantic import ConfigDict
@@ -60,6 +61,7 @@ def validate_target(self) -> ReviewCommand:
 
 
 class ReviewFinding(BaseModel):
+    id: str = Field(default_factory=lambda: uuid4().hex, min_length=1)
     issue: str = Field(min_length=1)
 
     model_config = ConfigDict(extra="allow", frozen=True)

src/metis/sarif/triage.py

@@ -12,6 +12,7 @@
 from metis.json_io import write_json_atomic
 
 METIS_TRIAGED_KEY = "metisTriaged"
+METIS_FINDING_ID_KEY = "metisFindingId"
 METIS_TRIAGE_STATUS_KEY = "metisTriageStatus"
 METIS_TRIAGE_REASON_KEY = "metisTriageReason"
 METIS_TRIAGE_TIMESTAMP_KEY = "metisTriageTimestamp"
@@ -51,6 +52,57 @@ def save_sarif_file(path: str | Path, payload: dict[str, Any]) -> None:
     write_json_atomic(p, payload, indent=4)
 
 
+def apply_triage_annotations(report_data: Any, sarif_payload: Any) -> Any:
+    if not isinstance(report_data, dict) or not isinstance(sarif_payload, dict):
+        return report_data
+    reviews = report_data.get("reviews")
+    runs = sarif_payload.get("runs")
+    if not isinstance(reviews, list) or not isinstance(runs, list):
+        return report_data
+
+    properties_by_id: dict[str, dict[str, Any]] = {}
+    ambiguous_ids: set[str] = set()
+    for run in runs:
+        for result in run.get("results", ()) if isinstance(run, dict) else ():
+            properties = result.get("properties") if isinstance(result, dict) else None
+            if not isinstance(properties, dict):
+                continue
+            finding_id = str(properties.get(METIS_FINDING_ID_KEY) or "").strip()
+            if not finding_id:
+                continue
+            if finding_id in properties_by_id:
+                ambiguous_ids.add(finding_id)
+            else:
+                properties_by_id[finding_id] = properties
+    for finding_id in ambiguous_ids:
+        properties_by_id.pop(finding_id, None)
+
+    issues_by_id: dict[str, dict[str, Any]] = {}
+    ambiguous_report_ids: set[str] = set()
+    for review in reviews:
+        issues = review.get("reviews") if isinstance(review, dict) else None
+        if not isinstance(issues, list):
+            continue
+        for issue in issues:
+            if not isinstance(issue, dict):
+                continue
+            finding_id = str(issue.get("id") or "").strip()
+            if not finding_id:
+                continue
+            if finding_id in issues_by_id:
+                ambiguous_report_ids.add(finding_id)
+            else:
+                issues_by_id[finding_id] = issue
+    for finding_id in ambiguous_report_ids:
+        issues_by_id.pop(finding_id, None)
+
+    for finding_id, issue in issues_by_id.items():
+        properties = properties_by_id.get(finding_id)
+        if properties is not None:
+            _apply_triage_properties(issue, properties)
+    return report_data
+
+
 def extract_findings(
     payload: dict[str, Any], *, include_triaged: bool = False
 ) -> list[SarifFinding]:
@@ -241,3 +293,15 @@ def _apply_triage_metadata(
         properties[METIS_THREAT_MODEL_POLICY_KEY] = dict(threat_model_policy)
     else:
         properties.pop(METIS_THREAT_MODEL_POLICY_KEY, None)
+
+
+def _apply_triage_properties(issue: dict[str, Any], properties: dict[str, Any]) -> None:
+    if METIS_TRIAGED_KEY in properties:
+        issue[METIS_TRIAGED_KEY] = bool(properties.get(METIS_TRIAGED_KEY))
+    for key in (
+        METIS_TRIAGE_STATUS_KEY,
+        METIS_TRIAGE_REASON_KEY,
+        METIS_TRIAGE_TIMESTAMP_KEY,
+    ):
+        if key in properties:
+            issue[key] = str(properties.get(key) or "")

src/metis/sarif/writer.py

@@ -1,6 +1,9 @@
 # SPDX-FileCopyrightText: Copyright 2025 Arm Limited and/or its affiliates <open-source-office@arm.com>
 # SPDX-License-Identifier: Apache-2.0
 
+from uuid import uuid4
+
+from metis.sarif.triage import METIS_FINDING_ID_KEY
 from metis.version import __version__ as TOOL_VERSION
 from metis.sarif.utils import anchor_fingerprint, create_fingerprint, read_file_lines
 
@@ -107,6 +110,7 @@ def generate_sarif(
 
     run = sarif["runs"][0]
 
+    finding_ids: set[str] = set()
     for review in results.get("reviews", []):
         file_path = review.get("file_path")
         artifact_uri = review.get("file") or file_path or "<unknown>"
@@ -115,6 +119,11 @@ def generate_sarif(
         total_lines = len(lines) if source_available else 0
 
         for issue in review.get("reviews", []):
+            finding_id = str(issue.get("id") or "").strip()
+            if not finding_id or finding_id in finding_ids:
+                finding_id = uuid4().hex
+                issue["id"] = finding_id
+            finding_ids.add(finding_id)
             text = issue.get("issue", "unspecified")
             anchor = (
                 issue.get("anchor") if isinstance(issue.get("anchor"), dict) else None
@@ -167,7 +176,7 @@ def generate_sarif(
                 end = line_num + snippet_line_count - 1
                 context = snippet_text or "<context unavailable>"
 
-            properties = {}
+            properties = {METIS_FINDING_ID_KEY: finding_id}
             cwe_id = issue.get("cwe")
             if isinstance(cwe_id, str) and cwe_id.strip():
                 properties["cwe"] = cwe_id.strip()