feat(#89): reactive consumer (stage 1) — dependents reader + staleness predicates

src/tallyman_xorq/dependents.py

@@ -0,0 +1,108 @@
+"""Read the cross-entry dependency DAG recorded in each manifest (#84/#89).
+
+``manifest.parents`` records every entry's resolved ``from_catalog`` edges
+(``{hash, ref, follow}``) at build time, but nothing reads them: #76 deleted
+``lineage.py`` (``catalog_parents`` / ``catalog_dag``) along with the column-
+lineage views that were its only consumer. The reactive consumer needs the DAG
+walk back — not the column lineage — to find the dependents of a changed entry.
+This is that thin reader: forward edges (a child's parents), the reverse index
+(a parent's children), and the topologically-ordered descendant cone a recalc
+rebuilds in dependency order.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+
+from tallyman_core.manifest import ParentRef, read_manifest
+from tallyman_core.paths import entry_dir
+from tallyman_xorq.build import list_entries
+
+
+class DependencyCycleError(ValueError):
+    """The recorded parent edges form a cycle. Build guards against a recipe
+    reading its own alias (io.py), but the reader must not assume acyclicity."""
+
+
+def parents_of(project: str, content_hash: str) -> list[ParentRef]:
+    """The recorded parent edges of one entry (``[]`` for a root)."""
+    return list(read_manifest(entry_dir(project, content_hash)).parents or [])
+
+
+def build_dag(project: str) -> dict[str, list[ParentRef]]:
+    """Forward edges for every live entry: ``{child_hash: [ParentRef, ...]}``.
+
+    Enumerates via ``build.list_entries`` (every on-disk ``manifest.json``), not
+    ``read_tallyman_state``, so a freshly built but not-yet-checkpointed entry is
+    still seen.
+    """
+    dag: dict[str, list[ParentRef]] = {}
+    for entry in list_entries(project):
+        parents = entry.get("parents") or []
+        dag[entry["content_hash"]] = [ParentRef.model_validate(p) for p in parents]
+    return dag
+
+
+def dependents_index(project: str) -> dict[str, set[str]]:
+    """The reverse index: ``{parent_hash: {child_hash, ...}}``.
+
+    The recalc walk needs this to go from a changed entry to everything that
+    depends on it; it does not exist anywhere else.
+    """
+    return _children_map(build_dag(project))
+
+
+def descendant_cone(project: str, root_hashes: list[str]) -> list[str]:
+    """The roots and all their transitive dependents, parents before children.
+
+    Topologically ordered so a recalc rebuilds in dependency order. Raises
+    ``DependencyCycleError`` on a cycle rather than looping.
+    """
+    dag = build_dag(project)
+    children = _children_map(dag)
+    return _order_cone(root_hashes, dag, children)
+
+
+def _children_map(dag: dict[str, list[ParentRef]]) -> dict[str, set[str]]:
+    children: dict[str, set[str]] = {}
+    for child, parents in dag.items():
+        for parent in parents:
+            children.setdefault(parent.hash, set()).add(child)
+    return children
+
+
+def _order_cone(
+    roots: list[str],
+    dag: dict[str, list[ParentRef]],
+    children: dict[str, set[str]],
+) -> list[str]:
+    # 1. Collect the cone: every node reachable from a root via child edges,
+    #    plus the roots themselves.
+    cone: set[str] = set()
+    stack = list(roots)
+    while stack:
+        node = stack.pop()
+        if node in cone:
+            continue
+        cone.add(node)
+        stack.extend(children.get(node, ()))
+
+    # 2. Topologically sort the induced subgraph (Kahn). In-degree counts only
+    #    intra-cone parents, so a node whose stable parents sit outside the cone
+    #    is a source within it. Sorting keeps the order deterministic.
+    indegree = {node: sum(1 for parent in dag.get(node, []) if parent.hash in cone) for node in cone}
+    queue = deque(sorted(node for node in cone if indegree[node] == 0))
+    order: list[str] = []
+    while queue:
+        node = queue.popleft()
+        order.append(node)
+        for child in sorted(children.get(node, ())):
+            if child not in cone:
+                continue
+            indegree[child] -= 1
+            if indegree[child] == 0:
+                queue.append(child)
+
+    if len(order) != len(cone):
+        raise DependencyCycleError(f"dependency cycle among entries: {sorted(cone - set(order))}")
+    return order

src/tallyman_xorq/staleness.py

@@ -0,0 +1,131 @@
+"""Decide whether a catalog entry is stale relative to its recorded inputs (#89).
+
+Everything the consumer needs is already captured at build: ``manifest.parents``
+(the resolved ``from_catalog`` edges and their read-intent) and
+``manifest.sources`` (each source's content digest). Nothing reads it. This is
+the read-only half — no recompute — of the reactive consumer: compare the
+recorded inputs against the current world and report what moved.
+
+Two independent axes, both computable from already-recorded data:
+
+1. **Upstream advanced.** A ``follow=True`` parent (an alias argument) is stale
+   when its alias head now resolves to a different hash than the one recorded at
+   build. A ``follow=False`` parent (a literal-hash pin) is never stale here — it
+   asked for that exact revision.
+2. **Source drifted.** A recorded ``(rel_path, digest)`` is stale when the file
+   on disk no longer digests to the recorded value. Under ``off`` identity mode
+   the manifest records no digests, so this axis is reported ``unknown`` — never
+   silently ``fresh``.
+
+``result_digest`` is deliberately *not* a staleness input: a recompute-differs
+entry is *nondeterministic*, not stale, and recompute cannot make it fresh — that
+is the #83/#121 path, surfaced by the recompute action, not this scan.
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+
+from tallyman_core import aliases
+from tallyman_core.manifest import read_manifest
+from tallyman_core.paths import entry_dir
+from tallyman_xorq import source_identity
+from tallyman_xorq.build import list_entries
+from tallyman_xorq.dependents import descendant_cone
+from tallyman_xorq.io import ProjectDataNotFound, project_path
+
+# digest_for memoizes on (mtime_ns, size, inode); a same-stat in-place content
+# swap would otherwise serve the cached digest and the source axis would
+# false-negative. Force a faithful read for the staleness check (the env var is
+# the documented bypass, source_identity.py).
+_REHASH_ENV = "TALLYMAN_SOURCE_REHASH"
+
+
+@dataclass(frozen=True)
+class StaleReason:
+    axis: str  # "alias" | "source"
+    ref: str  # the alias name or the source's rel_path
+    was: str  # the value recorded at build (parent hash / source digest)
+    now: str | None  # the current value (alias head / current digest)
+
+
+@dataclass
+class StaleVerdict:
+    content_hash: str
+    stale: bool  # directly stale: this entry's own recorded inputs moved
+    reasons: list[StaleReason] = field(default_factory=list)
+    unknown_axes: list[str] = field(default_factory=list)  # axes that can't be evaluated
+    transitively_stale: bool = False  # set by scan(): a directly-stale ancestor exists
+
+
+@contextmanager
+def _force_source_rehash():
+    prev = os.environ.get(_REHASH_ENV)
+    os.environ[_REHASH_ENV] = "1"
+    try:
+        yield
+    finally:
+        if prev is None:
+            os.environ.pop(_REHASH_ENV, None)
+        else:
+            os.environ[_REHASH_ENV] = prev
+
+
+def entry_staleness(project: str, content_hash: str) -> StaleVerdict:
+    """Whether *content_hash*'s own recorded inputs have moved (read-only)."""
+    manifest = read_manifest(entry_dir(project, content_hash))
+    reasons: list[StaleReason] = []
+    unknown: list[str] = []
+
+    # Axis 1: a followed alias advanced past the recorded head.
+    for parent in manifest.parents or []:
+        if not parent.follow:
+            continue  # a hash pin is never stale on this axis
+        head = aliases.get_alias(project, parent.ref)
+        if head is None:
+            unknown.append(f"alias:{parent.ref}")  # alias gone — deletion is out of scope here
+        elif head != parent.hash:
+            reasons.append(StaleReason(axis="alias", ref=parent.ref, was=parent.hash, now=head))
+
+    # Axis 2: a recorded source drifted on disk.
+    if manifest.sources is None:
+        unknown.append("source")  # built under mode=off; the axis is unavailable
+    else:
+        with _force_source_rehash():
+            for rel_path, recorded in manifest.sources.items():
+                try:
+                    current = source_identity.digest_for(project, project_path(rel_path, project))
+                except ProjectDataNotFound:
+                    unknown.append(f"source:{rel_path}")  # source file removed
+                    continue
+                if current != recorded:
+                    reasons.append(StaleReason(axis="source", ref=rel_path, was=recorded, now=current))
+
+    return StaleVerdict(
+        content_hash=content_hash,
+        stale=bool(reasons),
+        reasons=reasons,
+        unknown_axes=unknown,
+    )
+
+
+def scan(project: str) -> dict[str, StaleVerdict]:
+    """Staleness for every live entry, tagging direct vs transitive staleness.
+
+    A **directly** stale entry has its own recorded input moved. A
+    **transitively** stale entry is a (not directly stale) descendant of one in
+    the dependency cone. The recalc surfaces use this to flag both while
+    distinguishing the entry the user must act on from the ones a cascade carries.
+    """
+    verdicts = {
+        entry["content_hash"]: entry_staleness(project, entry["content_hash"]) for entry in list_entries(project)
+    }
+    directly_stale = [h for h, v in verdicts.items() if v.stale]
+    if directly_stale:
+        for content_hash in descendant_cone(project, directly_stale):
+            verdict = verdicts.get(content_hash)
+            if verdict is not None and not verdict.stale:
+                verdict.transitively_stale = True
+    return verdicts

tests/test_dependents.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import pytest
+
+from tallyman_core.manifest import Manifest, ParentRef, write_manifest
+from tallyman_core.paths import entry_dir
+from tallyman_xorq.dependents import (
+    DependencyCycleError,
+    build_dag,
+    dependents_index,
+    descendant_cone,
+    parents_of,
+)
+
+
+def _write_entry(project: str, content_hash: str, parents=None) -> str:
+    """Write a minimal entry (manifest.json only) with the given parent edges.
+
+    The dependents reader walks ``manifest.parents`` and never executes a
+    recipe, so a hand-written manifest is a faithful and fast graph fixture —
+    no xorq build required. ``parents`` is a list of ``(hash, ref, follow)``.
+    """
+    d = entry_dir(project, content_hash)
+    d.mkdir(parents=True, exist_ok=True)
+    refs = [ParentRef(hash=h, ref=r, follow=f) for (h, r, f) in (parents or [])]
+    write_manifest(d, Manifest(content_hash=content_hash, project=project, parents=refs or None))
+    return content_hash
+
+
+def test_parents_of_root_is_empty(project):
+    _write_entry(project, "root")
+    assert parents_of(project, "root") == []
+
+
+def test_parents_of_returns_recorded_edges(project):
+    _write_entry(project, "root")
+    _write_entry(project, "child", [("root", "root", True)])
+    edges = parents_of(project, "child")
+    assert [(e.hash, e.ref, e.follow) for e in edges] == [("root", "root", True)]
+
+
+def test_build_dag_maps_every_live_entry(project):
+    _write_entry(project, "a")
+    _write_entry(project, "b", [("a", "a", True)])
+    dag = build_dag(project)
+    assert set(dag) == {"a", "b"}
+    assert dag["a"] == []
+    assert [e.hash for e in dag["b"]] == ["a"]
+
+
+def test_dependents_index_inverts_the_dag(project):
+    _write_entry(project, "a")
+    _write_entry(project, "b", [("a", "a", True)])
+    _write_entry(project, "c", [("a", "a", True)])
+    idx = dependents_index(project)
+    assert idx == {"a": {"b", "c"}}
+
+
+def test_descendant_cone_three_level_chain_is_topologically_ordered(project):
+    _write_entry(project, "a")
+    _write_entry(project, "b", [("a", "a", True)])
+    _write_entry(project, "c", [("b", "b", True)])
+    assert descendant_cone(project, ["a"]) == ["a", "b", "c"]
+
+
+def test_descendant_cone_diamond_orders_parents_before_children(project):
+    _write_entry(project, "a")
+    _write_entry(project, "b", [("a", "a", True)])
+    _write_entry(project, "c", [("a", "a", True)])
+    _write_entry(project, "d", [("b", "b", True), ("c", "c", True)])
+    cone = descendant_cone(project, ["a"])
+    assert set(cone) == {"a", "b", "c", "d"}
+    assert cone[0] == "a"
+    assert cone[-1] == "d"
+    assert cone.index("b") < cone.index("d")
+    assert cone.index("c") < cone.index("d")
+
+
+def test_descendant_cone_includes_both_pinned_and_followed_children(project):
+    _write_entry(project, "a")
+    _write_entry(project, "followed", [("a", "a", True)])
+    _write_entry(project, "pinned", [("a", "a", False)])
+    cone = descendant_cone(project, ["a"])
+    assert set(cone) == {"a", "followed", "pinned"}
+
+
+def test_descendant_cone_detects_a_cycle_instead_of_hanging(project):
+    _write_entry(project, "a", [("b", "b", True)])
+    _write_entry(project, "b", [("a", "a", True)])
+    with pytest.raises(DependencyCycleError):
+        descendant_cone(project, ["a"])
+
+
+def test_descendant_cone_of_a_leaf_is_just_the_leaf(project):
+    _write_entry(project, "solo")
+    assert descendant_cone(project, ["solo"]) == ["solo"]