feat(corpus-seeded-mechanisms): reports name the mechanism, its known fix guard and pair provenance

Task 5.1 (Req 6.1, 6.3).

- reports.py: the markdown and SARIF writers take the mechanisms a run
  cites; a finding cites one through its mechanism:<id> tag (variant) or
  its merged overlay record's mechanism_id; markdown shows Mechanism,
  Known fix guard and Learned from pairs per finding plus a Mechanisms
  section, SARIF properties carry mechanism_id/summary/guard/pairs;
  variant findings keep suspicion, merged inventory findings keep their
  own evidence level; reports without mechanisms are unchanged.
- cli: _search_variants returns the cited mechanisms from the store and
  the scan passes them to both writers.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

src/openultrasast/cli.py

@@ -405,6 +405,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     wrote_overlay = False
     variants_payload: dict[str, object] | None = None
     variant_findings: list[StaticFinding] = []
+    mechanisms_cited: dict[str, dict[str, object]] = {}
     if Stage.MAP in plan.requested:
         built_map = runtime.run_stage(
             "map",
@@ -438,7 +439,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
                 findings = findings + extra
                 write_findings(findings, findings_path)
         if config.variants.enabled:
-            variant_findings, overlay_records, variants_payload = _search_variants(
+            variant_findings, overlay_records, variants_payload, mechanisms_cited = _search_variants(
                 root=run.target,
                 targets=targets,
                 overlay_records=overlay_records,
@@ -545,6 +546,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             complexity_map=built_map,
             verdicts=verdict_records if wrote_verdicts else None,
             overlay=overlay_records if wrote_overlay else None,
+            mechanisms=mechanisms_cited or None,
         ),
     )
     runtime.run_stage(
@@ -554,6 +556,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
             verifications,
             sarif_path,
             overlay=overlay_records if wrote_overlay else None,
+            mechanisms=mechanisms_cited or None,
         ),
     )
     runtime.run_stage(
@@ -865,19 +868,23 @@ def _search_variants(
     facts: SemanticFacts | FactLoadError,
     max_mechanisms: int,
     runtime: HarnessRuntime,
-) -> tuple[list[StaticFinding], list[OverlayRecord], dict[str, object]]:
-    """Third MAP proposer (corpus-seeded-mechanisms Req 3): shapes from the mechanism store, merged with overlay flows."""
+) -> tuple[list[StaticFinding], list[OverlayRecord], dict[str, object], dict[str, dict[str, object]]]:
+    """Third MAP proposer (corpus-seeded-mechanisms Req 3): shapes from the mechanism store, merged with overlay flows.
+
+    Also returns the mechanisms the findings cite (summary, guard, pairs, cwe) for the reports (Req 6.1).
+    """
     from .semantic.mechanisms import corpus_mechanisms
     from .semantic.variant_search import hits_to_findings, search_tree
 
     store = MechanismStore(root / CALIBRATION_DIR / "mechanisms.jsonl")
     empty: dict[str, object] = {"mechanisms_searched": 0, "files_searched": 0, "findings": 0, "merged_into_overlay": 0}
     if isinstance(facts, FactLoadError):
-        return [], overlay_records, empty
+        return [], overlay_records, empty, {}
     result = runtime.run_stage("variants", lambda: search_tree(root, targets, store, facts, max_mechanisms=max_mechanisms))
     for degradation in result.degradations:
         runtime.state["degradations"].append(degradation)
