Add --diff-only mode for gating on newly introduced findings

stellargate/cli.py

@@ -5,10 +5,11 @@
 import json
 import logging
 import sys
+from pathlib import Path
 
-from stellargate.aggregator import all_findings, run_all
+from stellargate.aggregator import ToolRunResult, all_findings, run_all
 from stellargate.config import VALID_SEVERITIES, Config, ConfigError
-from stellargate.report import gate_passed, to_json, to_markdown
+from stellargate.report import diff_findings, finding_key, gate_passed, to_json, to_markdown
 
 logger = logging.getLogger("stellargate")
 
@@ -58,6 +59,14 @@ def main(argv: list[str] | None = None) -> int:
     run_parser.add_argument("--json-report", default=None, help="Write JSON report to this path")
     run_parser.add_argument("--md-report", default=None, help="Write Markdown report to this path")
     run_parser.add_argument("--fail-on", default=None, help="Override fail_on threshold from config")
+    run_parser.add_argument(
+        "--diff-only",
+        default=None,
+        metavar="BASELINE.json",
+        help=("Gate only on findings newly introduced since a baseline StellarGate "
+              "JSON report. Loads the baseline, keeps only findings not present in it, "
+              "and bases both the gate and the report on those new findings."),
+    )
 
     args = parser.parse_args(argv)
 
@@ -95,30 +104,80 @@ def _run(args: argparse.Namespace) -> int:
 
     results = run_all(config)
     findings = all_findings(results)
-    passed = gate_passed(findings, fail_on)
-
-    for r in results:
-        if r.error:
-            logger.warning("Tool %s errored: %s", r.tool, r.error)
-        else:
-            logger.info("Tool %s produced %d finding(s)", r.tool, len(r.findings))
+    diff_mode = False
+
+    if args.diff_only:
+        baseline = _load_baseline(args.diff_only)
+        if baseline is None:
+            return 2
+        diff_mode = True
+        findings = diff_findings(findings, baseline)
+        keep = {finding_key(f) for f in findings}
+        results = [
+            ToolRunResult(
+                r.tool,
+                [f for f in r.findings if finding_key(f) in keep],
+                r.error,
+            )
+            for r in results
+        ]
 
-    logger.info("Gate result: %s (fail_on=%s)", "passed" if passed else "failed", fail_on)
+    passed = gate_passed(findings, fail_on)
 
-    # The Markdown report is the tool's primary output: it must go to stdout
-    # verbatim so CI/scripts can capture it. Diagnostics use logging (stderr).
-    print(to_markdown(results, fail_on, passed))
+    print(to_markdown(results, fail_on, passed, diff_mode))
 
     if args.json_report:
         with open(args.json_report, "w") as f:
-            json.dump(to_json(results, fail_on, passed), f, indent=2)
+            json.dump(to_json(results, fail_on, passed, diff_mode), f, indent=2)
 
     if args.md_report:
         with open(args.md_report, "w") as f:
-            f.write(to_markdown(results, fail_on, passed))
+            f.write(to_markdown(results, fail_on, passed, diff_mode))
 
     return 0 if passed else 1
 
 
+def _load_baseline(path: str) -> dict | None:
+    """Load and validate a baseline StellarGate JSON report.
+
+    Returns the parsed report on success, or None after printing a clear
+    error to stderr. None signals the caller to exit non-zero — a missing
+    or corrupt baseline is never silently treated as an empty baseline.
+    """
+    p = Path(path)
+    if not p.exists():
+        print(f"diff-only: baseline report not found: {path}", file=sys.stderr)
+        return None
+    try:
+        with open(p) as f:
+            report = json.load(f)
+    except json.JSONDecodeError as e:
+        print(f"diff-only: baseline {path} is not valid JSON: {e}", file=sys.stderr)
+        return None
+    except OSError as e:
+        print(f"diff-only: cannot read baseline {path}: {e}", file=sys.stderr)
+        return None
+
+    if not isinstance(report, dict) or "findings" not in report:
+        print(
+            f"diff-only: {path} is not a StellarGate report "
+            "(expected an object with a 'findings' list)",
+            file=sys.stderr,
+        )
+        return None
+    if not isinstance(report["findings"], list):
+        print(f"diff-only: baseline {path} has a non-list 'findings' field", file=sys.stderr)
+        return None
+    for i, item in enumerate(report["findings"]):
+        if not isinstance(item, dict) or "tool" not in item or "rule_id" not in item:
+            print(
+                f"diff-only: baseline {path} finding[{i}] is malformed "
+                "(each finding needs 'tool' and 'rule_id')",
+                file=sys.stderr,
+            )
+            return None
+    return report
+
+
 if __name__ == "__main__":
     sys.exit(main())

