feat(gate): add tamper-evident finding baseline ratchet

README.md

@@ -924,6 +924,12 @@ structorium config set NEO4J_PASSWORD <password>
 
 ## 🚧 New-Code Gate
 
+For legacy repositories that need a finding-level ratchet in addition to changed-line
+gating, capture the current active debt with `structorium baseline capture`, commit
+the artifact, and run `structorium baseline check` in CI. The checksum-protected
+baseline fails only on newly introduced findings and reports debt removed since the
+capture. See the [baseline ratchet guide](docs/BASELINES.md).
+
 <p align="center">
   <img src="assets/readme/new_code_gate.png" alt="New-Code Gate" width="100%"/>
 </p>
@@ -2172,4 +2178,3 @@ copies or substantial portions of the Software.
 <p align="center">
   <a href="#structorium">↑ Back to top</a>
 </p>
-

app/cli_support/parser.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import argparse
-from importlib.metadata import PackageNotFoundError, version as get_version
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as get_version
 
 from app.cli_support.parser_groups import (
+    _add_baseline_parser,
     _add_config_parser,
     _add_detect_parser,
     _add_dev_parser,
@@ -117,6 +119,7 @@ def create_parser(*, langs: list[str], detector_names: list[str]) -> argparse.Ar
         parser_class=_NoAbbrevArgumentParser,
     )
     _add_scan_parser(sub)
+    _add_baseline_parser(sub)
     _add_status_parser(sub)
     _add_tree_parser(sub)
     _add_show_parser(sub)

app/cli_support/parser_groups.py

@@ -19,6 +19,7 @@
 )
 
 __all__ = [
+    "_add_baseline_parser",
     "_add_config_parser",
     "_add_detect_parser",
     "_add_dev_parser",
@@ -40,6 +41,22 @@
 ]
 
 
