perf(cache): source-cache + baked result-cache, drop blanket result.parquet (#73)

scripts/perf_report.py

@@ -424,7 +424,10 @@ def child_build(key: str, workdir: Path, scale: float) -> None:
                 "wall_s": round(wall, 2),
                 "execute_s": res.execute_seconds,
                 "peak_mb": round(max(sampler.peak(), _ru_maxrss_bytes()) / MB),
-                "result_mb": round(result_path.stat().st_size / MB, 1),
+                # #73: a build no longer writes result.parquet (cheap entries
+                # materialise nothing; expensive ones bake into the compute
+                # cache). None when absent — the comparable signal is wall/exec.
+                "result_mb": round(result_path.stat().st_size / MB, 1) if result_path.exists() else None,
             }
         )
     )

scripts/rebuild_parking_catalog.py

@@ -324,11 +324,24 @@ def _stage_sources(src_dir: Path, project: str, rows: int) -> None:
             sys.exit(f"source not found: {src}")
         target = dst / name
         if rows and rows > 0:
+            import pyarrow as pa
             import pyarrow.parquet as pq
 
-            table = pq.read_table(src)
-            pq.write_table(table.slice(0, rows), target)
-            print(f"  staged {name}: {min(rows, table.num_rows):,} rows (truncated)")
+            # Stream only the first `rows` rows — never materialise the whole
+            # source. parking_plt_only4 is 107M rows; pq.read_table() of it
+            # spikes resident memory to tens of GB (a perf-rebuild blowup the
+            # watchdog caught), so pull row-group batches until we have enough
+            # and stop. See scripts/run_under_watchdog.py.
+            pf = pq.ParquetFile(src)
+            batches, collected = [], 0
+            for batch in pf.iter_batches(batch_size=min(rows, 1_000_000)):
+                batches.append(batch)
+                collected += batch.num_rows
+                if collected >= rows:
+                    break
+            table = pa.Table.from_batches(batches, schema=pf.schema_arrow).slice(0, rows)
+            pq.write_table(table, target)
+            print(f"  staged {name}: {table.num_rows:,} rows (truncated)")
         else:
             shutil.copy2(src, target)
             print(f"  staged {name}: full copy")

scripts/run_under_watchdog.py

@@ -0,0 +1,98 @@
+"""Run a command under an RSS watchdog: SIGKILL the whole process tree the
+instant its combined resident memory crosses a cap.
+
+macOS ``ulimit -v`` caps address space, not RSS, so it does not stop a process
+that balloons resident memory — the box just swaps to death and locks. This
+wrapper polls the process tree's RSS with psutil and kills it ourselves before
+that happens. It is the guard used to run the memory-heavy perf surfaces safely:
+
+    uv run python scripts/run_under_watchdog.py 20 \\
+        uv run pytest -m perf tests/test_perf_integration.py -s
+    uv run python scripts/run_under_watchdog.py 16 \\
+        uv run python scripts/rebuild_parking_catalog.py --rows 1000000
+
+Child stdout/stderr are inherited (streamed straight through). A heartbeat line
+goes to stderr every ~15s so a long quiet stretch still shows live RSS. Exits
+137 if the cap was breached (the tree was killed), else the child's exit code.
+
+This is a backstop for *unknown* spikes (e.g. a full-source ``pq.read_table`` of
+a 100M-row parquet). The known reconstruction cycle has its own in-process
+backstop — ``result_cache._MAX_RECON_DEPTH`` — which turns a runaway
+``from_catalog`` cycle into a fast ``BuildError`` instead of an OOM.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import time
+
+import psutil
+
+
+def tree_rss(proc: psutil.Process) -> int:
+    """Combined RSS of *proc* and every descendant, in bytes."""
+    total = 0
+    try:
+        procs = [proc, *proc.children(recursive=True)]
+    except psutil.Error:
+        return 0
+    for p in procs:
+        try:
+            total += p.memory_info().rss
+        except psutil.Error:
+            pass
+    return total
+
+
+def main() -> None:
+    if len(sys.argv) < 3:
+        sys.exit("usage: run_under_watchdog.py <cap_gb> <cmd> [args...]")
+    cap_gb = float(sys.argv[1])
+    cmd = sys.argv[2:]
+    cap = cap_gb * 1024**3
+
+    t0 = time.monotonic()
+    proc = subprocess.Popen(cmd)
+    mon = psutil.Process(proc.pid)
+    peak = 0
+    last_log = 0.0
+    killed = False
+
+    while proc.poll() is None:
+        rss = tree_rss(mon)
+        peak = max(peak, rss)
+        now = time.monotonic()
+        if now - last_log > 15:
+            sys.stderr.write(
+                f"[watchdog t={now - t0:.0f}s] tree RSS={rss / 1024**3:.2f}GB peak={peak / 1024**3:.2f}GB\n"
+            )
+            sys.stderr.flush()
+            last_log = now
+        if rss > cap:
+            killed = True
+            sys.stderr.write(
+                f"\n!! WATCHDOG KILL: tree RSS {rss / 1024**3:.1f}GB > cap {cap_gb}GB at t={now - t0:.0f}s\n"
+            )
+            sys.stderr.flush()
+            for p in [mon, *mon.children(recursive=True)]:
+                try:
+                    p.kill()
+                except psutil.Error:
+                    pass
+            break
+        time.sleep(0.25)
+
+    try:
+        proc.wait(timeout=15)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+    status = "KILLED" if killed else f"exit {proc.returncode}"
+    peak_gb, elapsed = peak / 1024**3, time.monotonic() - t0
+    sys.stderr.write(f"\n=== watchdog: {status} peak={peak_gb:.2f}GB elapsed={elapsed:.0f}s cap={cap_gb}GB ===\n")
+    sys.stderr.flush()
+    sys.exit(137 if killed else (proc.returncode or 0))
+
+
+if __name__ == "__main__":
+    main()

src/tallyman_companion/app.py

@@ -156,17 +156,16 @@ def _compute_disk_usage(project: str) -> dict:
     """Walk a project's on-disk footprint and return the disk_usage payload.
 
     Covers raw input data, each entry's result.parquet / xorq_build /
