"""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 @dataclass 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) @staticmethod 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