Skip to content

Commit b8eb009

Browse files
paddymulclaude
andcommitted
feat(#89): dependents DAG reader + staleness predicates
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) <noreply@anthropic.com>
1 parent 4b83a7d commit b8eb009

2 files changed

Lines changed: 239 additions & 0 deletions

File tree

src/tallyman_xorq/dependents.py

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
"""Read the cross-entry dependency DAG recorded in each manifest (#84/#89).
2+
3+
``manifest.parents`` records every entry's resolved ``from_catalog`` edges
4+
(``{hash, ref, follow}``) at build time, but nothing reads them: #76 deleted
5+
``lineage.py`` (``catalog_parents`` / ``catalog_dag``) along with the column-
6+
lineage views that were its only consumer. The reactive consumer needs the DAG
7+
walk back — not the column lineage — to find the dependents of a changed entry.
8+
This is that thin reader: forward edges (a child's parents), the reverse index
9+
(a parent's children), and the topologically-ordered descendant cone a recalc
10+
rebuilds in dependency order.
11+
"""
12+
13+
from __future__ import annotations
14+
15+
from collections import deque
16+
17+
from tallyman_core.manifest import ParentRef, read_manifest
18+
from tallyman_core.paths import entry_dir
19+
from tallyman_xorq.build import list_entries
20+
21+
22+
class DependencyCycleError(ValueError):
23+
"""The recorded parent edges form a cycle. Build guards against a recipe
24+
reading its own alias (io.py), but the reader must not assume acyclicity."""
25+
26+
27+
def parents_of(project: str, content_hash: str) -> list[ParentRef]:
28+
"""The recorded parent edges of one entry (``[]`` for a root)."""
29+
return list(read_manifest(entry_dir(project, content_hash)).parents or [])
30+
31+
32+
def build_dag(project: str) -> dict[str, list[ParentRef]]:
33+
"""Forward edges for every live entry: ``{child_hash: [ParentRef, ...]}``.
34+
35+
Enumerates via ``build.list_entries`` (every on-disk ``manifest.json``), not
36+
``read_tallyman_state``, so a freshly built but not-yet-checkpointed entry is
37+
still seen.
38+
"""
39+
dag: dict[str, list[ParentRef]] = {}
40+
for entry in list_entries(project):
41+
parents = entry.get("parents") or []
42+
dag[entry["content_hash"]] = [ParentRef.model_validate(p) for p in parents]
43+
return dag
44+
45+
46+
def dependents_index(project: str) -> dict[str, set[str]]:
47+
"""The reverse index: ``{parent_hash: {child_hash, ...}}``.
48+
49+
The recalc walk needs this to go from a changed entry to everything that
50+
depends on it; it does not exist anywhere else.
51+
"""
52+
return _children_map(build_dag(project))
53+
54+
55+
def descendant_cone(project: str, root_hashes: list[str]) -> list[str]:
56+
"""The roots and all their transitive dependents, parents before children.
57+
58+
Topologically ordered so a recalc rebuilds in dependency order. Raises
59+
``DependencyCycleError`` on a cycle rather than looping.
60+
"""
61+
dag = build_dag(project)
62+
children = _children_map(dag)
63+
return _order_cone(root_hashes, dag, children)
64+
65+
66+
def _children_map(dag: dict[str, list[ParentRef]]) -> dict[str, set[str]]:
67+
children: dict[str, set[str]] = {}
68+
for child, parents in dag.items():
69+
for parent in parents:
70+
children.setdefault(parent.hash, set()).add(child)
71+
return children
72+
73+
74+
def _order_cone(
75+
roots: list[str],
76+
dag: dict[str, list[ParentRef]],
77+
children: dict[str, set[str]],
78+
) -> list[str]:
79+
# 1. Collect the cone: every node reachable from a root via child edges,
80+
# plus the roots themselves.
81+
cone: set[str] = set()
82+
stack = list(roots)
83+
while stack:
84+
node = stack.pop()
85+
if node in cone:
86+
continue
87+
cone.add(node)
88+
stack.extend(children.get(node, ()))
89+
90+
# 2. Topologically sort the induced subgraph (Kahn). In-degree counts only
91+
# intra-cone parents, so a node whose stable parents sit outside the cone
92+
# is a source within it. Sorting keeps the order deterministic.
93+
indegree = {node: sum(1 for parent in dag.get(node, []) if parent.hash in cone) for node in cone}
94+
queue = deque(sorted(node for node in cone if indegree[node] == 0))
95+
order: list[str] = []
96+
while queue:
97+
node = queue.popleft()
98+
order.append(node)
99+
for child in sorted(children.get(node, ())):
100+
if child not in cone:
101+
continue
102+
indegree[child] -= 1
103+
if indegree[child] == 0:
104+
queue.append(child)
105+
106+
if len(order) != len(cone):
107+
raise DependencyCycleError(f"dependency cycle among entries: {sorted(cone - set(order))}")
108+
return order

src/tallyman_xorq/staleness.py

