scripts.architecture_static_check
Reference › scripts › architecture_static_check
A static import-graph layer check that complements the runtime-oriented architecture_checker with a recorded baseline. No module or test imports it; it is run by hand or from a gate.
Invocation: main takes an optional path (default src/orchestrata_ultimate), -v/--verbose to list violating edges, --write-baseline FILE to save the current counts as JSON, and --baseline FILE to fail when the counts regress against a saved baseline. A missing path is an error.
collect walks the package and builds the import graph and the set of violations, which _kind classifies as upward or peer imports; _is_known tests an edge against the accepted exceptions. tarjan_scc finds strongly connected components, so import cycles are counted as cycles and as modules in cycles. build_report assembles per-scope upward and peer counts, the total and the cycle figures, stamping the report with the current git commit through _git_commit. flatten reduces the counts to a flat map for comparison. When writing a baseline, the edge lists are dropped unless -v is given, which keeps baselines compact. In ratchet mode main compares each flattened count with the baseline and fails on any increase, so existing debt is tolerated but cannot grow.
At a glance
| Source | scripts/architecture_static_check.py |
| Lines of code | 270 |
| Classes | 0 |
| Functions | 8 |
| Imports (internal) | none |
| Imported by | none |
| Tests that import it | none |
Class diagram
No classes; see Functions.
Dependency diagram
No internal imports and no importers.
API
Functions
| Function | Signature | Summary |
|---|---|---|
_git_commit |
_git_commit() -> str |
|
_kind |
_kind(violation: str) -> str |
|
_is_known |
_is_known(importer: str, imported: str) -> bool |
|
collect |
collect(root: Path) -> tuple[dict, dict[str, dict[str, str]], dict[str, set[str]]] |
Walk all modules; return (counts, edges, graph). |
tarjan_scc |
tarjan_scc(graph: dict[str, set[str]]) -> list[list[str]] |
Iterative Tarjan; returns SCCs of size > 1 (genuine cycles). |
build_report |
build_report(root: Path) -> dict |
|
flatten |
flatten(counts: dict) -> dict[str, int] |
|
main |
main(argv: list[str] | None = None) -> int |
Source
#!/usr/bin/env python3
"""Static import-graph layer check — the review-methodology complement to
architecture_checker.py.
architecture_checker.py is the *runtime* gate: top-level imports only,
function-scoped imports merely counted as accepted (RULED D7), TYPE_CHECKING
edges skipped. This script is the *static* gate used by the architecture
review (orchestrata-review-2026-09-17) and the migration plan: EVERY import
edge is classified by scope (top-level / TYPE_CHECKING / function-scoped) and
checked against the same LAYER_ASSIGNMENTS, so the full design debt is
visible and ratchetable.
It also runs a Tarjan SCC pass over the static module graph so the cycle
count (10 genuine cycles / 40 modules at the 2026-09-17 review) can be
ratcheted alongside.
Usage:
python scripts/architecture_static_check.py # report
python scripts/architecture_static_check.py --write-baseline scripts/architecture_static_baseline.json
python scripts/architecture_static_check.py --baseline scripts/architecture_static_baseline.json # CI ratchet
Ratchet semantics: any count ABOVE the baseline fails the gate; improvements
(counts below baseline) pass and should be baked in by regenerating the
baseline in the same PR.
"""
from __future__ import annotations
import argparse
import ast
import json
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from architecture_checker import ( # noqa: E402
KNOWN_VIOLATIONS,
_is_inside_def,
_is_inside_type_checking,
_iter_ultimate_imports,
check_import,
)
SCOPES = ("top_level", "type_checking", "function_scoped")
KINDS = ("upward", "peer")
def _git_commit() -> str:
try:
return subprocess.run(
["git", "rev-parse", "--short", "HEAD"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
except Exception:
return "unknown"
def _kind(violation: str) -> str:
return "upward" if violation.startswith("upward import:") else "peer"
def _is_known(importer: str, imported: str) -> bool:
return any(
importer.startswith(imp_pat) and imported.startswith(ed_pat)
for imp_pat, ed_pat in KNOWN_VIOLATIONS
)
def collect(root: Path) -> tuple[dict, dict[str, dict[str, str]], dict[str, set[str]]]:
"""Walk all modules; return (counts, edges, graph).
counts: {scope: {kind: n}} plus totals.
edges: {scope: {kind: "importer -> imported"}} for reporting.
graph: module -> set(imported modules) over ALL scopes (for SCC).
"""
src_root = root.parent
counts: dict = {s: {k: 0 for k in KINDS} for s in SCOPES}
edges: dict[str, dict[str, str]] = {s: {} for s in SCOPES}
graph: dict[str, set[str]] = {}
files = 0
for py in sorted(root.rglob("*.py")):
if "__pycache__" in py.parts:
continue
rel = py.relative_to(src_root)
module = str(rel.with_suffix("")).replace("/", ".").replace("\\", ".")
try:
tree = ast.parse(py.read_text(encoding="utf-8"))
except OSError, SyntaxError, UnicodeDecodeError:
continue
files += 1
graph.setdefault(module, set())
for node, imported in _iter_ultimate_imports(tree):
imported_top = imported # module-level target, as the checker sees it
graph[module].add(imported_top)
violation = check_import(module, imported)
if violation is None or _is_known(module, imported):
continue
if _is_inside_type_checking(node, tree):
scope = "type_checking"
elif _is_inside_def(node, tree):
scope = "function_scoped"
else:
scope = "top_level"
kind = _kind(violation)
counts[scope][kind] += 1
edges[scope][f"{module} -> {imported}"] = kind
counts["total"] = sum(counts[s][k] for s in SCOPES for k in KINDS)
counts["files"] = files
return counts, edges, graph
def tarjan_scc(graph: dict[str, set[str]]) -> list[list[str]]:
"""Iterative Tarjan; returns SCCs of size > 1 (genuine cycles)."""
index_of: dict[str, int] = {}
lowlink: dict[str, int] = {}
on_stack: set[str] = set()
stack: list[str] = []
sccs: list[list[str]] = []
counter = [0]
for start in graph:
if start in index_of:
continue
work = [(start, iter(sorted(graph.get(start, ()))))]
index_of[start] = lowlink[start] = counter[0]
counter[0] += 1
stack.append(start)
on_stack.add(start)
while work:
node, it = work[-1]
advanced = False
for nxt in it:
if nxt not in graph:
continue
if nxt not in index_of:
index_of[nxt] = lowlink[nxt] = counter[0]
counter[0] += 1
stack.append(nxt)
on_stack.add(nxt)
work.append((nxt, iter(sorted(graph.get(nxt, ())))))
advanced = True
break
if nxt in on_stack:
lowlink[node] = min(lowlink[node], index_of[nxt])
if advanced:
continue
work.pop()
if work:
parent = work[-1][0]
lowlink[parent] = min(lowlink[parent], lowlink[node])
if lowlink[node] == index_of[node]:
scc = []
while True:
w = stack.pop()
on_stack.discard(w)
scc.append(w)
if w == node:
break
if len(scc) > 1:
sccs.append(sorted(scc))
return sccs
def build_report(root: Path) -> dict:
counts, edges, graph = collect(root)
sccs = tarjan_scc(graph)
return {
"recorded_at": __import__("datetime").date.today().isoformat(),
"git_commit": _git_commit(),
"root": str(root),
"counts": counts,
"cycles": {
"count": len(sccs),
"modules_in_cycles": sum(len(s) for s in sccs),
"members": sccs,
},
"edges": edges,
}
def flatten(counts: dict) -> dict[str, int]:
flat = {f"{s}.{k}": counts[s][k] for s in SCOPES for k in KINDS}
flat["total"] = counts["total"]
return flat
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("path", nargs="?", default="src/orchestrata_ultimate")
parser.add_argument("--write-baseline", metavar="FILE")
parser.add_argument(
"--baseline", metavar="FILE", help="ratchet: fail on regression vs FILE"
)
parser.add_argument(
"-v", "--verbose", action="store_true", help="list violating edges"
)
args = parser.parse_args(argv)
root = Path(args.path)
if not root.exists():
print(f"ERROR: path does not exist: {root}", file=sys.stderr)
return 1
report = build_report(root)
c = report["counts"]
print(f"Static layer check under {root.resolve()} ({c['files']} files)")
for s in SCOPES:
print(f" {s:15s} upward={c[s]['upward']:<4d} peer={c[s]['peer']:<4d}")
print(f" {'TOTAL':15s} {c['total']} violations")
print(
f" cycles: {report['cycles']['count']} SCCs "
f"({report['cycles']['modules_in_cycles']} modules)"
)
if args.verbose:
for s in SCOPES:
for edge, kind in sorted(report["edges"][s].items()):
print(f" [{s}/{kind}] {edge}")
if args.write_baseline:
out = dict(report)
if not args.verbose:
out.pop("edges") # baselines stay compact; -v writes them for review
Path(args.write_baseline).write_text(json.dumps(out, indent=2) + "\n")
print(f"Baseline written to {args.write_baseline}")
return 0
if args.baseline:
base = json.loads(Path(args.baseline).read_text())
failures = []
for key, now in flatten(c).items():
then = base["counts"]
ref = (
then["total"]
if key == "total"
else then[key.split(".")[0]][key.split(".")[1]]
)
if now > ref:
failures.append(f" RATCHET {key}: {now} > baseline {ref}")
for key, now, ref in (
("cycles.count", report["cycles"]["count"], base["cycles"]["count"]),
(
"cycles.modules",
report["cycles"]["modules_in_cycles"],
base["cycles"]["modules_in_cycles"],
),
):
if now > ref:
failures.append(f" RATCHET {key}: {now} > baseline {ref}")
if failures:
print("RATCHET FAILED — static architecture debt increased:")
print("\n".join(failures))
print(
"Fix the new violations, or regenerate the baseline with "
"--write-baseline if this is a sanctioned, documented exception."
)
return 1
print(
f"Ratchet OK — all counts ≤ baseline ({base['git_commit']}, {base['recorded_at']})"
)
return 0
if __name__ == "__main__":
sys.exit(main())