Merge pull request #127 from buckaroo-data/fix/115-digest-pinned-reconstruction

fix(source-identity): digest-pinned reconstruction — cold reads serve built bytes (#115)

Author: paddymul

src/tallyman_xorq/io.py

@@ -52,6 +52,19 @@ def from_project(rel_path: str, project: str | None = None):
 
     proj = resolve_project(project)
     path = project_path(rel_path, proj)
+    if si.mode() == "cas":
+        recorded = _reconstructing_source_digest(proj, rel_path)
+        if recorded is not None:
+            # Reconstruction (#115): this from_project is re-running an already-built
+            # entry's recipe (expr.py), not authoring a new entry. Resolve to the FROZEN
+            # .cas clone the entry was built from — named by the digest in its
+            # manifest.sources — instead of re-digesting the live file, which may have
+            # been edited in place since the build. Note that same frozen digest (the
+            # source collector is independent of from_catalog's _RECONSTRUCTING-gated
+            # parent capture, so both fire) so a child build reconstructing this entry
+            # records the frozen digest and gc_cas keeps the clone alive.
+            si.note_source(rel_path, recorded)
+            return deferred_read_parquet(str(si.recon_cas_path(proj, path, recorded)))
     if si.mode() != "off":
         digest = si.digest_for(proj, path)
         si.note_source(rel_path, digest)
@@ -60,6 +73,26 @@ def from_project(rel_path: str, project: str | None = None):
     return deferred_read_parquet(str(path))
 
 
+def _reconstructing_source_digest(proj: str, rel_path: str) -> str | None:
+    """The digest *rel_path* was built with, if a recipe for *proj* is being
+    reconstructed on this call stack (#115); otherwise None.
+
+    Reads the per-entry ``{rel_path: digest}`` map ``result_cache._recipe_expr``
+    threads through ``_RECON_SOURCES``. Returns None outside reconstruction (the build
+    path), when the reconstructing entry belongs to a different project, or when this
+    source was not recorded — from_project then falls through to its live-file path.
+    """
+    from tallyman_xorq.result_cache import _RECON_SOURCES
+
+    ctx = _RECON_SOURCES.get()
+    if ctx is None:
+        return None
+    recon_proj, sources = ctx
+    if recon_proj != proj:
+        return None
+    return sources.get(rel_path)
+
+
 def from_catalog(alias_or_hash: str, project: str | None = None):
     """Load an existing catalog entry's result as a xorq expression.
 

src/tallyman_xorq/result_cache.py

@@ -114,6 +114,35 @@ def cache_worthy(project: str, content_hash: str) -> bool:
 # machine-locking memory blowup into a fast, clear error instead.
 _MAX_RECON_DEPTH = 64
 
+# The source digests ({rel_path: digest} from manifest.sources) the entry being
+# reconstructed on this stack was BUILT from, threaded into from_project so a cold
+# read resolves each source to the frozen data/.cas/<digest> clone it was built from —
+# not a re-digest of the (possibly edited-in-place) live file (#115). Set per-entry in
+# _recipe_expr alongside _RECONSTRUCTING and OVERWRITTEN (then restored) by each nested
+# reconstruction, so a grandparent's from_project sees the grandparent's recorded
+# sources. Carries the project so a from_project resolving a *different* project falls
+# through to the live path. None outside reconstruction (the build path); an entry that
+# recorded no sources (mode=off / pre-#86) yields an empty map, so from_project also
+# falls through.
+_RECON_SOURCES: contextvars.ContextVar[tuple[str, dict] | None] = contextvars.ContextVar(
+    "_recon_sources", default=None
+)
+
+
+def _recorded_sources(project: str, content_hash: str) -> dict:
+    """The entry's recorded ``manifest.sources`` ({rel_path: digest}), or ``{}``.
+
+    Empty when the manifest is unreadable or the entry recorded no sources (mode=off,
+    or a build predating #86); from_project then falls through to its live-file path.
+    """
+    from tallyman_core import read_manifest
+    from tallyman_core.paths import entry_dir
+
+    try:
+        return read_manifest(entry_dir(project, content_hash)).sources or {}
+    except (OSError, ValueError):
+        return {}
+
 
 def _resolve_noncyclic_hash(project: str, requested: str, content_hash: str) -> str:
     """The hash ``from_catalog`` should load, stepped out of any reconstruction cycle.
@@ -176,10 +205,17 @@ def _recipe_expr(project: str, content_hash: str):
     code = (entry_dir(project, content_hash) / "expr.py").read_text().replace(PLACEHOLDER, str(project_dir(project)))
     # Mark this entry in-flight for the duration of the recipe exec, so any
     # from_catalog the recipe issues can step back out of a self-reference (#74).
+    # In the SAME window, pin this entry's recorded sources so any from_project the
+    # recipe issues resolves to the frozen .cas clone it was built from, not a
+    # re-digest of the edited-in-place live file (#115). Both contextvars must wrap
+    # _import_script — that is where the recipe's from_catalog / from_project run —
+    # and reset in this finally, NOT the sys.modules-cleanup finally below.
     token = _RECONSTRUCTING.set(active | {(project, content_hash)})
+    recon_token = _RECON_SOURCES.set((project, _recorded_sources(project, content_hash)))
     try:
         module, tmp = _import_script(code)
     finally:
+        _RECON_SOURCES.reset(recon_token)
         _RECONSTRUCTING.reset(token)
     try:
         expr = getattr(module, _RECIPE_VAR, None)

src/tallyman_xorq/source_identity.py

@@ -36,6 +36,7 @@
 import contextvars
 import hashlib
 import json
+import logging
 import os
 import shutil
 import subprocess
@@ -44,6 +45,11 @@
 
 from tallyman_core import artifacts_dir, data_dir
 
+# Shares tallyman.perf with result_cache so the one branch where reconstruction
+# can't be content-faithful surfaces on the same namespace as the #83 self-heal
+# faithfulness warning.
+_log = logging.getLogger("tallyman.perf")
+
 _MODE_ENV = "TALLYMAN_SOURCE_IDENTITY"
 _REHASH_ENV = "TALLYMAN_SOURCE_REHASH"
 _MODES = ("off", "cas", "salt")
@@ -134,6 +140,40 @@ def ensure_cas_path(project: str, src: Path, digest: str) -> Path:
     return dst
 
 
+def recon_cas_path(project: str, live_src: Path, digest: str) -> Path:
+    """The frozen ``.cas`` clone named by *digest*, for digest-pinned reconstruction (#115).
+
+    Returns ``data/.cas/<digest><suffix>`` — the bytes the entry was built from —
+    WITHOUT re-digesting the live source, so a cold read after an in-place edit serves
+    what was built, not the edited file. The clone is normally already on disk: the
+    ``.cas`` GC (``gc_cas``) preserves any clone a live entry's ``manifest.sources``
+    references.
+
+    If the clone is absent — manually deleted, or a fresh cross-machine catalog clone,
+    since ``.cas`` lives under ``data/`` outside the catalog git repo — re-materialise
+    it from the live source IFF the live bytes still hash to *digest*. The check hashes
+    file *content* (``_digest_file``), never ``digest_for``, so the stat-keyed memo
+    can't certify an in-place edit that preserved ``(mtime_ns, size, inode)`` as the
+    original (the documented memo hole). When the clone is gone AND the live source has
+    drifted, the original bytes are unrecoverable: serve the live source best-effort
+    (never crash a read) but warn — this is the one branch that is NOT content-faithful.
+    """
+    cas_dir = data_dir(project) / ".cas"
+    dst = cas_dir / f"{digest}{live_src.suffix}"
+    if dst.exists():
+        return dst
+    if live_src.exists() and _digest_file(live_src) == digest:
+        return ensure_cas_path(project, live_src, digest)
+    _log.warning(
+        "recon_cas_path: frozen .cas clone %s%s missing and live source %s drifted — "
+        "serving live bytes; this cold read is NOT faithful to the build (#115)",
+        digest,
+        live_src.suffix,
+        live_src,
+    )
+    return live_src
+
+
 def gc_cas(project: str, live_digests: set[str]) -> int:
     """Delete ``.cas`` clones whose digest no live entry references.
 

tests/test_result_cache.py

@@ -872,6 +872,224 @@ def _just_raise():
     assert not p.exists()
 
 
+# ---------------------------------------------------------------------------
+# #115: digest-pinned reconstruction — a cold read must serve the bytes the entry
+# was BUILT from (the frozen data/.cas/<digest> clone), not a re-digest of the
+# live source after it is edited in place. #86 made *build* identity content-aware
+# (a rebuild forks the hash) but left cold *reconstruction* re-running the recipe's
+# from_project over the live file. These pin the closure for both #74 hazards: a
+# cheap entry's cold read, and an expensive entry self-healing an evicted snapshot.
+# ---------------------------------------------------------------------------
+
+
+def _write_ints(path, values) -> None:
+    import pandas as pd
+
+    pd.DataFrame({"x": list(values)}).to_parquet(path)
+
+
+def _src_read_code(project: str, rel: str) -> str:  # cheap: a bare project read
+    return f"from tallyman_xorq.io import from_project\nexpr = from_project({rel!r}, project={project!r})\n"
+
+
+def _src_group_code(project: str, rel: str) -> str:  # expensive: group_by → Aggregate → baked snapshot
+    # order_by('x') pins a deterministic row order so the #83 result_digest is stable
+    # build-to-self-heal (an unordered aggregate would reshuffle and read as drift).
+    return (
+        "from tallyman_xorq.io import from_project\n"
+        f"t = from_project({rel!r}, project={project!r})\n"
+        "expr = t.group_by('x').aggregate(c=t.count()).order_by('x')\n"
+    )
+
+
+def test_cas_cold_read_after_inplace_edit_serves_built_bytes(project, monkeypatch):
+    """#115: a cheap entry's cold read serves the rows it was built from, even after
+    the source is edited in place.
+
+    Pre-#115 the cold read re-imports the recipe and re-runs from_project, which under
+    cas re-digests the now-edited live file and clones it under a NEW digest — serving
+    the edited rows under the entry's ORIGINAL content_hash. Digest-pinned
+    reconstruction resolves to the frozen .cas clone the entry recorded instead.
+    """
+    from tallyman_core import data_dir
+    from tallyman_xorq import build_and_persist
+    from tallyman_xorq import source_identity as si
+    from tallyman_xorq.result_cache import cached_result_expr
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas")  # several sibling tests force off
+    assert si.mode() == "cas"
+
+    src = data_dir(project) / "src.parquet"
+    _write_ints(src, [0, 1, 2])
+    h = build_and_persist(project, _src_read_code(project, "src.parquet")).content_hash
+
+    # Edit the source in place AFTER building. The 3→5 row edit changes the file
+    # size, so the stat-keyed digest memo invalidates without TALLYMAN_SOURCE_REHASH.
+    _write_ints(src, [0, 1, 2, 3, 4])
+    cached_result_expr.cache_clear()  # force a genuinely COLD read (re-import the recipe)
+
+    df = cached_result_expr(project, h).execute()
+    # Assert on VALUES, not just the count, so a fix that read some other 3-row file
+    # can't false-green: only the frozen clone holds exactly {0,1,2} now.
+    assert sorted(df["x"].tolist()) == [0, 1, 2]
+
+
+def test_cas_expensive_self_heal_after_edit_serves_built_bytes(project, monkeypatch, caplog):
+    """#115: an expensive entry self-healing an evicted snapshot recomputes from the
+    FROZEN source, not the edited live file.
+
+    The self-heal re-imports the recipe (cold _resolve_result_plan) and re-executes
+    the baked op; pre-#115 that recompute read the edited source, so the snapshot
+    self-healed to different bytes under the original content_hash — exactly the #83
+    UNFAITHFUL signal. Pinned reconstruction reads the frozen clone, so the heal is
+    faithful and the warning does not fire.
+    """
+    import logging
+
+    from tallyman_core import data_dir
+    from tallyman_xorq import build_and_persist
+    from tallyman_xorq import source_identity as si
+    from tallyman_xorq.result_cache import baked_snapshot_path, cached_result_expr
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas")
+    assert si.mode() == "cas"
+
+    src = data_dir(project) / "src.parquet"
+    _write_ints(src, [0, 1, 2])
+    h = build_and_persist(project, _src_group_code(project, "src.parquet")).content_hash
+
+    # Capture the snapshot path BEFORE editing. Post-edit, baked_snapshot_path
+    # re-imports the recipe and (on current code) re-digests the edited source, so it
+    # returns a NEW, non-existent path — unlinking that would error in setup, not test
+    # the fix.
+    p = baked_snapshot_path(project, h)
+    assert p is not None and p.exists()
+
+    _write_ints(src, [0, 1, 2, 3, 4])  # edit the source in place
+    p.unlink()  # evict the baked snapshot → the next read must self-heal
+    cached_result_expr.cache_clear()
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING, logger="tallyman.perf"):
+        df = cached_result_expr(project, h).execute()
+
+    # group_by('x') over the frozen {0,1,2} → 3 groups; the edited {0,1,2,3,4} would
+    # give 5. The self-heal must recompute from the frozen clone.
+    assert sorted(df["x"].tolist()) == [0, 1, 2]
+    assert p.exists()  # the heal rewrote the ORIGINAL-keyed snapshot, not a drifted one
+    # The #83 faithfulness check must NOT fire: the recompute matches the build digest
+    # because it read the frozen bytes (pre-fix it self-healed to edited bytes and
+    # logged UNFAITHFUL).
+    msgs = [r.getMessage() for r in caplog.records if r.name == "tallyman.perf"]
+    assert not any("UNFAITHFUL" in m for m in msgs), msgs
+
+
+def test_cas_nested_chain_reconstruction_pins_each_entrys_sources(project, monkeypatch):
+    """#115: nested reconstruction restores the per-entry source map across the chain.
+
+    B unions from_catalog(A) with its own from_project(b). A cold read of B
+    reconstructs A — whose from_project must resolve to A's recorded a.parquet — then
+    reads B's own recorded b.parquet. The frozen-source contextvar is overwritten for
+    A's reconstruction and restored for B's remaining reads, so each from_project
+    pins ITS OWN built bytes even after both sources are edited in place.
+    """
+    from tallyman_core import data_dir
+    from tallyman_xorq import source_identity as si
+    from tallyman_xorq.result_cache import cached_result_expr
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas")
+    assert si.mode() == "cas"
+
+    srca = data_dir(project) / "a.parquet"
+    srcb = data_dir(project) / "b.parquet"
+    _write_ints(srca, [0, 1, 2])
+    _write_ints(srcb, [10, 11])
+
+    catalog_create("pa", _src_read_code(project, "a.parquet"))  # cheap parent over a.parquet
+    b_code = (
+        "from tallyman_xorq.io import from_project, from_catalog\n"
+        "a = from_catalog('pa')\n"
+        f"b = from_project('b.parquet', project={project!r})\n"
+        "expr = a.union(b)\n"
+    )
+    res = catalog_create("pb", b_code)
+    assert "error" not in res, res
+    b_h = _hash_of(project)
+
+    _write_ints(srca, [0, 1, 2, 3, 4])  # edit BOTH sources in place after building
+    _write_ints(srcb, [10, 11, 12])
+    cached_result_expr.cache_clear()
+
+    df = cached_result_expr(project, b_h).execute()
+    # Frozen a = {0,1,2} ∪ frozen b = {10,11}. A leaked contextvar would serve the
+    # edited {0,1,2,3,4} and/or {10,11,12}.
+    assert sorted(df["x"].tolist()) == [0, 1, 2, 10, 11]
+
+
+def test_cas_child_records_reconstructed_parent_frozen_digest(project, monkeypatch):
+    """#115: building a child records the parent's FROZEN source digest, keeping the
+    parent's .cas clone alive for GC.
+
+    Building a child reconstructs the parent's recipe, re-running the parent's
+    from_project. That must record the parent's frozen digest (the bytes actually
+    read) into the child's manifest.sources — not a re-digest of the edited live file
+    — so gc_cas keeps the frozen clone the child's reconstruction depends on.
+    """
+    from tallyman_core import data_dir, entry_dir, read_manifest
+    from tallyman_xorq import source_identity as si
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas")
+    assert si.mode() == "cas"
+
+    src = data_dir(project) / "src.parquet"
+    _write_ints(src, [0, 1, 2])
+    catalog_create("pp", _src_read_code(project, "src.parquet"))
+    p_h = _hash_of(project)
+    p_digest = read_manifest(entry_dir(project, p_h)).sources["src.parquet"]
+
+    _write_ints(src, [0, 1, 2, 3, 4])  # edit in place, then build a child off the parent
+    cc_code = "from tallyman_xorq.io import from_catalog\nt = from_catalog('pp')\nexpr = t.mutate(y=t.x * 2)\n"
+    res = catalog_create("cc", cc_code)
+    assert "error" not in res, res
+    c_h = _hash_of(project)
+
+    child_sources = read_manifest(entry_dir(project, c_h)).sources or {}
+    assert child_sources.get("src.parquet") == p_digest
+    # Sanity: the frozen digest the child pinned is NOT the edited live digest.
+    assert si.digest_for(project, src) != p_digest
+
+
+def test_off_built_entry_reads_under_cas_default_without_crash(project, monkeypatch):
+    """#115 graceful degradation: an entry built under mode=off (no recorded sources)
+    read under the cas default falls through to a live read, not a crash.
+
+    The digest-pinned hook is a no-op when manifest.sources is None — there is no
+    frozen clone to pin — so reconstruction reads the live source as before. (Per the
+    single-user rebuild-always policy, rebuilding such an entry under cas is the path
+    to pinning; this only pins the no-crash contract.)
+    """
+    from tallyman_core import data_dir, entry_dir, read_manifest
+    from tallyman_xorq import build_and_persist
+    from tallyman_xorq.result_cache import cached_result_expr
+
+    monkeypatch.setenv("TALLYMAN_PROJECT", project)
+    src = data_dir(project) / "src.parquet"
+    _write_ints(src, [0, 1, 2])
+
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "off")
+    h = build_and_persist(project, _src_read_code(project, "src.parquet")).content_hash
+    assert read_manifest(entry_dir(project, h)).sources is None  # off records no sources
+
+    monkeypatch.setenv("TALLYMAN_SOURCE_IDENTITY", "cas")  # flip to the default
+    cached_result_expr.cache_clear()
+    df = cached_result_expr(project, h).execute()  # must not raise; reads the live source
+    assert sorted(df["x"].tolist()) == [0, 1, 2]
+
+
 def test_reset_to_clears_result_plan_memo(project, orders_parquet, monkeypatch):
     """reset_to retires the on-disk compute cache to the target revision's warm
     set; it must also invalidate the in-process ``_resolve_result_plan`` memo,