+def _add_baseline_parser(sub) -> None:
+    p_baseline = sub.add_parser(
+        "baseline", help="Capture existing findings and fail only on regressions"
+    )
+    actions = p_baseline.add_subparsers(dest="baseline_action", required=True)
+    capture = actions.add_parser("capture", help="Capture active findings as a baseline")
+    capture.add_argument("--state", type=str, default=None, help="Path to state file")
+    capture.add_argument("--output", default=".structorium/baseline.json", help="Baseline output path")
+    capture.add_argument("--force", action="store_true", help="Explicitly replace an existing baseline")
+    check = actions.add_parser("check", help="Compare active findings with a baseline")
+    check.add_argument("--state", type=str, default=None, help="Path to state file")
+    check.add_argument("--baseline", default=".structorium/baseline.json", help="Baseline input path")
+    check.add_argument("--max-new", type=int, default=0, help="Allowed new findings before failure")
+    check.add_argument("--json", action="store_true", help="Emit machine-readable comparison JSON")
+
+
 def _add_scan_parser(sub) -> None:
     p_scan = sub.add_parser(
         "scan",

app/commands/baseline_cmd.py

@@ -0,0 +1,83 @@
+"""Capture and enforce version-controlled Structorium finding baselines."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from app.commands.helpers.runtime import command_runtime
+from core.output_api import colorize
+from engine._state.baseline import (
+    build_baseline,
+    compare_baseline,
+    load_baseline,
+    write_baseline,
+)
+
+
+def _require_scan(state: dict) -> None:
+    if not state.get("last_scan"):
+        print(colorize("No completed scan found. Run `structorium scan` first.", "red"), file=sys.stderr)
+        raise SystemExit(2)
+
+
+def _capture(args: argparse.Namespace, state: dict) -> None:
+    document = build_baseline(state)
+    output = write_baseline(
+        document,
+        Path(args.output),
+        overwrite=bool(getattr(args, "force", False)),
+    )
+    print(colorize(f"Captured {document['finding_count']} findings in {output}", "green"))
+
+
+def _check(args: argparse.Namespace, state: dict) -> None:
+    document = load_baseline(Path(args.baseline))
+    diff = compare_baseline(state, document)
+    payload = {
+        "baseline": str(Path(args.baseline)),
+        "new_count": len(diff["new"]),
+        "resolved_count": len(diff["resolved"]),
+        "unchanged_count": len(diff["unchanged"]),
+        **diff,
+    }
+    if getattr(args, "json", False):
+        print(json.dumps(payload, indent=2, sort_keys=True))
+    else:
+        print(colorize("\nStructorium baseline ratchet", "bold"))
+        print(f"  New: {payload['new_count']}")
+        print(f"  Resolved since capture: {payload['resolved_count']}")
+        print(f"  Unchanged: {payload['unchanged_count']}")
+        for finding in diff["new"]:
+            print(colorize(f"  + T{finding['tier']} {finding['id']}", "red"))
+
+    max_new = int(getattr(args, "max_new", 0))
+    if payload["new_count"] > max_new:
+        print(
+            colorize(
+                f"Baseline failed: {payload['new_count']} new findings exceeds --max-new {max_new}",
+                "red",
+            ),
+            file=sys.stderr,
+        )
+        raise SystemExit(3)
+    print(colorize("Baseline passed", "green"), file=sys.stderr)
+
+
+def cmd_baseline(args: argparse.Namespace) -> None:
+    """Dispatch baseline capture/check operations."""
+    state = command_runtime(args).state
+    _require_scan(state)
+    action = getattr(args, "baseline_action", None)
+    if action == "capture":
+        _capture(args, state)
+        return
+    if action == "check":
+        _check(args, state)
+        return
+    raise ValueError("baseline action must be capture or check")
+
+
+__all__ = ["cmd_baseline"]

app/commands/registry.py

@@ -12,6 +12,7 @@
 
 def _build_handlers() -> dict[str, CommandHandler]:
     """Import all command modules and build the handler dict on first access."""
+    from app.commands.baseline_cmd import cmd_baseline
     from app.commands.config_cmd import cmd_config
     from app.commands.detect import cmd_detect
     from app.commands.dev_cmd import cmd_dev
@@ -31,6 +32,7 @@ def _build_handlers() -> dict[str, CommandHandler]:
     from app.commands.zone_cmd import cmd_zone
 
     return {
+        "baseline": cmd_baseline,
         "scan": cmd_scan,
         "status": cmd_status,
         "show": cmd_show,

docs/BASELINES.md

@@ -0,0 +1,39 @@
+# Freeze existing debt, ratchet every change
+
+Large existing repositories often cannot fix every finding before enabling a CI
+gate. Structorium's baseline workflow records the current active finding identities
+and then fails only when a later scan introduces more than the allowed number of new
+findings.
+
+```bash
+structorium scan --path . --profile ci
+structorium baseline capture --output .structorium/baseline.json
+git add .structorium/baseline.json
+
+# In later CI runs
+structorium scan --path . --profile ci
+structorium baseline check --baseline .structorium/baseline.json
+```
+
+`check` exits with code `3` when new findings exceed `--max-new` (zero by
+default). It also reports baseline findings that were resolved, making the artifact
+a one-way ratchet rather than a permanent exemption list.
+
+The baseline is deterministic, sorted, and protected by a SHA-256 checksum. Editing
+individual entries by hand fails closed; recapturing requires the explicit
+`capture --force` operation. Resolved and suppressed findings are never captured.
+
+## Research provenance
+
+The design is an original Structorium implementation informed by two public
+patterns:
+
+- [ArchUnit FreezingArchRule](https://www.archunit.org/userguide/html/000_Index.html#_freezing_arch_rules)
+  records existing violations and reports only new ones, allowing incremental
+  adoption in grown projects.
+- [Qodana baselines](https://www.jetbrains.com/help/qodana/quality-gate.html#baseline)
+  separate accepted existing problems from newly introduced analysis results.
+
+Unlike those systems, this artifact is built directly from Structorium's stable
+finding IDs, includes a tamper-evident checksum, and exposes resolved baseline debt
+in the same comparison.

engine/_state/baseline.py

@@ -0,0 +1,154 @@
+"""Version-controlled finding baselines and fail-closed regression comparison."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, TypedDict
+
+from core.discovery_api import safe_write_text
+
+BASELINE_SCHEMA_VERSION = 1
+
+
+class BaselineDiff(TypedDict):
+    """Deterministic comparison between active findings and a captured baseline."""
+
+    new: list[dict[str, Any]]
+    resolved: list[dict[str, Any]]
+    unchanged: list[dict[str, Any]]
+
+
+def _canonical_json(value: object) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _fingerprint(finding: Mapping[str, Any]) -> str:
+    identity = "\0".join(
+        (
+            str(finding.get("detector") or "unknown"),
+            str(finding.get("file") or ".").replace("\\", "/"),
+            str(finding.get("id") or ""),
+        )
+    )
+    return hashlib.sha256(identity.encode("utf-8", errors="replace")).hexdigest()
+
+
+def _active_entries(state: Mapping[str, Any]) -> list[dict[str, Any]]:
+    findings = state.get("findings")
+    if not isinstance(findings, Mapping):
+        return []
+    entries: list[dict[str, Any]] = []
+    for value in findings.values():
+        if not isinstance(value, Mapping):
+            continue
+        if value.get("status") != "open" or value.get("suppressed"):
+            continue
+        entries.append(
+            {
+                "fingerprint": _fingerprint(value),
+                "id": str(value.get("id") or ""),
+                "detector": str(value.get("detector") or "unknown"),
+                "file": str(value.get("file") or ".").replace("\\", "/"),
+                "tier": int(value.get("tier") or 4),
+                "confidence": str(value.get("confidence") or "unknown"),
+            }
+        )
+    return sorted(entries, key=lambda item: item["fingerprint"])
+
+
+def _checksum(entries: list[dict[str, Any]]) -> str:
+    return hashlib.sha256(_canonical_json(entries).encode("utf-8")).hexdigest()
+
+
+def build_baseline(state: Mapping[str, Any]) -> dict[str, Any]:
+    """Capture active, unsuppressed findings in a deterministic baseline document."""
+    entries = _active_entries(state)
+    return {
+        "schema_version": BASELINE_SCHEMA_VERSION,
+        "source": {
+            "last_scan": state.get("last_scan"),
+            "scan_count": int(state.get("scan_count") or 0),
+        },
+        "finding_count": len(entries),
+        "findings": entries,
+        "checksum": f"sha256:{_checksum(entries)}",
+    }
+
+
+def validate_baseline(document: Mapping[str, Any]) -> None:
+    """Reject malformed or silently edited baseline documents."""
+    if document.get("schema_version") != BASELINE_SCHEMA_VERSION:
+        raise ValueError(
+            f"unsupported baseline schema: {document.get('schema_version')!r}"
+        )
+    findings = document.get("findings")
+    if not isinstance(findings, list) or any(not isinstance(item, dict) for item in findings):
+        raise ValueError("baseline findings must be a list of objects")
+    fingerprints = [str(item.get("fingerprint") or "") for item in findings]
+    if not all(fingerprints) or len(fingerprints) != len(set(fingerprints)):
+        raise ValueError("baseline fingerprints must be non-empty and unique")
+    if fingerprints != sorted(fingerprints):
+        raise ValueError("baseline findings must be sorted by fingerprint")
+    expected = f"sha256:{_checksum(findings)}"
+    if document.get("checksum") != expected:
+        raise ValueError("baseline checksum mismatch; recapture it explicitly")
+    if document.get("finding_count") != len(findings):
+        raise ValueError("baseline finding_count does not match findings")
+
+
+def compare_baseline(
+    state: Mapping[str, Any], document: Mapping[str, Any]
+) -> BaselineDiff:
+    """Return new, resolved, and unchanged findings relative to a valid baseline."""
+    validate_baseline(document)
+    current = {entry["fingerprint"]: entry for entry in _active_entries(state)}
+    captured = {
+        str(entry["fingerprint"]): entry
+        for entry in document["findings"]
+        if isinstance(entry, dict)
+    }
+    return {
+        "new": [current[key] for key in sorted(current.keys() - captured.keys())],
+        "resolved": [captured[key] for key in sorted(captured.keys() - current.keys())],
+        "unchanged": [current[key] for key in sorted(current.keys() & captured.keys())],
+    }
+
+
+def load_baseline(path: Path) -> dict[str, Any]:
+    """Load and validate a baseline JSON document."""
+    try:
+        document = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise ValueError(f"baseline not found: {path}") from exc
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError(f"could not read baseline {path}: {exc}") from exc
+    if not isinstance(document, dict):
+        raise ValueError("baseline root must be a JSON object")
+    validate_baseline(document)
+    return document
+
+
+def write_baseline(
+    document: Mapping[str, Any], path: Path, *, overwrite: bool = False
+) -> Path:
+    """Write a validated baseline atomically, refusing implicit replacement."""
+    validate_baseline(document)
+    if path.exists() and not overwrite:
+        raise ValueError(f"baseline already exists: {path}; pass --force to replace it")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    safe_write_text(path, json.dumps(document, indent=2, sort_keys=True) + "\n")
+    return path.resolve()
+
+
+__all__ = [
+    "BASELINE_SCHEMA_VERSION",
+    "BaselineDiff",
+    "build_baseline",
+    "compare_baseline",
+    "load_baseline",
+    "validate_baseline",
+    "write_baseline",
+]