Lines changed: 131 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,131 @@
1+
"""Decide whether a catalog entry is stale relative to its recorded inputs (#89).
2+
3+
Everything the consumer needs is already captured at build: ``manifest.parents``
4+
(the resolved ``from_catalog`` edges and their read-intent) and
5+
``manifest.sources`` (each source's content digest). Nothing reads it. This is
6+
the read-only half — no recompute — of the reactive consumer: compare the
7+
recorded inputs against the current world and report what moved.
8+
9+
Two independent axes, both computable from already-recorded data:
10+
11+
1. **Upstream advanced.** A ``follow=True`` parent (an alias argument) is stale
12+
when its alias head now resolves to a different hash than the one recorded at
13+
build. A ``follow=False`` parent (a literal-hash pin) is never stale here — it
14+
asked for that exact revision.
15+
2. **Source drifted.** A recorded ``(rel_path, digest)`` is stale when the file
16+
on disk no longer digests to the recorded value. Under ``off`` identity mode
17+
the manifest records no digests, so this axis is reported ``unknown`` — never
18+
silently ``fresh``.
19+
20+
``result_digest`` is deliberately *not* a staleness input: a recompute-differs
21+
entry is *nondeterministic*, not stale, and recompute cannot make it fresh — that
22+
is the #83/#121 path, surfaced by the recompute action, not this scan.
23+
"""
24+
25+
from __future__ import annotations
26+
27+
import os
28+
from contextlib import contextmanager
29+
from dataclasses import dataclass, field
30+
31+
from tallyman_core import aliases
32+
from tallyman_core.manifest import read_manifest
33+
from tallyman_core.paths import entry_dir
34+
from tallyman_xorq import source_identity
35+
from tallyman_xorq.build import list_entries
36+
from tallyman_xorq.dependents import descendant_cone
37+
from tallyman_xorq.io import ProjectDataNotFound, project_path
38+
39+
# digest_for memoizes on (mtime_ns, size, inode); a same-stat in-place content
40+
# swap would otherwise serve the cached digest and the source axis would
41+
# false-negative. Force a faithful read for the staleness check (the env var is
42+
# the documented bypass, source_identity.py).
43+
_REHASH_ENV = "TALLYMAN_SOURCE_REHASH"
44+
45+
46+
@dataclass(frozen=True)
47+
class StaleReason:
48+
axis: str # "alias" | "source"
49+
ref: str # the alias name or the source's rel_path
50+
was: str # the value recorded at build (parent hash / source digest)
51+
now: str | None # the current value (alias head / current digest)
52+
53+
54+
@dataclass
55+
class StaleVerdict:
56+
content_hash: str
57+
stale: bool # directly stale: this entry's own recorded inputs moved
58+
reasons: list[StaleReason] = field(default_factory=list)
59+
unknown_axes: list[str] = field(default_factory=list) # axes that can't be evaluated
60+
transitively_stale: bool = False # set by scan(): a directly-stale ancestor exists
61+
62+
63+
@contextmanager
64+
def _force_source_rehash():
65+
prev = os.environ.get(_REHASH_ENV)
66+
os.environ[_REHASH_ENV] = "1"
67+
try:
68+
yield
69+
finally:
70+
if prev is None:
71+
os.environ.pop(_REHASH_ENV, None)
72+
else:
73+
os.environ[_REHASH_ENV] = prev
74+
75+
76+
def entry_staleness(project: str, content_hash: str) -> StaleVerdict:
77+
"""Whether *content_hash*'s own recorded inputs have moved (read-only)."""
78+
manifest = read_manifest(entry_dir(project, content_hash))
79+
reasons: list[StaleReason] = []
80+
unknown: list[str] = []
81+
82+
# Axis 1: a followed alias advanced past the recorded head.
83+
for parent in manifest.parents or []:
84+
if not parent.follow:
85+
continue # a hash pin is never stale on this axis
86+
head = aliases.get_alias(project, parent.ref)
87+
if head is None:
88+
unknown.append(f"alias:{parent.ref}") # alias gone — deletion is out of scope here
89+
elif head != parent.hash:
90+
reasons.append(StaleReason(axis="alias", ref=parent.ref, was=parent.hash, now=head))
91+
92+
# Axis 2: a recorded source drifted on disk.
93+
if manifest.sources is None:
94+
unknown.append("source") # built under mode=off; the axis is unavailable
95+
else:
96+
with _force_source_rehash():
97+
for rel_path, recorded in manifest.sources.items():
98+
try:
99+
current = source_identity.digest_for(project, project_path(rel_path, project))
100+
except ProjectDataNotFound:
101+
unknown.append(f"source:{rel_path}") # source file removed
102+
continue
103+
if current != recorded:
104+
reasons.append(StaleReason(axis="source", ref=rel_path, was=recorded, now=current))
105+
106+
return StaleVerdict(
107+
content_hash=content_hash,
108+
stale=bool(reasons),
109+
reasons=reasons,
110+
unknown_axes=unknown,
111+
)
112+
113+
114+
def scan(project: str) -> dict[str, StaleVerdict]:
115+
"""Staleness for every live entry, tagging direct vs transitive staleness.
116+
117+
A **directly** stale entry has its own recorded input moved. A
118+
**transitively** stale entry is a (not directly stale) descendant of one in
119+
the dependency cone. The recalc surfaces use this to flag both while
120+
distinguishing the entry the user must act on from the ones a cascade carries.
121+
"""
122+
verdicts = {
123+
entry["content_hash"]: entry_staleness(project, entry["content_hash"]) for entry in list_entries(project)
124+
}
125+
directly_stale = [h for h, v in verdicts.items() if v.stale]
126+
if directly_stale:
127+
for content_hash in descendant_cone(project, directly_stale):
128+
verdict = verdicts.get(content_hash)
129+
if verdict is not None and not verdict.stale:
130+
verdict.transitively_stale = True
131+
return verdicts

0 commit comments

Comments
 (0)