-    .buckaroo_stat_cache, and the two project-level caches added in #12 — the
-    xorq ParquetSnapshotCache (result_cache/) and the per-pair Buckaroo diff
-    stat cache (diff_stat_cache/), either of which can dwarf the rest.
+    .buckaroo_stat_cache, and the per-pair Buckaroo diff stat cache
+    (diff_stat_cache/), which can dwarf the rest. (The retired result_cache/
+    dir is gone as of #73 — baked results now live in the compute cache.)
 
     This is the expensive path; callers should rate-limit it via the
     api_disk_usage TTL cache rather than invoking per request.
     """
     from tallyman_core.paths import data_dir as _data_dir
     from tallyman_core.paths import diff_stat_cache_root as _diff_stat_cache_root
     from tallyman_core.paths import entries_dir as _entries_dir
-    from tallyman_core.paths import result_cache_dir as _result_cache_dir
 
     # raw input files
     data = _dir_size(_data_dir(project))
@@ -190,26 +189,23 @@ def _compute_disk_usage(project: str) -> dict:
             if c.is_dir():
                 cache += _dir_size(c)
 
-    # Project-level caches (#12), previously uncounted. _dir_size returns 0 for
-    # a dir that doesn't exist yet, so no existence guard needed.
-    result_cache = _dir_size(_result_cache_dir(project))
+    # Project-level diff stat cache (#12), previously uncounted. _dir_size
+    # returns 0 for a dir that doesn't exist yet, so no existence guard needed.
     diff_cache = _dir_size(_diff_stat_cache_root(project))
 
-    total = data + results + builds + cache + result_cache + diff_cache
+    total = data + results + builds + cache + diff_cache
     return {
         "data": data,
         "results": results,
         "builds": builds,
         "cache": cache,
-        "result_cache": result_cache,
         "diff_cache": diff_cache,
         "total": total,
         "formatted": {
             "data": _fmt_bytes(data),
             "results": _fmt_bytes(results),
             "builds": _fmt_bytes(builds),
             "cache": _fmt_bytes(cache),
-            "result_cache": _fmt_bytes(result_cache),
             "diff_cache": _fmt_bytes(diff_cache),
             "total": _fmt_bytes(total),
         },
@@ -310,6 +306,17 @@ def create_app(
         except (FileNotFoundError, ValueError):
             pass
 
+    # #73 upgrade migration: drop the old per-entry build-time result.parquet
+    # once. Runs at construction (before any lazy ensure_result regenerates one)
+    # and is guarded by a per-project marker, so it's a no-op on later starts.
+    if seed and not read_only:
+        try:
+            from tallyman_xorq.build import migrate_drop_result_parquet
+
+            migrate_drop_result_parquet(seed)
+        except Exception:  # best-effort; cleanup must never block startup
+            log.warning("result.parquet migration failed for %s", seed, exc_info=True)
+
     app = FastAPI(title="tallyman companion")
     if STATIC_DIR.exists():
         app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
@@ -514,10 +521,15 @@ def api_data(project: str, content_hash: str, offset: int = 0, limit: int = 200)
         _require_hash(content_hash)
         if limit < 0:
             raise HTTPException(400, "limit must be >= 0")
-        entry = entry_dir(project, content_hash)
-        result_path = entry / ENTRY_RESULT_FILENAME
-        if not result_path.exists():
-            raise HTTPException(404, "no result.parquet")
+        from tallyman_core.paths import entry_build_dir  # noqa: PLC0415
+        from tallyman_xorq.result_cache import ensure_result  # noqa: PLC0415
+
+        if not entry_build_dir(project, content_hash).is_dir():
+            raise HTTPException(404, "no entry")
+        # #73: cheap entries write no result.parquet at build; ensure_result
+        # materialises one on demand (expensive → result_cache snapshot, cheap →
+        # regenerated in place) so this paginated read always has a parquet.
+        result_path = ensure_result(project, content_hash)
         pf = pq.ParquetFile(result_path)
         total = pf.metadata.num_rows
         needed = offset + limit if limit > 0 else 0
@@ -603,8 +615,9 @@ def api_entry_detail(project: str, content_hash: str):
         )
         buckaroo_ws_base = buckaroo.ws_base_url if buckaroo and buckaroo.is_running else None
 
-        result_path = entry / ENTRY_RESULT_FILENAME
-        total_rows = pq.ParquetFile(result_path).metadata.num_rows if result_path.exists() else 0
+        # #73: row count comes from the manifest, not a result.parquet (cheap
+        # entries no longer write one at build time).
+        total_rows = (manifest or {}).get("row_count", 0)
 
         return {
             "project": project,
@@ -729,9 +742,8 @@ def api_notebook_full(project: str):
                 if (entry / ENTRY_MANIFEST_FILENAME).exists():
                     entry_meta = json.loads((entry / ENTRY_MANIFEST_FILENAME).read_text())
                     schema = json.loads((entry / ENTRY_SCHEMA_FILENAME).read_text())
-                result_path = entry / ENTRY_RESULT_FILENAME
-                if result_path.exists():
-                    total_rows = pq.ParquetFile(result_path).metadata.num_rows
+                    # #73: row count from the manifest, not a result.parquet.
+                    total_rows = entry_meta.get("row_count", 0)
                 chart_spec = get_chart(project, latest)
                 if buckaroo_available:
                     buckaroo_session = buckaroo.ensure_session(latest, project)
@@ -815,9 +827,7 @@ def api_diff_data(project: str, alias: str, va: int, vb: int):
             if keys:
                 session_id = f"diff-{a_hash[:12]}-{b_hash[:12]}"
                 try:
-                    build_path, overrides = _build_compare_expr(
-                        project, a_hash, b_hash, tuple(keys)
-                    )
+                    build_path, overrides = _build_compare_expr(project, a_hash, b_hash, tuple(keys))
                     if buckaroo.diff_session_is_loaded(session_id):
                         compare_session = session_id
                         buckaroo_ws_base_url = buckaroo.ws_base_url

src/tallyman_companion/buckaroo_lifecycle.py

@@ -511,6 +511,22 @@ def ensure_session(
         build_dir = entry_build_dir(project, content_hash)
         if not build_dir.is_dir():
             return None
+        # Materialise the entry's baked snapshot before buckaroo replays the
+        # build via /load_expr. A from_catalog chain off an expensive parent
+        # embeds a *bare* read of the parent's snapshot (#75 strips the parent
+        # CachedNode to keep the composed expression on one backend), so on a
+        # cold compute cache — a fresh clone, or an entry the startup warmup
+        # didn't reach — the replay reads zero rows and the grid renders empty.
+        # cached_result_expr repopulates it (idempotent; lru-cached, so a warmed
+        # entry is ~free), mirroring ensure_result's heal for the paginated read.
+        # Done before the lock: a cold entry's recompute mustn't block other
+        # sessions, and a warm one is a cheap cache hit.
+        try:
+            from tallyman_xorq.result_cache import cached_result_expr  # noqa: PLC0415
+
+            cached_result_expr(project, content_hash)
+        except Exception:
+            log.debug("snapshot self-heal before /load_expr failed for %s", content_hash, exc_info=True)
         with self._session_lock:
             cached = self._sessions.get(content_hash)
             if cached:

src/tallyman_core/aliases.py

@@ -86,6 +86,22 @@ def version_of_hash(project: str, content_hash: str) -> tuple[str, int] | None:
     return None
 
 
+def previous_version(project: str, content_hash: str) -> str | None:
+    """The hash one revision earlier in this entry's alias history, or None.
+
+    None when the entry is a lineage root (version 1) or appears in no history.
+    """
+    info = version_of_hash(project, content_hash)
+    if not info:
+        return None
+    name, version = info
+    if version <= 1:
+        return None
+    hist = history_for(project, name)
+    idx = version - 2  # version is 1-based; the parent is the entry before it
+    return hist[idx] if 0 <= idx < len(hist) else None
+
+
 def set_alias(project: str, name: str, content_hash: str, *, expect_exists: bool | None = None) -> dict:
     """Point `name` at `content_hash` and append to history.
 

src/tallyman_core/catalog_state.py

@@ -9,7 +9,6 @@
     display_configs: [{content_hash, config}]
     notebook:        {cells: [{cell_id, alias, markdown}]}
     entry_hashes:    [content_hash, ...]      # pointer to untracked entries/<hash>/
-    result_cache:    [relpath, ...]           # pointer to untracked result cache
     compute_cache:   [relpath, ...]           # pointer to untracked compute cache
     alias_map:       {alias: latest_hash}     # tallyman's alias bookkeeping
     alias_history:   {alias: [hash, ...]}     # tallyman's alias bookkeeping
@@ -62,7 +61,6 @@
     entries_dir,
     notebooks_dir,
     post_processing_dir,
-    result_cache_dir,
     stats_dir,
 )
 
