test: doc-driven edge-case tests for the reactive system and caching

src/tallyman_xorq/io.py

@@ -9,6 +9,8 @@
 
 from __future__ import annotations
 
+import contextlib
+import contextvars
 from pathlib import Path
 
 from tallyman_core import data_dir, entry_dir, get_alias, resolve_project
@@ -18,6 +20,30 @@ class ProjectDataNotFound(FileNotFoundError):
     pass
 
 
+# A replay-time freeze for alias-pinned parents (#160). ``pinned_expr_from_alias("p")``
+# stores the alias *name* in the recipe, so a naive replay re-resolves the name to p's
+# CURRENT head and the pin silently advances — the docs promise a pin "stays on that
+# exact revision" and is "never disturbed by recalc". During a recalc replay the walk
+# knows the revision the pin resolved to at build (``manifest.parents[i].hash``) and
+# sets this map ``{alias_name: frozen_hash}`` so ``pinned_expr_from_alias`` re-resolves
+# to the recorded revision instead of the live head. A pin by *hash* stores the hash
+# literally and never consults this map — it is already replay-durable.
+_PIN_FREEZE: contextvars.ContextVar[dict[str, str] | None] = contextvars.ContextVar("tallyman_pin_freeze", default=None)
+
+
+@contextlib.contextmanager
+def pin_freeze(mapping: dict[str, str]):
+    """Freeze alias-pinned parents to their recorded revisions for the duration of a
+    recalc replay (#160). ``mapping`` is ``{alias_name: content_hash}``; while it is
+    active, ``pinned_expr_from_alias(name)`` resolves ``name`` to ``content_hash``
+    rather than the alias's current head."""
+    token = _PIN_FREEZE.set(mapping)
+    try:
+        yield
+    finally:
+        _PIN_FREEZE.reset(token)
+
+
 def project_path(rel_path: str, project: str | None = None, must_exist: bool = True) -> Path:
     """Resolve `rel_path` against the project's data dir. Raises if absent.
 
@@ -99,10 +125,19 @@ def read_project_file(rel_path: str, project: str | None = None):
 # Covers the types the MCP documents for tallyman_read_csv; an unmapped type
 # raises rather than silently inferring, so a schema mistake fails loudly.
 _IBIS_TO_POLARS = {
-    "int8": "Int8", "int16": "Int16", "int32": "Int32", "int64": "Int64",
-    "uint8": "UInt8", "uint16": "UInt16", "uint32": "UInt32", "uint64": "UInt64",
-    "float32": "Float32", "float64": "Float64",
-    "string": "String", "bool": "Boolean", "boolean": "Boolean",
+    "int8": "Int8",
+    "int16": "Int16",
+    "int32": "Int32",
+    "int64": "Int64",
+    "uint8": "UInt8",
+    "uint16": "UInt16",
+    "uint32": "UInt32",
+    "uint64": "UInt64",
+    "float32": "Float32",
+    "float64": "Float64",
+    "string": "String",
+    "bool": "Boolean",
+    "boolean": "Boolean",
     "date": "Date",
 }
 
@@ -273,7 +308,7 @@ def _normalize_schema(spec, header: list[str], *, reserved: tuple[str, ...] = ()
                 f"{len(speccable)} {list(speccable)}.{reserved_hint}"
             )
         if rest_dtype is None and len(cells) != len(speccable):
-            uncovered = list(speccable[len(cells):])
+            uncovered = list(speccable[len(cells) :])
             raise ValueError(
                 f"tallyman_read_csv: schema is not total — it leaves column(s) {uncovered} unspecified. "
                 'Name every column, or close the spec with ("&rest", "infer").'
@@ -289,7 +324,7 @@ def _normalize_schema(spec, header: list[str], *, reserved: tuple[str, ...] = ()
                 inferred = True
             else:
                 overrides[orig] = pl_dt
-        for orig in speccable[len(cells):]:  # the &rest tail keeps its header name
+        for orig in speccable[len(cells) :]:  # the &rest tail keeps its header name
             out_names.append(orig)
             pl_dt = _resolve_spec_dtype(rest_dtype)
             if pl_dt is None:
@@ -717,7 +752,16 @@ def pinned_expr_from_alias(alias_or_hash: str, project: str | None = None):
 
     proj = resolve_project(project)
     is_hash = entry_dir(proj, alias_or_hash).exists()
-    content_hash = alias_or_hash if is_hash else get_alias(proj, alias_or_hash)
+    if is_hash:
+        content_hash = alias_or_hash  # a literal hash pin is already replay-durable
+    else:
+        # An alias-name pin: during a recalc replay, resolve to the revision recorded
+        # at build (the freeze map) so the pin is not silently advanced by the replay
+        # (#160); otherwise resolve the alias's current head as usual.
+        frozen = _PIN_FREEZE.get()
+        content_hash = frozen.get(alias_or_hash) if frozen else None
+        if content_hash is None:
+            content_hash = get_alias(proj, alias_or_hash)
     if content_hash is None or not entry_dir(proj, content_hash).exists():
         raise ProjectDataNotFound(f"catalog entry {alias_or_hash!r} not found in project {proj!r}")
     content_hash = _resolve_noncyclic_hash(proj, alias_or_hash, content_hash)

src/tallyman_xorq/recalc.py

@@ -46,7 +46,8 @@
 from tallyman_core.errors import errors_by_hash, record_error
 from tallyman_core.paths import entry_dir, entry_manifest_path, project_dir
 from tallyman_xorq.build import BuildError, build_and_persist
-from tallyman_xorq.dependents import DependencyCycleError, build_dag, descendant_cone
+from tallyman_xorq.dependents import DependencyCycleError, build_dag, descendant_cone, parents_of
+from tallyman_xorq.io import pin_freeze
 from tallyman_xorq.portable import PLACEHOLDER
 from tallyman_xorq.staleness import StaleReason, StaleVerdict, scan
 
@@ -102,6 +103,18 @@ def _recipe_code(project: str, content_hash: str) -> str:
     return code.replace(PLACEHOLDER, str(project_dir(project)))
 
 
+def _pinned_alias_freeze(project: str, content_hash: str) -> dict[str, str]:
+    """``{alias_name: recorded_hash}`` for this entry's alias-pinned parents (#160).
+
+    A ``follow=False`` parent whose ``ref`` is an alias *name* (not the pinned hash
+    itself) was resolved to ``ref``'s head at build; on replay the recipe re-runs
+    ``pinned_expr_from_alias(ref)`` and would re-resolve the name to the alias's *new*
+    head, silently advancing the pin. Freezing ``ref -> hash`` keeps the pin on its
+    recorded revision. A hash pin (``ref == hash``) stores the hash literally in the
+    recipe and is already replay-durable, so it needs no freeze entry."""
+    return {p.ref: p.hash for p in parents_of(project, content_hash) if not p.follow and p.ref != p.hash}
+
+
 def _aliases_heading(project: str, content_hash: str) -> list[str]:
     """Every alias whose *current* head is ``content_hash`` (usually one, never
     more for a freshly-built entry; a doubly-aliased head returns both)."""
@@ -118,23 +131,25 @@ def _preview_action(content_hash: str, verdict, cone: set[str], dag: dict) -> st
     return "unchanged"  # pinned to an out-of-cone parent, or a non-stale root
 
 
-def _live_cone(cone: list[str], roots: list[str], verdicts: dict[str, StaleVerdict]) -> list[str]:
-    """The cone with dragged-in husks removed (#154). ``descendant_cone`` re-expands
-    from a stale head through *every* recorded parent edge, so a superseded husk that
-    follows that head (via a since-advanced alias) is pulled back in even though the
-    head-gate dropped it from the root set. Replaying such a husk re-resolves its
-    followed alias to the advanced head and manufactures a junk entry that heads no
-    alias and re-points nothing — the f2dd/cb24 regression. Drop every non-head cone
-    member; an explicitly-named root is exempt (a user-supplied root is a deliberate
-    request, not scan-driven drag-in — the revise/recalc defaults never pass one).
+def _live_cone(cone: list[str], verdicts: dict[str, StaleVerdict]) -> list[str]:
+    """The cone with husks removed (#154). ``descendant_cone`` re-expands from a stale
+    head through *every* recorded parent edge, so a superseded husk that follows that
+    head (via a since-advanced alias) is pulled back in. Replaying such a husk
+    re-resolves its followed alias to the advanced head and manufactures a junk entry
+    that heads no alias and re-points nothing — the f2dd/cb24 regression.
+
+    Drop every husk (``live=False``), whether it was dragged in by the cone expansion
+    OR named explicitly as a root: mcp-server.md promises "roots naming no live entry
+    are silently dropped", and the scan's own ``live=False`` verdict for the husk is
+    the authority on that. An explicit husk root replayed under source drift mints the
+    same junk sibling as the scan-driven path, so the gate must cover both.
 
     Safe because a *live* entry can depend on a husk only through a ``follow=False``
     hash pin (nothing can follow a headless husk by name), and a hash pin replays to
     the same parent whether or not the husk itself was rebuilt — so nothing downstream
     is stranded by skipping it. Order is preserved, so the topological walk stays valid.
     """
-    roots_set = set(roots)
-    return [h for h in cone if h in roots_set or verdicts[h].live]
+    return [h for h in cone if verdicts[h].live]
 
 
 def recalc(project: str, roots: list[str], *, dry_run: bool = False) -> RecalcReport:
@@ -165,7 +180,7 @@ def recalc(project: str, roots: list[str], *, dry_run: bool = False) -> RecalcRe
         # the project), then index into it for the cone. Keeps the staleness
         # semantics in one tested place rather than re-deriving them here.
         verdicts = scan(project)
-        cone = _live_cone(cone, roots, verdicts)  # drop dragged-in husks so preview matches the real walk (#154)
+        cone = _live_cone(cone, verdicts)  # drop husks so preview matches the real walk (#154)
         cone_set = set(cone)
         dag = build_dag(project)
         entries = [
@@ -222,7 +237,7 @@ def _replay_cone(
 
     if verdicts is None:
         verdicts = scan(project)
-    cone = _live_cone(cone, roots, verdicts)  # a husk dragged into a stale head's cone must not replay into junk (#154)
+    cone = _live_cone(cone, verdicts)  # a husk (dragged-in or explicitly named) must not replay into junk (#154)
     remap: dict[str, str] = {}
     entries: list[RecalcEntry] = []
     status = "ok"
@@ -246,8 +261,13 @@ def _replay_cone(
         # Resolve the heading alias(es) BEFORE re-pointing anything for this entry,
         # so a yet-to-be-processed downstream alias is read at its pre-recalc head.
         heading = _aliases_heading(project, old_hash)
+        # Freeze alias-pinned parents to the revision they resolved to at build, so a
+        # replay forced by a *tracked* parent advancing does not silently re-resolve
+        # `pinned_expr_from_alias("p")` to p's new head (#160).
+        freeze = _pinned_alias_freeze(project, old_hash)
         try:
-            result = build_and_persist(project, _recipe_code(project, old_hash))
+            with pin_freeze(freeze):
+                result = build_and_persist(project, _recipe_code(project, old_hash))
         except BuildError as exc:
             # Persist the failure (outside the catalog repo, so it survives
             # reset_to) keyed by the failed hash — the orphan-stale tie-back.

src/tallyman_xorq/staleness.py

@@ -118,29 +118,39 @@ def entry_staleness(project: str, content_hash: str) -> StaleVerdict:
 def scan(project: str) -> dict[str, StaleVerdict]:
     """Actionable staleness for every live entry, tagging direct vs transitive.
 
-    An entry is **directly** stale only when it is a *current alias head* whose own
-    recorded input moved (#154). A superseded historical version pins alias heads
-    that have since advanced, so its inputs read as moved forever — but it heads no
-    alias, so recomputing it re-points nothing (and, for a recipe shape no current
-    head uses, manufactures a junk entry), and it is not an invariant break to flag.
-    Such a non-head is marked ``live=False`` and forced ``stale=False`` here (its
-    ``reasons`` are kept for forensics). A **transitively** stale entry is a (not
-    directly stale) descendant of a directly-stale head in the dependency cone.
+    An entry is **live** unless it is a *superseded husk* — a version that appears
+    in some alias's history but is no longer that alias's head (#154). A husk pins
+    alias heads that have since advanced, so its inputs read as moved forever, yet
+    recomputing it re-points nothing (and, for a recipe shape no current head uses,
+    manufactures a junk entry). Such a husk is marked ``live=False`` and forced
+    ``stale=False`` here (its ``reasons`` are kept for forensics).
+
+    A never-aliased *scratch* entry (a ``catalog_run`` follower that appears in no
+    alias history) is **not** a husk: nothing has superseded it, so it stays live
+    and goes actionably stale when a parent it follows advances — it rebuilds like
+    any follower, minus the alias bookkeeping (reactive-recalc.md's scratch-rebuild
+    semantics). The gate is "superseded", not "is a head".
+
+    A **transitively** stale entry is a (not directly stale) descendant of a
+    directly-stale head in the dependency cone.
 
     ``entry_staleness`` stays the raw per-entry primitive — did *this* entry's inputs
-    move, regardless of headship; this function layers the head-reachability gate
-    that makes ``stale`` mean *actionable*. The verdict dict still keys **every**
-    entry (heads and husks alike), because the recalc replay indexes cone members
-    that are not themselves heads (a hash-pinned child).
+    move, regardless of headship; this function layers the husk gate that makes
+    ``stale`` mean *actionable*. The verdict dict still keys **every** entry (live
+    entries and husks alike), because the recalc replay indexes cone members that
+    are not themselves heads (a hash-pinned child).
     """
     heads = set(aliases.load_aliases(project).values())  # current alias heads, read once
+    # A husk is a hash that appears in some alias history but is no longer a head;
+    # a never-aliased scratch entry appears in no history, so it is not a husk.
+    superseded = {h for hist in aliases.load_history(project).values() for h in hist} - heads
     verdicts = {
         entry["content_hash"]: entry_staleness(project, entry["content_hash"]) for entry in list_entries(project)
     }
     for content_hash, verdict in verdicts.items():
-        verdict.live = content_hash in heads
+        verdict.live = content_hash not in superseded
         if not verdict.live:
-            verdict.stale = False  # a non-head is never actionably stale (#154)
+            verdict.stale = False  # a superseded husk is never actionably stale (#154)
     directly_stale = [h for h, v in verdicts.items() if v.stale]
     if directly_stale:
         for content_hash in descendant_cone(project, directly_stale):

tests/test_auto_recalc.py

@@ -16,7 +16,7 @@
 
 from tallyman_cli.fixtures import write_shoe_orders
 from tallyman_core import data_dir
-from tallyman_core.aliases import get_alias
+from tallyman_core.aliases import get_alias, history_for
 from tallyman_core.catalog_state import current_step, reset_to
 from tallyman_core.config import auto_recalc_enabled, set_auto_recalc
 from tallyman_core.errors import error_for_hash
@@ -605,3 +605,30 @@ def test_scan_driven_recalc_does_not_replay_husk_under_source_drift(project, ord
     new = {e["content_hash"] for e in list_entries(project)} - before
     # exactly two new entries — the leaf source-recompute and u's live-head recompute — no junk husk replay
     assert new == {get_alias(project, "leaf"), get_alias(project, "u")}
+
+
+# ---------------------------------------------------------------------------
+# 9. a byte-identical revise is a no-op head advance: no new revision, no cascade
+# ---------------------------------------------------------------------------
+
+
+def test_revise_identical_code_keeps_version_and_cascades_nothing(project, orders_parquet, monkeypatch):
+    """docs/reactive-recalc.md: "a new alias revision is created exactly when an
+    alias head advances to a hash it was not already pointing at" and
+    "set_alias dedupes a consecutive duplicate hash regardless". Re-submitting
+    the identical recipe rebuilds to the same content hash, so the alias history
+    must not grow, followers must not go stale, and the cascade must be empty."""
+    _clean_env(monkeypatch)
+    a1 = _hash(catalog_create("a", _src_code(project)))
+    b1 = _hash(catalog_create("b", _child_code("a")))
+
+    out = catalog_revise("a", _src_code(project))  # byte-identical recipe
+
+    assert "error" not in out, out
+    assert out["hash"] == a1  # build idempotency: same structure, same hash
+    assert get_alias(project, "a") == a1
+    assert history_for(project, "a") == [a1]  # deduped — still version 1
+    assert out.get("version") == 1
+    assert get_alias(project, "b") == b1  # nothing went stale, nothing cascaded
+    assert (out.get("recalc") or {}).get("remap") in (None, {})
+    assert scan(project)[b1].stale is False

tests/test_io.py

@@ -32,3 +32,18 @@ def test_read_project_file_returns_xorq_expr(orders_parquet: Path, project: str,
     schema = expr.schema()
     assert "region" in schema.names
     assert "price" in schema.names
+
+
+def test_tracked_expr_from_alias_rejects_a_hash(orders_parquet: Path, project: str, monkeypatch):
+    """docs/reactive-recalc.md: ``tracked_expr_from_alias`` accepts an alias only
+    — "pass it a hash and it raises, because a hash has no head to follow". The
+    hash form must not silently degrade into an untracked/pinned read."""
+    from tallyman_xorq import build_and_persist
+    from tallyman_xorq.io import tracked_expr_from_alias
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    code = 'from tallyman_xorq.io import read_project_file\nexpr = read_project_file("orders.parquet")\n'
+    content_hash = build_and_persist(project, code).content_hash
+
+    with pytest.raises(Exception):
+        tracked_expr_from_alias(content_hash, project=project)