test(#89): reactive consumer — dependents reader + staleness predicates

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) <noreply@anthropic.com>

Author: paddymul

tests/test_dependents.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import pytest
+from tallyman_xorq.dependents import (
+    DependencyCycleError,
+    build_dag,
+    dependents_index,
+    descendant_cone,
+    parents_of,
+)
+
+from tallyman_core.manifest import Manifest, ParentRef, write_manifest
+from tallyman_core.paths import entry_dir
+
+
+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"]

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