From 4b83a7da2ce69ac973099446c489b627465af4b2 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 18 Jun 2026 15:22:21 -0400 Subject: [PATCH 1/2] =?UTF-8?q?test(#89):=20reactive=20consumer=20?= =?UTF-8?q?=E2=80=94=20dependents=20reader=20+=20staleness=20predicates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Failing-first (TDD): tests/test_dependents.py and tests/test_staleness.py exercise the read-only half of the reactive consumer (epic #89), which does not exist yet — both error on import of tallyman_xorq.dependents / tallyman_xorq.staleness. dependents: parents_of / build_dag / dependents_index / descendant_cone over manifest.parents, covered with synthetic manifests (chain, diamond, mixed follow/pin, cycle-raises, leaf). staleness: entry_staleness's two axes (followed-alias-advanced, source-drifted) and scan's direct-vs-transitive tagging, covered with real catalog_create / catalog_revise builds — alias advance marks a follow child stale, a hash pin does not, an in-place source edit marks the entry stale under cas, off mode reports the source axis unknown (not fresh), and a source edit under a chain root tags descendants transitively stale. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_dependents.py | 96 ++++++++++++++++++++++++++ tests/test_staleness.py | 142 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 238 insertions(+) create mode 100644 tests/test_dependents.py create mode 100644 tests/test_staleness.py diff --git a/tests/test_dependents.py b/tests/test_dependents.py new file mode 100644 index 0000000..cd07501 --- /dev/null +++ b/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"] diff --git a/tests/test_staleness.py b/tests/test_staleness.py new file mode 100644 index 0000000..4a08be5 --- /dev/null +++ b/tests/test_staleness.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +from tallyman_cli.fixtures import write_shoe_orders +from tallyman_core import data_dir +from tallyman_core.aliases import get_alias +from tallyman_mcp.server import catalog_create, catalog_revise +from tallyman_xorq.staleness import entry_staleness, scan + + +def _base_code(project: str) -> str: # from_project root over orders.parquet + return f""" +from tallyman_xorq.io import from_project +t = from_project("orders.parquet", project={project!r}) +expr = t.select("region", "price") +""" + + +def _base_code_v2(project: str) -> str: # a different graph → a new content hash + return f""" +from tallyman_xorq.io import from_project +t = from_project("orders.parquet", project={project!r}) +expr = t.select("region", "price").mutate(extra=1) +""" + + +def _child_code(parent: str) -> str: # from_catalog child (alias name or literal hash) + return f""" +from tallyman_xorq.io import from_catalog +t = from_catalog({parent!r}) +expr = t.mutate(doubled=t.price * 2) +""" + + +def _agg_child_code(parent: str) -> str: # expensive (Aggregate) child → bakes a snapshot + return f""" +from tallyman_xorq.io import from_catalog +t = from_catalog({parent!r}) +expr = t.group_by("region").aggregate(total=t.price.sum()) +""" + + +def _const_child_code(parent: str) -> str: # cheap child, schema-independent + return f""" +from tallyman_xorq.io import from_catalog +t = from_catalog({parent!r}) +expr = t.mutate(flag=1) +""" + + +def _hash(result: dict) -> str: + assert "hash" in result, result + return result["hash"] + + +def test_followed_alias_advance_marks_child_stale(project, orders_parquet, monkeypatch): + monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas") + base_v1 = _hash(catalog_create("base", _base_code(project))) + child_hash = _hash(catalog_create("child", _child_code("base"))) + + # The followed edge still matches the alias head — not stale yet. + assert entry_staleness(project, child_hash).stale is False + + _hash(catalog_revise("base", _base_code_v2(project))) + base_v2 = get_alias(project, "base") + assert base_v2 != base_v1 + + v = entry_staleness(project, child_hash) + assert v.stale is True + alias_reasons = [r for r in v.reasons if r.axis == "alias"] + assert len(alias_reasons) == 1 + assert (alias_reasons[0].ref, alias_reasons[0].was, alias_reasons[0].now) == ( + "base", + base_v1, + base_v2, + ) + + +def test_hash_pinned_parent_is_not_stale_when_alias_advances(project, orders_parquet, monkeypatch): + monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas") + base_v1 = _hash(catalog_create("base", _base_code(project))) + # A literal-hash argument is recorded follow=False (a pin), so the alias + # advancing must not make this child stale. + child_hash = _hash(catalog_create("child", _child_code(base_v1))) + + _hash(catalog_revise("base", _base_code_v2(project))) + assert get_alias(project, "base") != base_v1 + + v = entry_staleness(project, child_hash) + assert v.stale is False + assert [r for r in v.reasons if r.axis == "alias"] == [] + + +def test_source_edit_marks_entry_stale_under_cas(project, orders_parquet, monkeypatch): + monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas") + base_hash = _hash(catalog_create("base", _base_code(project))) + assert entry_staleness(project, base_hash).stale is False + + # Rewrite the source in place with different data (new bytes, new mtime). + write_shoe_orders(data_dir(project) / "orders.parquet", n_rows=200, seed=1) + + v = entry_staleness(project, base_hash) + assert v.stale is True + source_reasons = [r for r in v.reasons if r.axis == "source"] + assert len(source_reasons) == 1 + assert source_reasons[0].ref == "orders.parquet" + + +def test_off_mode_reports_source_axis_unknown_not_fresh(project, orders_parquet, monkeypatch): + monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "off") + base_hash = _hash(catalog_create("base", _base_code(project))) + + v = entry_staleness(project, base_hash) + assert v.stale is False + assert "source" in v.unknown_axes + + # off mode records no source digests, so it cannot see drift — unknown, not stale. + write_shoe_orders(data_dir(project) / "orders.parquet", n_rows=200, seed=1) + v2 = entry_staleness(project, base_hash) + assert v2.stale is False + assert "source" in v2.unknown_axes + + +def test_scan_distinguishes_direct_from_transitive_staleness(project, orders_parquet, monkeypatch): + monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas") + a_v1 = _hash(catalog_create("a", _base_code(project))) + # b is an expensive (Aggregate) intermediate: reconstructing it reads its + # baked snapshot rather than re-running its recipe, so its from_catalog("a") + # edge does not leak into c. (A cheap intermediate re-runs inline, so c would + # follow "a" directly and there would be no transitive-only node.) + b = _hash(catalog_create("b", _agg_child_code("a"))) + c = _hash(catalog_create("c", _const_child_code("b"))) + + # Advance alias "a". b follows "a" so it is DIRECTLY stale; c follows "b" + # (unchanged), so it is only TRANSITIVELY stale via its ancestor b. + _hash(catalog_revise("a", _base_code_v2(project))) + assert get_alias(project, "a") != a_v1 + + verdicts = scan(project) + assert verdicts[b].stale is True + assert verdicts[b].transitively_stale is False + assert verdicts[c].stale is False + assert verdicts[c].transitively_stale is True From b8eb0092cea314c8cdec8b90208809855ed2413a Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Thu, 18 Jun 2026 15:49:27 -0400 Subject: [PATCH 2/2] feat(#89): dependents DAG reader + staleness predicates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the read-only half of the reactive consumer (epic #89), turning tests/test_dependents.py and tests/test_staleness.py green. - tallyman_xorq/dependents.py — parents_of / build_dag / dependents_index / descendant_cone over manifest.parents. Rebuilds the DAG walk #76 deleted with lineage.py (the walk, not the column lineage), enumerating via build.list_entries so an uncheckpointed entry is still seen. descendant_cone is topologically ordered (Kahn) and raises DependencyCycleError rather than looping on a cycle. - tallyman_xorq/staleness.py — entry_staleness over two read-only axes (followed-alias-advanced; source-drifted, forced past digest_for's stat memo) and scan tagging direct vs transitive staleness across the dependency cone. off mode reports the source axis unknown, not fresh. result_digest is not a staleness input — recompute nondeterminism is the #83/#121 path, not this one. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/tallyman_xorq/dependents.py | 108 ++++++++++++++++++++++++++ src/tallyman_xorq/staleness.py | 131 ++++++++++++++++++++++++++++++++ 2 files changed, 239 insertions(+) create mode 100644 src/tallyman_xorq/dependents.py create mode 100644 src/tallyman_xorq/staleness.py diff --git a/src/tallyman_xorq/dependents.py b/src/tallyman_xorq/dependents.py new file mode 100644 index 0000000..5f0b388 --- /dev/null +++ b/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 diff --git a/src/tallyman_xorq/staleness.py b/src/tallyman_xorq/staleness.py new file mode 100644 index 0000000..164dd75 --- /dev/null +++ b/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