"""Knowtation code-intelligence helpers backing ``muse code deps`` / ``impact``. This module sits between the CLI commands (``muse code deps`` and ``muse code impact``) and the link index (Phase 3.1). It produces JSON-shaped results that mirror the structure used by the Code domain so the two commands can be invoked uniformly regardless of the active domain. Public API ---------- - :func:`note_deps` — direct forward / reverse note-link queries for ``deps``. - :func:`note_impact` — transitive blast-radius BFS for ``impact``. - :func:`describe_link` — JSON-friendly representation of a single edge. Design notes ------------ The graph backing both ``deps`` and ``impact`` for vaults is the :class:`~muse.plugins.knowtation.link_index.LinkIndex`. For ``deps``: - **Forward (deps)** = outgoing links from a note (the note's "imports"). - **Reverse (deps --reverse)** = incoming links to a note (its "importers"). For ``impact``: - **Reverse BFS** = transitive set of notes whose changes propagate **to** the target (notes that link to me, plus notes that link to those, …). - **Forward BFS** (``--forward``) = transitive notes the target links to. By default, broken links are excluded from both queries. Pass ``include_broken=True`` to opt in. """ from __future__ import annotations import logging import pathlib from collections import deque from typing import Any from muse.plugins.knowtation.link_index import ( LinkIndex, ResolvedLink, build_link_index, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Serialisation helpers # --------------------------------------------------------------------------- def describe_link(link: ResolvedLink) -> dict[str, Any]: """Return a JSON-friendly representation of *link*. Args: link: A :class:`ResolvedLink` from the link index. Returns: Dict with keys: ``source``, ``target``, ``raw``, ``kind``, ``fragment``, ``broken``, ``alias``. """ return { "source": link.source, "target": link.target, "raw": link.raw, "kind": link.kind, "fragment": link.fragment, "broken": link.broken, "alias": link.alias, } # --------------------------------------------------------------------------- # deps queries # --------------------------------------------------------------------------- def note_deps( index: LinkIndex, note_path: str, *, reverse: bool = False, include_broken: bool = False, kinds: tuple[str, ...] = ("wikilink", "md_link"), ) -> dict[str, Any]: """Return forward or reverse link results for a single note. Args: index: Pre-built :class:`LinkIndex` for the active vault. note_path: Vault-relative POSIX path of the note to query. reverse: When ``True``, return incoming links instead of outgoing. include_broken: When ``False`` (default), broken links are filtered out of the result. Only meaningful for forward queries. kinds: Edge kinds to include. Defaults to ``("wikilink", "md_link")``; pass e.g. ``("wikilink", "md_link", "entity")`` to include entity co-occurrence edges. Returns: Dict suitable for JSON output, with keys: - ``note``: the queried note path. - ``direction``: ``"outgoing"`` or ``"incoming"``. - ``count``: number of edges in ``links``. - ``links``: list of edge dicts (see :func:`describe_link`). """ if reverse: edges = index.incoming(note_path) direction = "incoming" else: edges = index.outgoing(note_path) direction = "outgoing" filtered: list[ResolvedLink] = [] for link in edges: if link.kind not in kinds: continue if link.broken and not include_broken: continue filtered.append(link) return { "note": note_path, "direction": direction, "count": len(filtered), "links": [describe_link(link) for link in filtered], } # --------------------------------------------------------------------------- # impact (BFS) # --------------------------------------------------------------------------- def note_impact( index: LinkIndex, note_path: str, *, forward: bool = False, depth: int = 0, include_broken: bool = False, include_attachments: bool = True, ) -> dict[str, Any]: """Compute the transitive blast radius for *note_path*. Args: index: :class:`LinkIndex` for the active vault. note_path: Vault-relative POSIX path of the seed note. forward: When ``True``, walk outgoing edges (note's dependencies). When ``False`` (default), walk incoming edges (notes that depend on this note). depth: Maximum BFS depth. ``0`` (default) means unlimited. Must be non-negative. include_broken: When ``True``, include broken edges in the walk. They contribute no new nodes, so this only affects the broken-link count in the result. include_attachments: When ``True`` (default), notes sharing a mist attachment ID are treated as entangled (Phase 3.5 extension placeholder; this flag is honoured by callers but the underlying join is added in 3.5). Returns: Dict suitable for JSON output, with keys: - ``note``: the seed note path. - ``direction``: ``"forward"`` (callees) or ``"reverse"`` (callers). - ``depth_limit``: the configured maximum depth (``0`` for unlimited). - ``by_depth``: dict mapping depth (str) → sorted list of note paths. - ``total``: total number of distinct notes in the blast radius (excluding the seed note itself). - ``include_attachments``: echoed parameter. Raises: ValueError: When *depth* is negative. """ if depth < 0: raise ValueError(f"note_impact: depth must be non-negative, got {depth}") by_depth: dict[int, set[str]] = {} visited: set[str] = {note_path} frontier: deque[tuple[str, int]] = deque([(note_path, 0)]) # Phase 3.5: precompute note → mist_ids for O(1) attachment lookups. note_to_mist: dict[str, set[str]] = {} if include_attachments: for mist_id, members in index.attachment_index.items(): for member in members: note_to_mist.setdefault(member, set()).add(mist_id) while frontier: current, d = frontier.popleft() if depth != 0 and d >= depth: continue if forward: edges = index.outgoing(current) else: edges = index.incoming(current) next_depth = d + 1 for link in edges: # Forward BFS: follow link.target (where the current note points). # Reverse BFS: follow link.source (which note links to current). if forward: neighbour = link.target if neighbour is None: if include_broken: continue # broken forward edge contributes no neighbour continue else: neighbour = link.source if not neighbour or neighbour in visited: continue visited.add(neighbour) by_depth.setdefault(next_depth, set()).add(neighbour) frontier.append((neighbour, next_depth)) # Phase 3.5: mist-attachment co-references add cross-edges at # next_depth. Notes that share any mist ID with `current` are # entangled and contribute to the blast radius regardless of forward # vs reverse direction. if include_attachments and current in note_to_mist: for mist_id in note_to_mist[current]: for neighbour in index.attachment_index.get(mist_id, ()): if neighbour == current or neighbour in visited: continue visited.add(neighbour) by_depth.setdefault(next_depth, set()).add(neighbour) frontier.append((neighbour, next_depth)) serialised: dict[str, list[str]] = { str(d): sorted(by_depth[d]) for d in sorted(by_depth) } total = sum(len(v) for v in by_depth.values()) return { "note": note_path, "direction": "forward" if forward else "reverse", "depth_limit": depth, "by_depth": serialised, "total": total, "include_attachments": include_attachments, } # --------------------------------------------------------------------------- # Convenience: build + query in one call (used by CLI when no cache exists) # --------------------------------------------------------------------------- def deps_for_note( root: pathlib.Path, note_path: str, *, reverse: bool = False, include_broken: bool = False, ) -> dict[str, Any]: """Build a link index and return :func:`note_deps` for *note_path*. Convenience wrapper for one-shot CLI invocations. For repeated queries against the same vault, build the :class:`LinkIndex` once and call :func:`note_deps` directly. Args: root: Vault root. note_path: Note to query. reverse: Reverse-edge query. include_broken: Include broken links. Returns: JSON-shaped dict (see :func:`note_deps`). """ index = build_link_index(root) return note_deps( index, note_path, reverse=reverse, include_broken=include_broken, ) def impact_for_note( root: pathlib.Path, note_path: str, *, forward: bool = False, depth: int = 0, include_attachments: bool = True, ) -> dict[str, Any]: """Build a link index and return :func:`note_impact` for *note_path*. Args: root: Vault root. note_path: Note to query. forward: Forward (callees) instead of reverse (callers). depth: BFS depth cap (``0`` = unlimited). include_attachments: Honour ``include_attachments`` in the result. Returns: JSON-shaped dict (see :func:`note_impact`). """ index = build_link_index(root) return note_impact( index, note_path, forward=forward, depth=depth, include_attachments=include_attachments, )