@@ -107,7 +105,6 @@ def read_tallyman_state(project: str) -> dict:
         "display_configs": raw.get("display_configs", []),
         "notebook": raw.get("notebook", {"cells": []}),
         "entry_hashes": raw.get("entry_hashes", []),
-        "result_cache": raw.get("result_cache", []),
         "compute_cache": raw.get("compute_cache", []),
         "alias_map": raw.get("alias_map", {}),
         "alias_history": raw.get("alias_history", {}),
@@ -167,7 +164,6 @@ def capture_tallyman_state(project: str) -> dict:
         ],
         "notebook": notebook.load(project),
         "entry_hashes": sorted(c.name for c in ed.iterdir() if c.is_dir()) if ed.exists() else [],
-        "result_cache": _list_cache_files(result_cache_dir(project)),
         "compute_cache": _list_cache_files(compute_cache_dir(project)),
     }
     # alias_map/alias_history are written live by aliases.py and already in
@@ -297,10 +293,6 @@ def _prune_cache(project: str, root: Path, key: str) -> int:
     return removed
 
 
-def prune_result_cache(project: str) -> int:
-    return _prune_cache(project, result_cache_dir(project), "result_cache")
-
-
 def prune_compute_cache(project: str) -> int:
     return _prune_cache(project, compute_cache_dir(project), "compute_cache")
 
@@ -324,13 +316,13 @@ def restore_from_bullpen(project: str) -> int:
         if not live.exists() and parked.is_dir():
             shutil.copytree(parked, live)
             restored += 1
-    for key, root in (("result_cache", result_cache_dir(project)), ("compute_cache", compute_cache_dir(project))):
-        for rel in raw.get(key) or []:
-            live, parked = root / rel, bp / key / rel
-            if not live.exists() and parked.is_file():
-                live.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copy2(parked, live)
-                restored += 1
+    root = compute_cache_dir(project)
+    for rel in raw.get("compute_cache") or []:
+        live, parked = root / rel, bp / "compute_cache" / rel
+        if not live.exists() and parked.is_file():
+            live.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(parked, live)
+            restored += 1
     return restored
 
 
@@ -484,7 +476,6 @@ def reset_to(project: str, ref: int | str) -> None:
             raise RuntimeError(f"catalog reset to {tag!r} failed: {err}")
         materialize(project)
         prune_entries(project)
-        prune_result_cache(project)
         prune_compute_cache(project)
         restore_from_bullpen(project)
         _assert_catalog_consistent(project)