diff --git a/docs/architecture-policy.md b/docs/architecture-policy.md new file mode 100644 index 0000000..ef15ff5 --- /dev/null +++ b/docs/architecture-policy.md @@ -0,0 +1,26 @@ +# Architecture policy as code + +Place `structorium.toml` at the scan root. Rules operate on the dependency +graph already built by the active language plugin and become normal ranked +findings, including in the new-code gate. + +```toml +[architecture] +enabled = true + +[[architecture.rules]] +id = "domain-does-not-import-web" +from = "src/domain/**" +deny = ["src/web/**", "src/infrastructure/http/**"] +tier = 1 + +[[architecture.rules.exceptions]] +from = "src/domain/legacy.py" +to = "src/infrastructure/http/legacy.py" +until = 2026-10-01 +``` + +Paths are root-relative, slash-normalized globs. An exception stops applying +after `until`; the resulting finding records that the exception expired. +Invalid TOML, duplicate IDs, empty globs, and tiers outside 1–4 fail the scan +instead of silently weakening policy. diff --git a/engine/planning/incremental_cache.py b/engine/planning/incremental_cache.py index d580dbb..0af7bf7 100644 --- a/engine/planning/incremental_cache.py +++ b/engine/planning/incremental_cache.py @@ -53,6 +53,9 @@ def _source_files(scan_path: Path, lang: LangRun) -> list[Path]: path = path.resolve() if path.is_file() and (path.is_relative_to(scan_root) or scan_root == root): resolved.add(path) + policy = root / "structorium.toml" + if policy.is_file(): + resolved.add(policy) return sorted(resolved) diff --git a/engine/planning/scan.py b/engine/planning/scan.py index d236e2d..9fce95b 100644 --- a/engine/planning/scan.py +++ b/engine/planning/scan.py @@ -148,6 +148,12 @@ def _generate_findings_from_lang( all_potentials = dict(cache_lookup.potentials or {}) else: findings, all_potentials = _run_phases(path, lang, phases) + from engine.policy.architecture import detect_architecture_policy + + policy_findings, policy_edges = detect_architecture_policy(path, lang) + findings.extend(policy_findings) + if policy_edges: + all_potentials["architecture_policy"] = policy_edges _stamp_finding_context(findings, lang) if cache_lookup is not None and not cache_lookup.hit: from engine.planning.incremental_cache import store_scan_cache diff --git a/engine/policy/architecture.py b/engine/policy/architecture.py new file mode 100644 index 0000000..282cd48 --- /dev/null +++ b/engine/policy/architecture.py @@ -0,0 +1,201 @@ +"""Architecture policy-as-code rules evaluated against language dependency graphs.""" + +from __future__ import annotations + +import fnmatch +import tomllib +from dataclasses import dataclass, field +from datetime import date +from pathlib import Path +from typing import Any + +import state as state_mod +from languages._framework.runtime import LangRun + + +@dataclass(frozen=True) +class PolicyException: + source: str + target: str + until: date | None = None + + +@dataclass(frozen=True) +class ArchitectureRule: + id: str + source: str + denied: tuple[str, ...] + tier: int = 2 + exceptions: tuple[PolicyException, ...] = field(default_factory=tuple) + + +class ArchitecturePolicyError(ValueError): + """Raised when `structorium.toml` contains an invalid policy contract.""" + + +def _as_patterns(value: object, *, field_name: str) -> tuple[str, ...]: + if value is None: + raise ArchitecturePolicyError(f"{field_name} must contain a path glob") + values = value if isinstance(value, list) else [value] + patterns = tuple(str(item).strip() for item in values if str(item).strip()) + if not patterns: + raise ArchitecturePolicyError(f"{field_name} must contain a path glob") + return patterns + + +def _parse_exception(payload: object, rule_id: str) -> PolicyException: + if not isinstance(payload, dict): + raise ArchitecturePolicyError(f"rule {rule_id}: exceptions must be tables") + until = payload.get("until") + if until is not None and not isinstance(until, date): + try: + until = date.fromisoformat(str(until)) + except ValueError as exc: + raise ArchitecturePolicyError( + f"rule {rule_id}: exception until must be YYYY-MM-DD" + ) from exc + return PolicyException( + source=str(payload.get("from", "**")), + target=str(payload.get("to", "**")), + until=until, + ) + + +def load_architecture_rules(root: Path) -> tuple[ArchitectureRule, ...]: + """Load validated rules from `/structorium.toml`.""" + policy_path = root / "structorium.toml" + if not policy_path.is_file(): + return () + try: + document = tomllib.loads(policy_path.read_text(encoding="utf-8")) + except (OSError, tomllib.TOMLDecodeError) as exc: + raise ArchitecturePolicyError(f"cannot parse {policy_path}: {exc}") from exc + architecture = document.get("architecture", {}) + if not isinstance(architecture, dict) or architecture.get("enabled", True) is False: + return () + raw_rules = architecture.get("rules", []) + if not isinstance(raw_rules, list): + raise ArchitecturePolicyError("architecture.rules must be an array of tables") + rules: list[ArchitectureRule] = [] + seen: set[str] = set() + for index, raw in enumerate(raw_rules, start=1): + if not isinstance(raw, dict): + raise ArchitecturePolicyError(f"rule {index} must be a table") + rule_id = str(raw.get("id", f"rule-{index}")).strip() + if not rule_id or rule_id in seen: + raise ArchitecturePolicyError(f"duplicate or empty rule id: {rule_id!r}") + seen.add(rule_id) + tier = int(raw.get("tier", 2)) + if tier not in {1, 2, 3, 4}: + raise ArchitecturePolicyError(f"rule {rule_id}: tier must be 1..4") + rules.append( + ArchitectureRule( + id=rule_id, + source=_as_patterns(raw.get("from"), field_name=f"rule {rule_id}.from")[0], + denied=_as_patterns(raw.get("deny"), field_name=f"rule {rule_id}.deny"), + tier=tier, + exceptions=tuple( + _parse_exception(item, rule_id) + for item in raw.get("exceptions", []) + ), + ) + ) + return tuple(rules) + + +def _relative(value: object, root: Path) -> str: + path = Path(str(value).replace("\\", "/")) + if path.is_absolute(): + try: + return path.resolve().relative_to(root.resolve()).as_posix() + except ValueError: + return path.as_posix() + return path.as_posix().lstrip("./") + + +def _matching_exception( + rule: ArchitectureRule, source: str, target: str, today: date +) -> tuple[bool, bool]: + for exception in rule.exceptions: + if not ( + fnmatch.fnmatchcase(source, exception.source) + and fnmatch.fnmatchcase(target, exception.target) + ): + continue + if exception.until is None or exception.until >= today: + return True, False + return False, True + return False, False + + +def evaluate_architecture_rules( + graph: dict[str, dict[str, Any]], + rules: tuple[ArchitectureRule, ...], + *, + root: Path, + today: date | None = None, +) -> tuple[list[dict[str, Any]], int]: + """Return normalized findings and the number of policy-eligible edges.""" + current_date = today or date.today() + findings: list[dict[str, Any]] = [] + eligible = 0 + for source_value, entry in sorted(graph.items()): + source = _relative(source_value, root) + imports = entry.get("imports", []) if isinstance(entry, dict) else [] + for target_value in sorted(imports): + target = _relative(target_value, root) + for rule in rules: + if not fnmatch.fnmatchcase(source, rule.source): + continue + denied = next( + (pattern for pattern in rule.denied if fnmatch.fnmatchcase(target, pattern)), + None, + ) + if denied is None: + continue + eligible += 1 + exempt, expired = _matching_exception( + rule, source, target, current_date + ) + if exempt: + continue + suffix = " (an exception expired)" if expired else "" + findings.append( + state_mod.make_finding( + "architecture_policy", + source, + f"{rule.id}:{target}", + tier=rule.tier, + confidence="high", + summary=f"{rule.id}: {source} must not depend on {target}{suffix}", + detail={ + "rule_id": rule.id, + "target": target, + "denied_pattern": denied, + "expired_exception": expired, + }, + ) + ) + return findings, eligible + + +def detect_architecture_policy( + path: Path, lang: LangRun +) -> tuple[list[dict[str, Any]], int]: + """Load project rules and evaluate them using the active language graph.""" + rules = load_architecture_rules(path.resolve()) + if not rules: + return [], 0 + graph = lang.dep_graph or lang.build_dep_graph(path) + lang.dep_graph = graph + return evaluate_architecture_rules(graph, rules, root=path.resolve()) + + +__all__ = [ + "ArchitecturePolicyError", + "ArchitectureRule", + "PolicyException", + "detect_architecture_policy", + "evaluate_architecture_rules", + "load_architecture_rules", +] diff --git a/tests/detectors/test_architecture_policy.py b/tests/detectors/test_architecture_policy.py new file mode 100644 index 0000000..cf4ed79 --- /dev/null +++ b/tests/detectors/test_architecture_policy.py @@ -0,0 +1,78 @@ +from datetime import date +from pathlib import Path + +import pytest + +from engine.policy.architecture import ( + ArchitecturePolicyError, + ArchitectureRule, + PolicyException, + evaluate_architecture_rules, + load_architecture_rules, +) + + +def _graph(root: Path): + return { + str(root / "src/domain/orders.py"): { + "imports": [str(root / "src/web/routes.py")] + } + } + + +def test_policy_detects_forbidden_dependency(tmp_path: Path) -> None: + rules = (ArchitectureRule("domain-purity", "src/domain/**", ("src/web/**",)),) + + findings, eligible = evaluate_architecture_rules( + _graph(tmp_path), rules, root=tmp_path + ) + + assert eligible == 1 + assert findings[0]["detector"] == "architecture_policy" + assert findings[0]["detail"]["rule_id"] == "domain-purity" + + +def test_active_exception_suppresses_and_expired_exception_reports(tmp_path: Path) -> None: + active = ArchitectureRule( + "domain-purity", + "src/domain/**", + ("src/web/**",), + exceptions=(PolicyException("**", "**", date(2030, 1, 1)),), + ) + expired = ArchitectureRule( + "domain-purity", + "src/domain/**", + ("src/web/**",), + exceptions=(PolicyException("**", "**", date(2020, 1, 1)),), + ) + + assert evaluate_architecture_rules( + _graph(tmp_path), (active,), root=tmp_path, today=date(2026, 1, 1) + )[0] == [] + finding = evaluate_architecture_rules( + _graph(tmp_path), (expired,), root=tmp_path, today=date(2026, 1, 1) + )[0][0] + assert finding["detail"]["expired_exception"] is True + + +def test_loads_toml_and_rejects_duplicate_ids(tmp_path: Path) -> None: + policy = tmp_path / "structorium.toml" + policy.write_text( + """ +[architecture] +enabled = true +[[architecture.rules]] +id = "domain-purity" +from = "src/domain/**" +deny = ["src/web/**", "src/db/**"] +tier = 1 +""" + ) + assert load_architecture_rules(tmp_path)[0].tier == 1 + + policy.write_text( + policy.read_text() + + '\n[[architecture.rules]]\nid = "domain-purity"\nfrom = "a/**"\ndeny = "b/**"\n' + ) + with pytest.raises(ArchitecturePolicyError, match="duplicate"): + load_architecture_rules(tmp_path)