from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Dict, Tuple

@dataclass(frozen=True)
class Node:
    name: str
    deps: Tuple[str, ...]
    fn: Callable[[], str]


def run_dag(nodes: Dict[str, Node]):
    # 输入验证与环检测使用 Kahn 拓扑排序思想。
    if any(d not in nodes for n in nodes.values() for d in n.deps):
        raise ValueError("unknown dependency")
    if any(n.name != key for key, n in nodes.items()):
        raise ValueError("node key/name mismatch")
    indegree = {k: len(set(n.deps)) for k, n in nodes.items()}
    dependents = {k: [] for k in nodes}
    for key, node in nodes.items():
        for dep in set(node.deps):
            dependents[dep].append(key)
    ready = sorted(k for k, deg in indegree.items() if deg == 0)
    order = []
    while ready:
        key = ready.pop(0)
        order.append(key)
        for child in dependents[key]:
            indegree[child] -= 1
            if indegree[child] == 0:
                ready.append(child)
                ready.sort()
    if len(order) != len(nodes):
        raise ValueError("dependency cycle")

    states = {}
    outputs = {}
    for key in order:
        node = nodes[key]
        if any(states[d] != "SUCCEEDED" for d in node.deps):
            states[key] = "BLOCKED"
            continue
        try:
            outputs[key] = node.fn()
            states[key] = "SUCCEEDED"
        except Exception as exc:
            states[key] = "FAILED"
            outputs[key] = type(exc).__name__
    return states, outputs, order


def ok():
    return "accepted"


def fail():
    raise RuntimeError("source unavailable")


if __name__ == "__main__":
    plan = {
        "topic": Node("topic", (), ok),
        "sources": Node("sources", ("topic",), ok),
        "review": Node("review", ("sources",), ok),
    }
    state, out, order = run_dag(plan)
    assert all(v == "SUCCEEDED" for v in state.values())
    assert order.index("topic") < order.index("sources") < order.index("review")

    broken = {
        "topic": Node("topic", (), ok),
        "sources": Node("sources", ("topic",), fail),
        "review": Node("review", ("sources",), ok),
    }
    state, _, _ = run_dag(broken)
    assert state["sources"] == "FAILED"
    assert state["review"] == "BLOCKED"

    cyclic = {
        "a": Node("a", ("b",), ok),
        "b": Node("b", ("a",), ok),
    }
    try:
        run_dag(cyclic)
        raise AssertionError("cycle not rejected")
    except ValueError as exc:
        assert str(exc) == "dependency cycle"

    print("PASS: dependency ordering, failure blocking, cycle detection")