Spaces:
Running on Zero
Running on Zero
| """Clue graph: gating, discovery, and deterministic reachability/fairness check. | |
| The clue graph (Section 6) gates what characters will discuss and what the world | |
| will surface. ``prerequisites`` for a node are its explicit ``requires`` plus any | |
| node that lists it in ``unlocks``. Fairness (Section 9.6) is a pure graph check. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from ..models import ClueNode | |
| class FairnessReport: | |
| ok: bool | |
| unreachable_required: list[str] | |
| cycles: list[list[str]] | |
| details: str = "" | |
| class ClueGraph: | |
| def __init__(self, nodes: list[ClueNode]) -> None: | |
| self.nodes = {n.id: n for n in nodes} | |
| self._prereqs = self._compute_prereqs(nodes) | |
| def _compute_prereqs(nodes: list[ClueNode]) -> dict[str, set[str]]: | |
| prereqs: dict[str, set[str]] = {n.id: set(n.requires) for n in nodes} | |
| for n in nodes: | |
| for target in n.unlocks: | |
| prereqs.setdefault(target, set()).add(n.id) | |
| # ensure all referenced ids exist as keys | |
| for n in nodes: | |
| prereqs.setdefault(n.id, set()) | |
| return prereqs | |
| def prerequisites(self, clue_id: str) -> set[str]: | |
| return self._prereqs.get(clue_id, set()) | |
| def is_unlocked(self, clue_id: str, discovered: set[str]) -> bool: | |
| """A node is discussable once all its prerequisites are discovered.""" | |
| return self.prerequisites(clue_id).issubset(discovered) | |
| def discoverable_now(self, discovered: set[str]) -> list[str]: | |
| return [ | |
| cid | |
| for cid in self.nodes | |
| if cid not in discovered and self.is_unlocked(cid, discovered) | |
| ] | |
| def exonerated_by(self, discovered: set[str]) -> set[str]: | |
| cleared: set[str] = set() | |
| for cid in discovered: | |
| node = self.nodes.get(cid) | |
| if node: | |
| cleared.update(node.exonerates) | |
| return cleared | |
| # -- fairness (Section 9.6) -------------------------------------------- | |
| def fairness(self) -> FairnessReport: | |
| """Every required clue must be reachable with no dependency cycle.""" | |
| cycles = self._find_cycles() | |
| in_cycle = {cid for cyc in cycles for cid in cyc} | |
| unreachable: list[str] = [] | |
| for cid, node in self.nodes.items(): | |
| if not node.required_for_solution: | |
| continue | |
| if not self._reachable(cid, in_cycle): | |
| unreachable.append(cid) | |
| ok = not unreachable and not cycles | |
| details = [] | |
| if cycles: | |
| details.append(f"dependency cycles: {cycles}") | |
| if unreachable: | |
| details.append(f"unreachable required clues: {unreachable}") | |
| return FairnessReport( | |
| ok=ok, | |
| unreachable_required=unreachable, | |
| cycles=cycles, | |
| details="; ".join(details) or "all required clues reachable, acyclic", | |
| ) | |
| def _reachable(self, clue_id: str, in_cycle: set[str]) -> bool: | |
| """Can we discover ``clue_id`` starting from the empty set?""" | |
| if clue_id in in_cycle: | |
| return False | |
| seen: set[str] = set() | |
| stack = [clue_id] | |
| while stack: | |
| cur = stack.pop() | |
| if cur in seen: | |
| continue | |
| seen.add(cur) | |
| prereqs = self.prerequisites(cur) | |
| for p in prereqs: | |
| if p in in_cycle: | |
| return False | |
| stack.append(p) | |
| return True | |
| def _find_cycles(self) -> list[list[str]]: | |
| """Detect cycles in the prerequisite DAG (DFS, simple cycle capture).""" | |
| WHITE, GRAY, BLACK = 0, 1, 2 | |
| color = {cid: WHITE for cid in self.nodes} | |
| cycles: list[list[str]] = [] | |
| path: list[str] = [] | |
| def dfs(node: str) -> None: | |
| color[node] = GRAY | |
| path.append(node) | |
| for nxt in self.prerequisites(node): | |
| if nxt not in color: | |
| continue | |
| if color[nxt] == GRAY: | |
| idx = path.index(nxt) | |
| cycles.append(path[idx:] + [nxt]) | |
| elif color[nxt] == WHITE: | |
| dfs(nxt) | |
| path.pop() | |
| color[node] = BLACK | |
| for cid in self.nodes: | |
| if color[cid] == WHITE: | |
| dfs(cid) | |
| return cycles | |