generated from coulomb/repo-seed
tegwick
4620244f39
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>
122 lines
4.2 KiB
Python
122 lines
4.2 KiB
Python
#!/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:]))
|