stellargate/report.py

@@ -2,12 +2,45 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
+from typing import Any
 
 from stellargate.aggregator import ToolRunResult
 from stellargate.schema import SEVERITY_ORDER, Finding
 
 
-def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> dict:
+def finding_key(finding: "Finding | dict[str, Any]") -> tuple:
+    """Uniqueness key for a finding: (tool, rule_id, location).
+
+    Accepts either a live Finding object or a dict as produced by
+    Finding.to_dict() (i.e. entries inside the JSON report's "findings"
+    list), so the same key works for diffing current results against a
+    previously serialized baseline report.
+    """
+    if isinstance(finding, Finding):
+        return (finding.tool, finding.rule_id, finding.location)
+    return (finding["tool"], finding["rule_id"], finding.get("location"))
+
+
+def diff_findings(
+    current: list[Finding], baseline_report: dict[str, Any]
+) -> list[Finding]:
+    """Return only findings newly introduced since a baseline report.
+
+    A finding is considered pre-existing if its (tool, rule_id, location)
+    key already appears in the baseline report's "findings" list. Only
+    brand-new findings are returned — this is what makes gating on
+    regressions possible without penalizing every historical finding.
+    """
+    baseline_keys = {finding_key(f) for f in baseline_report.get("findings", [])}
+    return [f for f in current if finding_key(f) not in baseline_keys]
+
+
+def to_json(
+    results: list[ToolRunResult],
+    fail_on: str,
+    gate_passed: bool,
+    diff_mode: bool = False,
+) -> dict:
     findings = [f for r in results for f in r.findings]
     counts = {sev: 0 for sev in SEVERITY_ORDER}
     for f in findings:
@@ -16,6 +49,7 @@ def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> di
     return {
         "generated_at": datetime.now(timezone.utc).isoformat(),
         "fail_on": fail_on,
+        "diff": bool(diff_mode),
         "passed": gate_passed,
         "summary": counts,
         "tools": [
@@ -30,12 +64,18 @@ def to_json(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> di
     }
 
 
-def to_markdown(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -> str:
+def to_markdown(
+    results: list[ToolRunResult],
+    fail_on: str,
+    gate_passed: bool,
+    diff_mode: bool = False,
+) -> str:
     findings = [f for r in results for f in r.findings]
     lines: list[str] = []
 
     status = "✅ PASSED" if gate_passed else "❌ FAILED"
-    lines.append(f"# StellarGate Compliance Report — {status}\n")
+    mode_label = " — diff mode (only new findings vs baseline)" if diff_mode else ""
+    lines.append(f"# StellarGate Compliance Report — {status}{mode_label}\n")
     lines.append(f"Threshold: fail on **{fail_on}** or above.\n")
 
     lines.append("## Per-tool summary\n")

tests/test_aggregator_and_report.py

@@ -1,12 +1,19 @@
+import json
 from unittest.mock import patch
 
 from hypothesis import given, settings
 from hypothesis import strategies as st
 
 from stellargate.aggregator import ToolRunResult, all_findings, run_all
 from stellargate.config import Config, ToolConfig
-from stellargate.report import gate_passed, to_json, to_markdown
-from stellargate.schema import SEVERITY_ORDER, Finding
+from stellargate.report import (
+    diff_findings,
+    finding_key,
+    gate_passed,
+    to_json,
+    to_markdown,
+)
+from stellargate.schema import Finding
 
 
 def make_results():