"""Canonical commit DAG traversal primitives for the Muse VCS. Every commit-graph walk in Muse — BFS ancestry, range exclusion, reachability, and merge-base search — is built on the three functions exported here. Inline BFS implementations scattered across commands and core modules should be replaced with calls to these primitives. Public API ---------- walk_dag Generic BFS/DFS DAG walker parameterised on node type T and an adjacency function. The foundational primitive: all other traversals are built on top of this. iter_ancestors BFS generator that yields :class:`~muse.core.store.CommitRecord` objects in visit order. Thin wrapper around :func:`walk_dag` with commit-specific adjacency logic. ancestor_ids Convenience wrapper around :func:`iter_ancestors` that returns a ``set[str]`` of commit IDs. Used for range exclusion (``A..B`` semantics) and ancestry-path filtering. find_merge_base Dual-frontier BFS that finds the Lowest Common Ancestor of two commits in O(distance_to_LCA) steps rather than O(total_history). Design notes ------------ * All traversals use :class:`collections.deque` for O(1) pops. ``list.pop(0)`` or ``list.insert(0, ...)`` would be O(n) per step. * walk_dag calls ``adjacency(node)`` before yielding the node so that wrappers like iter_ancestors can cache the result (avoids double reads). * Missing commits (``read_commit`` returns ``None``) are silently skipped and do not count against ``max_commits``. * The ``gc`` module intentionally reads raw bytes instead of using ``read_commit`` — this is a deliberate robustness choice (survives schema evolution) and is the only sanctioned exception to the "use graph.py" rule. """ import collections import pathlib from collections.abc import Callable, Iterable, Iterator from typing import AbstractSet, Hashable, Literal, TypeVar from muse.core.commits import ( CommitRecord, read_commit, ) T = TypeVar("T", bound=Hashable) # --------------------------------------------------------------------------- # walk_dag — generic DAG walker # --------------------------------------------------------------------------- def walk_dag( starts: "T | Iterable[T]", adjacency: "Callable[[T], Iterable[T]]", *, order: "Literal['bfs', 'dfs']" = "bfs", exclude: "AbstractSet[T]" = frozenset(), prune: "Callable[[T], bool] | None" = None, max_nodes: "int | None" = None, ) -> "Iterator[T]": """Generic BFS/DFS walker over any directed acyclic graph. Parameters ---------- starts: A single node or an iterable of start nodes. Bare strings are treated as single nodes; all other iterables are expanded. adjacency: Called with a node; returns its immediate neighbours. For commit DAGs, this reads parent IDs. The function is called *before* the node is yielded so wrappers can cache the result. order: ``"bfs"`` (breadth-first, default) or ``"dfs"`` (depth-first). Raises :class:`ValueError` for any other value. exclude: Nodes treated as already-visited. They and their subtrees are never yielded. The caller's set is never mutated. prune: Predicate called on each candidate node. When it returns ``True`` the node is *not* yielded and its subtree is not explored. max_nodes: Hard cap on the number of nodes yielded. ``None`` means no limit. ``0`` yields nothing. Yields ------ T Each node at most once, in BFS or DFS visit order. """ if order not in ("bfs", "dfs"): raise ValueError(f"order must be 'bfs' or 'dfs', got {order!r}") if isinstance(starts, str): start_list: list[T] = [starts] # type: ignore[list-item] elif hasattr(starts, "__iter__"): start_list = list(starts) # type: ignore[arg-type] else: start_list = [starts] seen: set[T] = set(exclude) yielded = 0 queue: collections.deque[T] = collections.deque(start_list) while queue: if max_nodes is not None and yielded >= max_nodes: return node = queue.popleft() if order == "bfs" else queue.pop() if node in seen: continue if prune is not None and prune(node): seen.add(node) continue seen.add(node) neighbours = list(adjacency(node)) if order == "bfs": for n in neighbours: if n not in seen: queue.append(n) else: for n in reversed(neighbours): if n not in seen: queue.append(n) yield node yielded += 1 # --------------------------------------------------------------------------- # iter_ancestors # --------------------------------------------------------------------------- def iter_ancestors( root: pathlib.Path, starts: "str | Iterable[str]", *, first_parent_only: bool = False, exclude: "AbstractSet[str]" = frozenset(), prune: "Callable[[str], bool] | None" = None, max_commits: "int | None" = None, ) -> "Iterator[CommitRecord]": """BFS over the commit DAG, yielding :class:`CommitRecord` in visit order. Thin wrapper around :func:`walk_dag` with commit-specific adjacency logic. This is the **canonical commit-graph walker** for all of Muse. All other BFS, ancestor-set, reachability, or range-exclusion computations should be built on top of this function. Parameters ---------- root: Repository root directory. starts: A single commit ID string or an iterable of commit ID strings. When more than one ID is given the BFS is seeded from all of them simultaneously (multi-source BFS). first_parent_only: When ``True``, follow only ``parent_commit_id`` (the first/linear parent) and skip ``parent2_commit_id``. Use this for producing linear first-parent history views. exclude: Commit IDs to treat as already-visited. Commits in this set — and all of their ancestors — are never yielded. Use this for range-exclusion (``A..B`` semantics): pass ``ancestor_ids(root, a)`` as *exclude* when walking from *b*. max_commits: Hard cap on the number of commits *yielded*. ``None`` means no limit. Missing commits that are skipped do not count against the cap. Useful for safety ceilings on very large repos. Yields ------ CommitRecord In BFS visit order (roughly newest-first for a linear chain). Each commit is yielded at most once even when reachable via multiple paths. Missing commits are silently skipped. Examples -------- Walk all ancestors of HEAD:: for commit in iter_ancestors(root, head_id): process(commit) Walk only the first-parent chain:: ids = [c.commit_id for c in iter_ancestors(root, head, first_parent_only=True)] Compute A..B range:: base_ids = ancestor_ids(root, a_id) for commit in iter_ancestors(root, b_id, exclude=base_ids): print(commit.message) Multi-source BFS from all branch tips:: tips = [read_ref(root, b) for b in all_branches(root)] for commit in iter_ancestors(root, tips): process(commit) """ cache: dict[str, CommitRecord] = {} def _adjacency(cid: str) -> list[str]: commit = read_commit(root, cid) if commit is None: return [] cache[cid] = commit parents: list[str] = [] p1 = commit.parent_commit_id p2 = commit.parent2_commit_id if p1: parents.append(p1) if not first_parent_only and p2: parents.append(p2) return parents yielded = 0 for cid in walk_dag(starts, _adjacency, exclude=exclude, prune=prune): if max_commits is not None and yielded >= max_commits: return commit = cache.get(cid) if commit is None: continue # missing commit — skip without counting against cap yield commit yielded += 1 # --------------------------------------------------------------------------- # ancestor_ids # --------------------------------------------------------------------------- def ancestor_ids( root: pathlib.Path, starts: str | Iterable[str], *, first_parent_only: bool = False, exclude: AbstractSet[str] = frozenset(), max_commits: int | None = None, ) -> set[str]: """Return the set of all commit IDs reachable from *starts*. A convenience wrapper around :func:`iter_ancestors` for callers that need a ``set[str]`` of IDs rather than a stream of records. Common uses ----------- * **Range exclusion** — ``ancestor_ids(root, base_id)`` as the ``exclude`` argument to a subsequent :func:`iter_ancestors` call implements ``base..tip`` semantics. * **Ancestry-path filtering** — check whether a commit lies on a direct path between two references. * **Reachability** — determine which commits are live (all IDs reachable from any branch tip). Parameters ---------- root, starts, first_parent_only, exclude, max_commits: Forwarded to :func:`iter_ancestors` unchanged. Returns ------- set[str] Commit ID strings (``sha256:…``-prefixed). """ return { c.commit_id for c in iter_ancestors( root, starts, first_parent_only=first_parent_only, exclude=exclude, max_commits=max_commits, ) } # --------------------------------------------------------------------------- # find_merge_base # --------------------------------------------------------------------------- def find_merge_base( root: pathlib.Path, commit_id_a: str, commit_id_b: str, *, max_ancestors: int | None = None, ) -> str | None: """Find the Lowest Common Ancestor (LCA) of two commits. Uses simultaneous bidirectional BFS — expanding both frontiers one step at a time and stopping as soon as any commit ID appears in both seen sets. This is O(distance_to_LCA) rather than O(total_history), making it efficient even when two commits share a long common ancestry. Parameters ---------- root: Repository root directory. commit_id_a, commit_id_b: The two commits to find the merge base for. max_ancestors: Hard cap on total commits visited across both frontiers combined. When exceeded, raises :class:`ValueError` so the caller can surface a useful error to the user. ``None`` means no limit. Returns ------- str | None The LCA commit ID, or ``None`` if the two commits share no common ancestor (disjoint histories). Raises ------ ValueError When *max_ancestors* is set and the combined ancestor graph exceeds the cap before a common ancestor is found. The message includes the cap value so callers can relay it in user-facing errors. """ if commit_id_a == commit_id_b: return commit_id_a seen_a: set[str] = {commit_id_a} seen_b: set[str] = {commit_id_b} frontier_a: collections.deque[str] = collections.deque([commit_id_a]) frontier_b: collections.deque[str] = collections.deque([commit_id_b]) total = 0 while frontier_a or frontier_b: if max_ancestors is not None and total >= max_ancestors: raise ValueError( f"max_ancestors={max_ancestors:,} exceeded during merge-base search — " "history too deep or DAG is malformed." ) if frontier_a: cid = frontier_a.popleft() total += 1 commit = read_commit(root, cid) if commit is not None: for parent in (commit.parent_commit_id, commit.parent2_commit_id): if parent is not None and parent not in seen_a: seen_a.add(parent) if parent in seen_b: return parent frontier_a.append(parent) if frontier_b: cid = frontier_b.popleft() total += 1 commit = read_commit(root, cid) if commit is not None: for parent in (commit.parent_commit_id, commit.parent2_commit_id): if parent is not None and parent not in seen_b: seen_b.add(parent) if parent in seen_a: return parent frontier_b.append(parent) return None