diff --git a/README.md b/README.md index 1ddd742..331011e 100644 --- a/README.md +++ b/README.md @@ -924,6 +924,12 @@ structorium config set NEO4J_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). +

New-Code Gate

@@ -2172,4 +2178,3 @@ copies or substantial portions of the Software.

↑ Back to top

- diff --git a/app/cli_support/parser.py b/app/cli_support/parser.py index 2f6b2ec..f9f3d59 100644 --- a/app/cli_support/parser.py +++ b/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) diff --git a/app/cli_support/parser_groups.py b/app/cli_support/parser_groups.py index d1705b5..b6580c5 100644 --- a/app/cli_support/parser_groups.py +++ b/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", diff --git a/app/commands/baseline_cmd.py b/app/commands/baseline_cmd.py new file mode 100644 index 0000000..1860a24 --- /dev/null +++ b/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"] diff --git a/app/commands/registry.py b/app/commands/registry.py index 2e4d8e0..be27fb7 100644 --- a/app/commands/registry.py +++ b/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, diff --git a/docs/BASELINES.md b/docs/BASELINES.md new file mode 100644 index 0000000..e3a90b5 --- /dev/null +++ b/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. diff --git a/engine/_state/baseline.py b/engine/_state/baseline.py new file mode 100644 index 0000000..c6511df --- /dev/null +++ b/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", +] diff --git a/tests/commands/test_baseline_cmd.py b/tests/commands/test_baseline_cmd.py new file mode 100644 index 0000000..f042b20 --- /dev/null +++ b/tests/commands/test_baseline_cmd.py @@ -0,0 +1,59 @@ +"""CLI tests for finding baseline capture and enforcement.""" + +from __future__ import annotations + +from argparse import Namespace +from pathlib import Path + +import pytest + +from app.commands.baseline_cmd import cmd_baseline +from app.commands.helpers.runtime import CommandRuntime +from cli import create_parser + + +def test_parser_exposes_capture_and_check() -> None: + capture = create_parser().parse_args(["baseline", "capture", "--force"]) + check = create_parser().parse_args(["baseline", "check", "--max-new", "2", "--json"]) + assert capture.baseline_action == "capture" + assert capture.force is True + assert check.baseline_action == "check" + assert check.max_new == 2 + assert check.json is True + + +def test_check_uses_distinct_exit_code_for_regression(tmp_path: Path) -> None: + baseline_path = tmp_path / "baseline.json" + clean_state = { + "last_scan": "2026-08-24T00:00:00+00:00", + "scan_count": 1, + "findings": {}, + } + capture_args = Namespace( + baseline_action="capture", + output=str(baseline_path), + force=False, + runtime=CommandRuntime(config={}, state=clean_state, state_path=None), + ) + cmd_baseline(capture_args) + + finding = { + "id": "coupling::src/new.py", + "detector": "coupling", + "file": "src/new.py", + "tier": 2, + "confidence": "high", + "status": "open", + } + check_args = Namespace( + baseline_action="check", + baseline=str(baseline_path), + max_new=0, + json=False, + runtime=CommandRuntime( + config={}, state={**clean_state, "findings": {finding["id"]: finding}}, state_path=None + ), + ) + with pytest.raises(SystemExit) as exc: + cmd_baseline(check_args) + assert exc.value.code == 3 diff --git a/tests/review/test_review_commands.py b/tests/review/test_review_commands.py index 823fcfe..1d980c1 100644 --- a/tests/review/test_review_commands.py +++ b/tests/review/test_review_commands.py @@ -775,9 +775,9 @@ def test_do_run_batches_dry_run_generates_packet_and_prompts( assert len(packet_files) == 1 blind_packet = tmp_path / ".structorium" / "review_packet_blind.json" assert blind_packet.exists() - prompt_files = list(runs_dir.glob("*/prompts/batch-*.md")) + prompt_files = sorted(runs_dir.glob("*/prompts/batch-*.md")) assert len(prompt_files) == 2 - prompt_text = prompt_files[0].read_text() + prompt_text = "\n".join(path.read_text() for path in prompt_files) assert "Blind packet:" in prompt_text assert str(blind_packet) in prompt_text assert "Previously flagged issues" in prompt_text diff --git a/tests/state/test_baseline.py b/tests/state/test_baseline.py new file mode 100644 index 0000000..0f4962d --- /dev/null +++ b/tests/state/test_baseline.py @@ -0,0 +1,86 @@ +"""Finding baseline integrity and ratchet tests.""" + +from __future__ import annotations + +import copy +from pathlib import Path + +import pytest + +from engine._state.baseline import ( + build_baseline, + compare_baseline, + load_baseline, + write_baseline, +) + + +def _finding(fid: str, *, status: str = "open", suppressed: bool = False) -> dict: + return { + "id": fid, + "detector": "coupling", + "file": fid.rsplit("::", 1)[-1], + "tier": 2, + "confidence": "high", + "status": status, + "suppressed": suppressed, + } + + +def _state(*findings: dict) -> dict: + return { + "last_scan": "2026-08-24T00:00:00+00:00", + "scan_count": 4, + "findings": {item["id"]: item for item in findings}, + } + + +def test_baseline_only_captures_active_unsuppressed_findings() -> None: + document = build_baseline( + _state( + _finding("coupling::src/a.py"), + _finding("coupling::src/b.py", status="fixed"), + _finding("coupling::src/c.py", suppressed=True), + ) + ) + assert document["finding_count"] == 1 + assert document["findings"][0]["id"] == "coupling::src/a.py" + + +def test_compare_reports_new_resolved_and_unchanged() -> None: + baseline = build_baseline( + _state(_finding("coupling::src/a.py"), _finding("coupling::src/b.py")) + ) + diff = compare_baseline( + _state(_finding("coupling::src/b.py"), _finding("coupling::src/c.py")), + baseline, + ) + assert [item["id"] for item in diff["new"]] == ["coupling::src/c.py"] + assert [item["id"] for item in diff["resolved"]] == ["coupling::src/a.py"] + assert [item["id"] for item in diff["unchanged"]] == ["coupling::src/b.py"] + + +def test_checksum_rejects_silent_baseline_edit() -> None: + document = build_baseline(_state(_finding("coupling::src/a.py"))) + edited = copy.deepcopy(document) + edited["findings"][0]["tier"] = 4 + with pytest.raises(ValueError, match="checksum mismatch"): + compare_baseline(_state(), edited) + + +def test_write_refuses_implicit_overwrite_and_round_trips(tmp_path: Path) -> None: + path = tmp_path / ".structorium" / "baseline.json" + document = build_baseline(_state(_finding("coupling::src/a.py"))) + write_baseline(document, path) + assert load_baseline(path) == document + with pytest.raises(ValueError, match="already exists"): + write_baseline(document, path) + + +def test_fingerprint_is_independent_of_tier_and_confidence() -> None: + first = build_baseline(_state(_finding("coupling::src/a.py"))) + changed = _finding("coupling::src/a.py") + changed["tier"] = 4 + changed["confidence"] = "low" + second = build_baseline(_state(changed)) + assert first["findings"][0]["fingerprint"] == second["findings"][0]["fingerprint"]