generated from coulomb/repo-seed
feat(explain): complete ATLAS-WP-0004 — graph edges, blast-radius, determinism tests
Some checks failed
validate-registry / validate (push) Has been cancelled
Some checks failed
validate-registry / validate (push) Has been cancelled
T05: tests/test_effective_config.py (6 tests) — order-independence, most-specific winner, no value/secret leak; wired into make validate + CI. T04: tools/blast_radius.py + make blast-radius — consumers, transitive dependents (cycle-safe), secret refs, fan-out risk band. T03: tools/config_graph.py + make graph/graph-query — emit config-typed edges to registry/indexes/graph.yaml (queryable by surface id); staleness check in the gate. WP-0004 finished (5/5). Read-first control-plane MVP complete: explain, graph, and blast-radius over the seeded surfaces. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
121
tools/blast_radius.py
Normal file
121
tools/blast_radius.py
Normal file
@@ -0,0 +1,121 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Blast-radius / dependency view (ATLAS-WP-0004-T04).
|
||||
|
||||
For a given configuration surface, traverse the relationship graph to list what a
|
||||
change would affect: direct consumers (services), dependent surfaces (those that
|
||||
reference or override it, transitively), referenced secrets, and the owner. This
|
||||
supports change-risk reasoning before a change -- the
|
||||
`config key -> service -> tenant -> feature -> secret -> owner` chain (research §5).
|
||||
|
||||
Reads only the registry (no live values). Reuses the surface loader from
|
||||
effective_config.
|
||||
|
||||
Usage:
|
||||
python3 tools/blast_radius.py surface.infotech.state-hub.api-config
|
||||
make blast-radius SURFACE=surface.infotech.state-hub.api-config
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections import deque
|
||||
|
||||
from effective_config import SURFACES_DIR, load_entry
|
||||
|
||||
|
||||
def load_all() -> dict[str, dict]:
|
||||
return {p.stem: load_entry(p) for p in sorted(SURFACES_DIR.glob("*.md"))}
|
||||
|
||||
|
||||
def relations(entry: dict) -> dict:
|
||||
return entry.get("relations", {}) or {}
|
||||
|
||||
|
||||
def build_reverse(entries: dict[str, dict]) -> dict[str, set[str]]:
|
||||
"""id -> set of surface ids that reference it via related_to or overrides."""
|
||||
rev: dict[str, set[str]] = {sid: set() for sid in entries}
|
||||
for sid, e in entries.items():
|
||||
rel = relations(e)
|
||||
for target in list(rel.get("related_to", [])) + list(rel.get("overrides", [])):
|
||||
rev.setdefault(target, set()).add(sid)
|
||||
return rev
|
||||
|
||||
|
||||
def dependent_closure(start: str, rev: dict[str, set[str]]) -> list[str]:
|
||||
"""All surfaces that depend on `start` (transitively), nearest first.
|
||||
|
||||
The start node is excluded from its own closure even when relationship cycles
|
||||
(e.g. mutual related_to edges) would otherwise loop back to it.
|
||||
"""
|
||||
seen: set[str] = {start}
|
||||
order: list[str] = []
|
||||
q = deque(sorted(rev.get(start, set())))
|
||||
while q:
|
||||
cur = q.popleft()
|
||||
if cur in seen:
|
||||
continue
|
||||
seen.add(cur)
|
||||
order.append(cur)
|
||||
for nxt in sorted(rev.get(cur, set())):
|
||||
if nxt not in seen:
|
||||
q.append(nxt)
|
||||
return order
|
||||
|
||||
|
||||
def risk_band(score: int) -> str:
|
||||
return "high" if score >= 4 else "medium" if score >= 2 else "low"
|
||||
|
||||
|
||||
def render(surface_id: str, entries: dict[str, dict]) -> str:
|
||||
if surface_id not in entries:
|
||||
raise KeyError(surface_id)
|
||||
e = entries[surface_id]
|
||||
rel = relations(e)
|
||||
rev = build_reverse(entries)
|
||||
|
||||
consumers = list(rel.get("consumed_by", []))
|
||||
secrets = list(rel.get("depends_on_secret", []))
|
||||
forward_related = list(rel.get("related_to", []))
|
||||
dependents = dependent_closure(surface_id, rev)
|
||||
|
||||
score = len(consumers) + len(dependents)
|
||||
lines: list[str] = []
|
||||
lines.append(f"blast radius {surface_id}")
|
||||
lines.append("")
|
||||
lines.append(f" {e.get('name', surface_id)}")
|
||||
lines.append(f" kind: {e.get('kind','?')} owner: {e.get('owner','?')} mutability: {e.get('mutability','?')}")
|
||||
lines.append(f" risk: {risk_band(score)} (fan-out {score}: {len(consumers)} consumer(s) + {len(dependents)} dependent surface(s))")
|
||||
lines.append("")
|
||||
lines.append(" direct consumers (services):")
|
||||
for c in consumers or ["(none)"]:
|
||||
lines.append(f" - {c}")
|
||||
lines.append(" dependent surfaces (reference/override this, transitively):")
|
||||
for d in dependents or ["(none)"]:
|
||||
owner = entries.get(d, {}).get("owner", "?")
|
||||
lines.append(f" - {d} [owner: {owner}]")
|
||||
lines.append(" referenced secrets (by reference only):")
|
||||
for s in secrets or ["(none)"]:
|
||||
lines.append(f" - {s}")
|
||||
if forward_related:
|
||||
lines.append(" related to:")
|
||||
for r in forward_related:
|
||||
lines.append(f" - {r}")
|
||||
lines.append("")
|
||||
lines.append(" · derived from registry relations only; no live values read")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if len(argv) != 1:
|
||||
print(__doc__)
|
||||
return 2
|
||||
entries = load_all()
|
||||
try:
|
||||
print(render(argv[0], entries))
|
||||
except KeyError as exc:
|
||||
print(f"error: no surface entry for id {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main(sys.argv[1:]))
|
||||
Reference in New Issue
Block a user