"""Dependency DAG for multi-agent coordination reservations. Agents working in parallel often have strict ordering constraints: agent B cannot safely start its work until agent A completes and releases its reservation. This module implements a lightweight **dependency DAG** that layers on top of the existing reservation system to express and evaluate those constraints without any central coordinator. Layout ------ :: .muse/coordination/ dependencies/.json — dependency record (write-once) Dependency schema ----------------- :: { "schema_version": "", "reservation_id": "", // the *dependent* reservation "depends_on": ["", ...], // must complete (release/expire) first "created_at": "2026-03-30T12:00:00+00:00" } A reservation is **blocked** when at least one of its declared dependencies is still *active* (not yet released and not yet expired). Agents poll :func:`is_blocked` to decide whether to start work or wait. Graph model ----------- Edges are directed from *dependent → dependency* ("A depends on B" is the edge ``A → B``). Topological order therefore lists dependencies before dependents — the correct execution sequence for a pipeline. DAG invariants -------------- * **Acyclicity** is enforced at write time by :func:`add_dependencies`. Any call that would introduce a cycle raises :exc:`ValueError` immediately, so a cycle can never exist on disk. * **Self-dependency** (``reservation_id in depends_on``) is rejected. * **Write-once**: each reservation has at most one dependency record. Calling :func:`add_dependencies` twice for the same ID raises :exc:`FileExistsError`. * **UUID-safe**: every ID is validated before constructing any file path, preventing directory traversal attacks. Security -------- Every reservation ID supplied by callers is validated against a UUID4 regex (same pattern as :func:`~muse.core.coordination._validate_reservation_id`). A valid UUID cannot contain ``/`` or ``..``, ruling out path traversal. Performance ----------- * :func:`load_all_dependencies` — one directory scan: O(n). * :func:`detect_cycle` — iterative DFS: O(V + E). * :func:`topological_sort` — Kahn's algorithm: O(V + E). * :func:`is_blocked` / :func:`get_blocking` — O(degree) set intersection. All graph operations are pure in-memory after the initial load. """ from __future__ import annotations import datetime import json import logging import pathlib import re from collections import deque from typing import Sequence from muse._version import __version__ from muse.core.store import write_text_atomic # JSON-compatible value type for dependency record serialization. JsonValue = str | int | float | bool | None | list["JsonValue"] | "JsonDict" JsonDict = dict[str, JsonValue] type AdjacencyMap = dict[str, set[str]] # node → set of direct dependencies type ColorMap = dict[str, int] # node → DFS color (WHITE/GREY/BLACK) type ParentMap = dict[str, str | None] # node → parent in DFS traversal type InDegreeMap = dict[str, int] # node → in-degree count logger = logging.getLogger(__name__) # ── Constants ───────────────────────────────────────────────────────────────── # UUID4 regex — identical to coordination._UUID_RE for consistency. _UUID_RE = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", re.IGNORECASE, ) # Maximum number of direct dependencies per reservation. _MAX_DEPS: int = 256 # ── Directory helpers ───────────────────────────────────────────────────────── def _coord_dir(root: pathlib.Path) -> pathlib.Path: """Return ``.muse/coordination/`` for *root*.""" return root / ".muse" / "coordination" def _dependencies_dir(root: pathlib.Path) -> pathlib.Path: """Return ``.muse/coordination/dependencies/`` for *root*.""" return _coord_dir(root) / "dependencies" def ensure_dag_dirs(root: pathlib.Path) -> None: """Create the dependencies directory if it does not exist.""" _dependencies_dir(root).mkdir(parents=True, exist_ok=True) # ── Validation ──────────────────────────────────────────────────────────────── def _validate_id(rid: str) -> None: """Raise :exc:`ValueError` if *rid* is not a well-formed UUID4. Security note ------------- ``rid`` is used to construct file paths inside ``.muse/coordination/dependencies/``. A valid UUID4 cannot contain ``/`` or ``..``, preventing directory traversal attacks. """ if not _UUID_RE.match(str(rid)): raise ValueError(f"reservation_id must be a valid UUID4: {rid!r}") # ── Time helpers ────────────────────────────────────────────────────────────── def _now_utc() -> datetime.datetime: return datetime.datetime.now(datetime.timezone.utc) def _parse_dt(s: str) -> datetime.datetime: return datetime.datetime.fromisoformat(s) # ── Data model ──────────────────────────────────────────────────────────────── class DependencyRecord: """Stores the direct dependency edges for one reservation. A :class:`DependencyRecord` is **write-once**: once created for a given ``reservation_id`` it is never modified. It declares that ``reservation_id`` must wait for every ID in ``depends_on`` to be released or expired before starting work. Attributes ---------- reservation_id: The *dependent* reservation — the one that must wait. depends_on: List of reservation IDs that must complete (be released or expire) before the dependent reservation is unblocked. De-duplicated and order-preserving at write time. created_at: UTC time when this dependency record was written to disk. """ def __init__( self, reservation_id: str, depends_on: list[str], created_at: datetime.datetime, ) -> None: self.reservation_id = reservation_id self.depends_on = depends_on self.created_at = created_at def to_dict(self) -> JsonDict: """Serialise to a JSON-compatible dict.""" return { "schema_version": __version__, "reservation_id": self.reservation_id, "depends_on": self.depends_on, "created_at": self.created_at.isoformat(), } @classmethod def from_dict(cls, d: JsonDict) -> "DependencyRecord": """Deserialise from a JSON-compatible dict. Missing fields default to safe values so records written by older versions remain loadable. """ created_raw = d.get("created_at") created_at = _parse_dt(str(created_raw)) if created_raw else _now_utc() deps_raw = d.get("depends_on", []) depends_on = ( [str(x) for x in deps_raw if x] if isinstance(deps_raw, list) else [] ) return cls( reservation_id=str(d.get("reservation_id", "")), depends_on=depends_on, created_at=created_at, ) # ── Loading helpers ─────────────────────────────────────────────────────────── def load_all_dependencies(root: pathlib.Path) -> list[DependencyRecord]: """Return all dependency records from ``.muse/coordination/dependencies/``. Corrupt or unreadable files are skipped with a warning — this mirrors the resilient loading strategy used throughout the coordination layer. """ ddir = _dependencies_dir(root) if not ddir.exists(): return [] records: list[DependencyRecord] = [] for path in ddir.glob("*.json"): try: raw = json.loads(path.read_text()) records.append(DependencyRecord.from_dict(raw)) except (json.JSONDecodeError, KeyError, ValueError) as exc: logger.warning("Corrupt dependency file %s: %s", path.name, exc) return records def load_dependencies( root: pathlib.Path, reservation_id: str, ) -> DependencyRecord | None: """Load the dependency record for *reservation_id*, or ``None`` if absent. Parameters ---------- root: Repository root. reservation_id: UUID4 of the reservation to look up. Raises ------ ValueError If *reservation_id* is not a valid UUID4. """ _validate_id(reservation_id) path = _dependencies_dir(root) / f"{reservation_id}.json" try: return DependencyRecord.from_dict(json.loads(path.read_text())) except (OSError, json.JSONDecodeError, KeyError, ValueError): return None # ── Graph construction ──────────────────────────────────────────────────────── def build_graph(deps: Sequence[DependencyRecord]) -> AdjacencyMap: """Return an adjacency map ``node → {direct_dependencies}``. Every node that appears (either as a dependent or as a dependency) is included in the returned graph with an explicit entry, so the set of graph nodes is complete. Nodes that are referenced only as dependencies (no record of their own) get an empty dependency set. Parameters ---------- deps: Sequence of :class:`DependencyRecord` objects, typically from :func:`load_all_dependencies`. Returns ------- dict[str, set[str]] Adjacency map: ``node → {ids this node directly depends on}``. """ graph: AdjacencyMap = {} for rec in deps: if rec.reservation_id not in graph: graph[rec.reservation_id] = set() for dep in rec.depends_on: graph[rec.reservation_id].add(dep) if dep not in graph: graph[dep] = set() return graph # ── Cycle detection ─────────────────────────────────────────────────────────── def detect_cycle(graph: AdjacencyMap) -> list[str] | None: """Return a cycle path if *graph* contains a directed cycle, else ``None``. Uses iterative DFS with three-colour marking: * **WHITE** (0) — not yet visited. * **GREY** (1) — currently on the DFS stack (back edge → cycle). * **BLACK** (2) — fully explored. Returns ------- list[str] | None A list of reservation IDs forming the cycle (first == last), or ``None`` if the graph is acyclic. Notes ----- Nodes that appear in the ``depends_on`` lists but have no record of their own are treated as leaf nodes with no outgoing edges — they cannot be part of a cycle. """ WHITE, GREY, BLACK = 0, 1, 2 color: ColorMap = {n: WHITE for n in graph} parent: ParentMap = {n: None for n in graph} def _reconstruct(node: str) -> list[str]: """Walk parent pointers back to *node* to extract the cycle.""" path = [node] cur = parent.get(node) while cur is not None and cur != node: path.append(cur) cur = parent.get(cur) path.append(node) path.reverse() return path for start in list(graph): if color.get(start, WHITE) != WHITE: continue stack = [(start, iter(graph.get(start, set())))] color[start] = GREY while stack: node, children = stack[-1] try: child = next(children) if child not in color: # Leaf node (dependency with no record) — cannot cycle. continue if color[child] == GREY: # Back edge detected — cycle found. parent[child] = node return _reconstruct(child) if color[child] == WHITE: color[child] = GREY parent[child] = node stack.append((child, iter(graph.get(child, set())))) except StopIteration: color[node] = BLACK stack.pop() return None # ── Topological sort ────────────────────────────────────────────────────────── def topological_sort( graph: AdjacencyMap, nodes: Sequence[str] | None = None, ) -> list[str]: """Return nodes in topological order using Kahn's algorithm. Dependencies appear **before** their dependents in the returned list — i.e. the correct execution order for a pipeline where each step must wait for its declared predecessors. Parameters ---------- graph: Adjacency map from :func:`build_graph`. nodes: Optional subset of node IDs to sort. When supplied, the function restricts to the subgraph reachable from those nodes (including all transitive dependencies). When ``None``, the full graph is sorted. Returns ------- list[str] Node IDs in execution order (dependencies first). Raises ------ ValueError If the graph contains a cycle. Call :func:`detect_cycle` first to get a human-readable description of the cycle. """ if nodes is not None: # Restrict to the subgraph reachable (downward through deps) from # the requested nodes. visited: set[str] = set() queue: deque[str] = deque(nodes) while queue: n = queue.popleft() if n in visited: continue visited.add(n) for dep in graph.get(n, set()): queue.append(dep) sub: AdjacencyMap = { n: {d for d in graph.get(n, set()) if d in visited} for n in visited } else: sub = {n: set(deps) for n, deps in graph.items()} # Build in-degree and reverse adjacency map (dep → {dependents}). in_degree: InDegreeMap = {n: len(sub[n]) for n in sub} reverse: AdjacencyMap = {n: set() for n in sub} for n, deps in sub.items(): for dep in deps: if dep in reverse: reverse[dep].add(n) # Kahn's: repeatedly dequeue nodes with no remaining prerequisites. ready: deque[str] = deque( sorted(n for n in sub if in_degree[n] == 0) ) result: list[str] = [] while ready: n = ready.popleft() result.append(n) for dependent in sorted(reverse.get(n, set())): in_degree[dependent] -= 1 if in_degree[dependent] == 0: ready.append(dependent) if len(result) != len(sub): raise ValueError( "DAG contains a cycle — call detect_cycle() for details" ) return result # ── Blocking status ─────────────────────────────────────────────────────────── def is_blocked( reservation_id: str, graph: AdjacencyMap, active_ids: frozenset[str], ) -> bool: """Return ``True`` if *reservation_id* has at least one active dependency. A reservation is *blocked* when any of its direct dependencies are still present in *active_ids* (i.e. not yet released and not yet expired). Parameters ---------- reservation_id: The UUID of the reservation to check. graph: Adjacency map from :func:`build_graph`. active_ids: Set of currently active reservation IDs, typically derived from :func:`~muse.core.coordination.active_reservations`. Returns ------- bool ``True`` if any direct dependency is still active. """ return bool(graph.get(reservation_id, set()) & active_ids) def get_blocking( reservation_id: str, graph: AdjacencyMap, active_ids: frozenset[str], ) -> list[str]: """Return the direct dependencies of *reservation_id* that are still active. Parameters ---------- reservation_id: The UUID of the reservation to check. graph: Adjacency map from :func:`build_graph`. active_ids: Set of currently active reservation IDs. Returns ------- list[str] Sorted list of active dependency IDs that are blocking *reservation_id*. Empty when the reservation is unblocked. """ return sorted(graph.get(reservation_id, set()) & active_ids) # ── Write: add_dependencies ─────────────────────────────────────────────────── def add_dependencies( root: pathlib.Path, reservation_id: str, depends_on: list[str], ) -> DependencyRecord: """Declare that *reservation_id* depends on each ID in *depends_on*. This is the sole write function for the dependency layer. It validates all IDs, enforces the acyclicity invariant against the current on-disk graph, and writes the record atomically. Parameters ---------- root: Repository root (directory containing ``.muse/``). reservation_id: UUID4 of the *dependent* reservation — the one that must wait. depends_on: List of UUID4 reservation IDs that must be released or expire before *reservation_id* is unblocked. Duplicates are silently de-duplicated (order preserved). An empty list is accepted and produces a no-op dependency record that can serve as an explicit audit marker. Returns ------- DependencyRecord The newly created dependency record. Raises ------ ValueError If any ID is not a valid UUID4, if ``reservation_id in depends_on`` (self-dependency), if more than :data:`_MAX_DEPS` dependencies are supplied, or if the new edge(s) would create a cycle in the existing DAG. FileExistsError If a dependency record for *reservation_id* already exists. Dependency declarations are write-once — use :func:`load_dependencies` to check first if idempotency is needed. Security -------- Every ID is validated as a UUID4 before any file path is constructed. The write is atomic (``write_text_atomic`` uses mkstemp → fsync → rename), ensuring no partial state is visible to concurrent readers. """ _validate_id(reservation_id) for dep_id in depends_on: _validate_id(dep_id) # Deduplicate while preserving order. unique_deps: list[str] = list(dict.fromkeys(depends_on)) if reservation_id in unique_deps: raise ValueError( f"reservation_id {reservation_id!r} cannot depend on itself" ) if len(unique_deps) > _MAX_DEPS: raise ValueError( f"too many dependencies: {len(unique_deps)} > {_MAX_DEPS}" ) ensure_dag_dirs(root) # ── Cycle check ─────────────────────────────────────────────────────────── # Build the current on-disk graph and tentatively add the new edges before # writing anything. If detect_cycle reports a cycle, reject the write. existing_deps = load_all_dependencies(root) graph = build_graph(existing_deps) if reservation_id not in graph: graph[reservation_id] = set() for dep_id in unique_deps: graph[reservation_id].add(dep_id) if dep_id not in graph: graph[dep_id] = set() cycle = detect_cycle(graph) if cycle is not None: raise ValueError( f"adding dependency would create a cycle: {' → '.join(cycle)}" ) # ── Write-once guard ────────────────────────────────────────────────────── dep_path = _dependencies_dir(root) / f"{reservation_id}.json" if dep_path.exists(): raise FileExistsError( f"dependency record for {reservation_id!r} already exists; " "dependency declarations are write-once" ) now = _now_utc() record = DependencyRecord( reservation_id=reservation_id, depends_on=unique_deps, created_at=now, ) write_text_atomic(dep_path, json.dumps(record.to_dict(), indent=2) + "\n") logger.debug( "Added %d dependency edge(s) for reservation %s", len(unique_deps), reservation_id[:8], ) return record # ── High-level convenience helper ───────────────────────────────────────────── def load_dag(root: pathlib.Path) -> AdjacencyMap: """Load all dependency records and return the full adjacency graph. Convenience wrapper around :func:`load_all_dependencies` + :func:`build_graph`. Returns an empty dict when no dependency records exist yet. """ return build_graph(load_all_dependencies(root))