81 lines
2.9 KiB
Python
81 lines
2.9 KiB
Python
"""
|
|
Failure Propagation — given a failed action, compute the forward blast radius.
|
|
|
|
The "blast radius" is the set of downstream actions that depend (transitively)
|
|
on the failed action and therefore cannot complete.
|
|
"""
|
|
|
|
from typing import Any, Dict, List, Set
|
|
from collections import defaultdict
|
|
|
|
|
|
class FailurePropagator:
|
|
"""
|
|
Maintains a reverse dependency index: for each action, the set of
|
|
actions that depend on it. Given a failed action, walks the reverse
|
|
graph to compute the impact set.
|
|
"""
|
|
|
|
def __init__(self):
|
|
# action -> set of actions that depend on it
|
|
self._revdeps: Dict[str, Set[str]] = defaultdict(set)
|
|
# action -> direct deps (so we can rebuild revdeps if needed)
|
|
self._deps: Dict[str, List[str]] = {}
|
|
|
|
# -----------------------------
|
|
# INGEST
|
|
# -----------------------------
|
|
def register_action(self, action: str, deps: List[str]):
|
|
"""Register an action and its direct dependencies."""
|
|
self._deps[action] = list(deps)
|
|
for d in deps:
|
|
self._revdeps[d].add(action)
|
|
|
|
def ingest_event(self, event: Dict[str, Any]):
|
|
"""Hook for EventBus subscription — picks up pipeline events
|
|
that declare dependencies and indexes them.
|
|
|
|
Handles both flat FesterEvent shape (deps in `meta.deps`) and
|
|
legacy nested shape (deps in `data.deps`).
|
|
"""
|
|
# Try flat shape first: action + deps at top level or in meta
|
|
action = event.get("action")
|
|
deps = event.get("deps")
|
|
if deps is None and isinstance(event.get("meta"), dict):
|
|
deps = event["meta"].get("deps")
|
|
# Legacy nested shape
|
|
if action is None or deps is None:
|
|
data = event.get("data")
|
|
if isinstance(data, dict):
|
|
if action is None:
|
|
action = data.get("action")
|
|
if deps is None:
|
|
deps = data.get("deps")
|
|
if action and deps:
|
|
self.register_action(action, deps)
|
|
|
|
# -----------------------------
|
|
# QUERY
|
|
# -----------------------------
|
|
def downstream(self, action: str) -> List[str]:
|
|
"""Return the transitive set of actions that depend on `action`."""
|
|
seen: Set[str] = set()
|
|
stack = [action]
|
|
while stack:
|
|
cur = stack.pop()
|
|
for child in self._revdeps.get(cur, set()):
|
|
if child not in seen:
|
|
seen.add(child)
|
|
stack.append(child)
|
|
return sorted(seen)
|
|
|
|
def impact_report(self, failed_action: str) -> Dict[str, Any]:
|
|
downstream = self.downstream(failed_action)
|
|
return {
|
|
"failed_action": failed_action,
|
|
"direct_dependents": sorted(self._revdeps.get(failed_action, set())),
|
|
"transitive_impact": downstream,
|
|
"impact_count": len(downstream),
|
|
"total_actions_known": len(self._deps),
|
|
}
|