-    summaries = {record.id: (record.summary, record.pairs, record.cwe) for record in corpus_mechanisms(store.load())}
+    records_by_id = {record.id: record for record in corpus_mechanisms(store.load())}
+    summaries = {record.id: (record.summary, record.pairs, record.cwe) for record in records_by_id.values()}
     findings, records = hits_to_findings(result.hits, overlay_records, summaries)
     merged = sum(1 for before, after in zip(overlay_records, records, strict=True) if before.mechanism_id is None and after.mechanism_id)
     payload: dict[str, object] = {
@@ -886,7 +893,15 @@ def _search_variants(
         "findings": len(findings),
         "merged_into_overlay": merged,
     }
-    return findings, records, payload
+    cited_ids = {tag.split(":", 1)[1] for finding in findings for tag in finding.tags if tag.startswith("mechanism:")} | {
+        str(record.mechanism_id) for record in records if record.mechanism_id
+    }
+    cited: dict[str, dict[str, object]] = {
+        mechanism_id: {"summary": record.summary, "guard": record.guard, "pairs": list(record.pairs), "cwe": record.cwe}
+        for mechanism_id, record in records_by_id.items()
+        if mechanism_id in cited_ids
+    }
+    return findings, records, payload, cited
 
 
 def _hotspot_from_variant(finding: StaticFinding) -> Hotspot:

src/openultrasast/reports.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import asdict
 from pathlib import Path
 
@@ -19,10 +19,12 @@ def write_markdown_report(
     complexity_map: object | None = None,
     verdicts: Sequence[object] | None = None,
     overlay: Sequence[object] | None = None,
+    mechanisms: Mapping[str, Mapping[str, object]] | None = None,
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications or [])
     overlay_by_id = _overlay_by_id(overlay or [])
+    cited: dict[str, Mapping[str, object]] = {}
     lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""]
     if not findings:
         lines.append("No quick-mode findings were emitted.")
@@ -51,7 +53,16 @@ def write_markdown_report(
         if disposition is not None:
             lines.append(f"- Disposition: `{getattr(disposition, 'disposition', '')}`")
             lines.append(f"- Overlay reason: `{getattr(disposition, 'reason', '')}`")
+        mechanism_id = _mechanism_id_for(finding, disposition)
+        if mechanism_id and mechanisms is not None and mechanism_id in mechanisms:
+            info = mechanisms[mechanism_id]
+            cited[mechanism_id] = info
+            lines.append(f"- Mechanism: `{mechanism_id}` — {info.get('summary', '')}")
+            lines.append(f"- Known fix guard: `{info.get('guard', 'none')}`")
+            lines.append(f"- Learned from pairs: `{', '.join(_pairs_of(info))}`")
         lines.extend(["", finding.rationale, ""])
+    if cited:
+        _append_mechanisms(lines, cited)
     if overlay:
         _append_overlay(lines, overlay)
     if complexity_map is not None:
@@ -133,6 +144,7 @@ def write_sarif_report(
     verifications: list[VerificationResult],
     path: Path,
     overlay: Sequence[object] | None = None,
+    mechanisms: Mapping[str, Mapping[str, object]] | None = None,
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications)
@@ -151,7 +163,9 @@ def write_sarif_report(
                     }
                 },
                 "results": [
-                    _sarif_result(finding, verification_by_id.get(finding.finding_id), overlay_by_id.get(finding.finding_id))
+                    _sarif_result(
+                        finding, verification_by_id.get(finding.finding_id), overlay_by_id.get(finding.finding_id), mechanisms=mechanisms
+                    )
                     for finding in findings
                 ],
             }
@@ -249,7 +263,13 @@ def _sarif_rules(findings: list[StaticFinding]) -> dict[str, dict[str, object]]:
     return rules
 
 
-def _sarif_result(finding: StaticFinding, verification: VerificationResult | None, overlay: object | None = None) -> dict[str, object]:
+def _sarif_result(
+    finding: StaticFinding,
+    verification: VerificationResult | None,
+    overlay: object | None = None,
+    *,
+    mechanisms: Mapping[str, Mapping[str, object]] | None = None,
+) -> dict[str, object]:
     properties: dict[str, object] = {
         "finding_id": finding.finding_id,
         "severity": finding.severity,
@@ -263,6 +283,14 @@ def _sarif_result(finding: StaticFinding, verification: VerificationResult | Non
     if overlay is not None:
         properties["disposition"] = getattr(overlay, "disposition", "")
         properties["overlay_reason"] = getattr(overlay, "reason", "")
+    mechanism_id = _mechanism_id_for(finding, overlay)
+    if mechanism_id:
+        properties["mechanism_id"] = mechanism_id
+        info = (mechanisms or {}).get(mechanism_id)
+        if info is not None:
+            properties["mechanism_summary"] = str(info.get("summary", ""))
+            properties["mechanism_guard"] = str(info.get("guard", "none"))
+            properties["mechanism_pairs"] = _pairs_of(info)
     return {
         "ruleId": _rule_id(finding),
         "level": _sarif_level(finding.severity),
@@ -354,3 +382,29 @@ def _relative_artifact(root: Path, path: Path) -> str:
         return path.relative_to(root).as_posix()
     except ValueError:
         return path.as_posix()
+
+
+def _mechanism_id_for(finding: StaticFinding, overlay: object | None) -> str | None:
+    """Mechanism cited by a finding: a `mechanism:<id>` tag on a variant finding, or the merged overlay record's id."""
+    for tag in finding.tags:
+        if tag.startswith("mechanism:"):
+            return tag.split(":", 1)[1]
+    merged = getattr(overlay, "mechanism_id", None)
+    return str(merged) if merged else None
+
+
+def _append_mechanisms(lines: list[str], cited: Mapping[str, Mapping[str, object]]) -> None:
+    lines.extend(
+        ["## Mechanisms", "", "Known mechanisms this report cites (corpus-seeded; variant findings stay suspicions until verified).", ""]
+    )
+    for mechanism_id, info in sorted(cited.items()):
+        pairs = ", ".join(_pairs_of(info)) or "unknown"
+        lines.append(
+            f"- `{mechanism_id}`: {info.get('summary', '')}; known fix guard `{info.get('guard', 'none')}`; learned from `{pairs}`"
+        )
+    lines.append("")
+
+
+def _pairs_of(info: Mapping[str, object]) -> list[str]:
+    pairs = info.get("pairs")
+    return [str(p) for p in pairs] if isinstance(pairs, list | tuple) else []

tests/test_mechanism_reports.py

@@ -0,0 +1,141 @@
+"""corpus-seeded-mechanisms task 5.1: reports name the mechanism, the known fix guard, and the pair provenance."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from openultrasast.findings import StaticFinding
+from openultrasast.reports import write_markdown_report, write_sarif_report
+from openultrasast.semantic.overlay import OverlayRecord
+
+
+def _variant() -> StaticFinding:
+    return StaticFinding(
+        finding_id="variant:corpus:abc:app.py:5",
+        path="app.py",
+        title="Variant of known mechanism: run_shell",
+        severity="medium",
+        confidence="low",
+        evidence_level="suspicion",
+        rationale="parameter into run_shell; fix adds allowlist_test. Structural variant of a mechanism learned from p1, p2.",
+        line=5,
+        function_name="handler",
+        reachability_status="unknown",
+        reachability_evidence=[],
+        reachability_conditions=[],
+        tags=["variant", "mechanism:corpus:abc"],
+        ranking_priority=0.0,
+    )
+
+
+MECHANISMS = {
+    "corpus:abc": {
+        "summary": "source_reaches_sink: parameter into run_shell (CWE-78); fix adds allowlist_test",
+        "guard": "allowlist_test",
+        "pairs": ["p1", "p2"],
+        "cwe": "CWE-78",
+    }
+}
+
+
+def test_markdown_names_mechanism_guard_and_provenance_and_keeps_the_suspicion_label(tmp_path: Path) -> None:
+    path = tmp_path / "report.md"
+    write_markdown_report([_variant()], path, mechanisms=MECHANISMS)
+    text = path.read_text()
+    assert "- Mechanism: `corpus:abc`" in text
+    assert "source_reaches_sink: parameter into run_shell" in text
+    assert "- Known fix guard: `allowlist_test`" in text
+    assert "- Learned from pairs: `p1, p2`" in text
+    assert "- Evidence level: `suspicion`" in text
+    assert "## Mechanisms" in text  # a section summarising every mechanism the report cites
+
+
+def test_sarif_carries_mechanism_properties_for_variant_and_merged_overlay_findings(tmp_path: Path) -> None:
+    merged = OverlayRecord(
+        proposal_id="python-os-command:app.py:9",
+        path="app.py",
+        line=9,
+        disposition="promote",
+        reason="request reaches os.system",
+        cwe="CWE-78",
+        sources=("request",),
+        sinks=("os.system",),
+        sanitizers=(),
+        evidence_level="static_corroboration",
+        origin="inventory",
+        engine="python-ast",
+        language="python",
+        mechanism_id="corpus:abc",
+    )
+    inventory = StaticFinding(
+        **{
+            **_variant().__dict__,
+            "finding_id": "python-os-command:app.py:9",
+            "line": 9,
+            "tags": ["injection"],
+            "evidence_level": "static_corroboration",
+        }
+    )
+    path = tmp_path / "report.sarif"
+    write_sarif_report([_variant(), inventory], [], path, overlay=[merged], mechanisms=MECHANISMS)
+    results = json.loads(path.read_text())["runs"][0]["results"]
+    by_id = {item["properties"]["finding_id"]: item["properties"] for item in results}
+    variant = by_id["variant:corpus:abc:app.py:5"]
+    assert (
+        variant["mechanism_id"] == "corpus:abc"
+        and variant["mechanism_guard"] == "allowlist_test"
+        and variant["mechanism_pairs"] == ["p1", "p2"]
+    )
+    assert variant["evidence_level"] == "suspicion"
+    merged_props = by_id["python-os-command:app.py:9"]
+    assert merged_props["mechanism_id"] == "corpus:abc" and merged_props["mechanism_guard"] == "allowlist_test"  # from the overlay record
+
+
+def test_reports_without_mechanisms_are_unchanged(tmp_path: Path) -> None:
+    plain = StaticFinding(**{**_variant().__dict__, "finding_id": "python-os-command:app.py:9", "tags": ["injection"]})
+    path = tmp_path / "report.md"
+    write_markdown_report([plain], path)
+    text = path.read_text()
+    assert "Mechanism:" not in text and "## Mechanisms" not in text
+
+
+def test_standard_scan_report_names_the_mechanism_behind_a_variant_finding(tmp_path: Path, monkeypatch) -> None:  # type: ignore[no-untyped-def]
+    from openultrasast.cli import main
+    from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair
+    from openultrasast.semantic.variants import Shape
+
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    (repo / "app.py").write_text("from shelltools import run_shell\n\n\ndef handler(q):\n    run_shell(q)\n")
+    shape = Shape(
+        language="python",
+        sink_name="run_shell",
+        arity=1,
+        source_positions=(0,),
+        source_kinds=("parameter",),
+        guard="allowlist_test",
+        mechanism="source_reaches_sink",
+    )
+    record = append_from_pair(
+        MechanismStore(repo / ".openultrasast" / "calibration" / "mechanisms.jsonl"),
+        shape,
+        summary="parameter into run_shell",
+        cwe="CWE-78",
+        pair="p1",
+        provenance="human",
+        tier="seeded",
+    )
+    monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
+    monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0")
+    assert main(["scan", str(repo), "--mode", "standard"]) == 0
+    run_dir = sorted((repo / ".runs").iterdir())[-1]
+    report = (run_dir / "report.md").read_text()
+    assert (
+        f"- Mechanism: `{record.id}`" in report
+        and "- Known fix guard: `allowlist_test`" in report
+        and "- Learned from pairs: `p1`" in report
+    )
+    sarif = json.loads((run_dir / "report.sarif").read_text())
+    props = [item["properties"] for item in sarif["runs"][0]["results"] if item["properties"]["finding_id"].startswith("variant:")]
+    assert props and props[0]["mechanism_guard"] == "allowlist_test" and props[0]["evidence_level"] == "suspicion"

tests/test_variant_scan.py

@@ -90,6 +90,12 @@ def test_variant_hit_merges_into_an_overlay_flow_at_the_same_call_site(tmp_path:
     assert not any(f["finding_id"].startswith("variant:") for f in findings)  # merged, not duplicated
     manifest = json.loads((run_dir / "manifest.json").read_text())
     assert manifest["variants"]["merged_into_overlay"] == 1 and manifest["variants"]["findings"] == 0
+    # Req 6.1 end to end: the merged inventory finding names the mechanism in both reports and keeps its own evidence level
+    report = (run_dir / "report.md").read_text()
+    assert f"- Mechanism: `{record.id}`" in report and "- Learned from pairs: `p`" in report
+    sarif = json.loads((run_dir / "report.sarif").read_text())
+    props = [item["properties"] for item in sarif["runs"][0]["results"] if item["properties"].get("mechanism_id") == record.id]
+    assert props and props[0]["mechanism_guard"] == "none" and props[0]["evidence_level"] != "suspicion"
 
 
 def test_quick_scan_never_searches_variants(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: