fix(diff): crash-safe build expansion + restart-safe diff sessions

Review fixes for the load-expr caching work:

- ensure_session gates expansion on a `.xorq_build_expanded.complete`
  marker written last, so a crash mid-expand re-expands rather than
  reusing a truncated dir (which fed buckaroo a zero-row build_dir, or
  raised FileExistsError out of the session lock).
- Track diff-compare sessions on BuckarooManager and clear them on a
  buckaroo restart (same started_at signal as _sessions); a post-restart
  diff view now re-POSTs instead of returning a dead session id.
- api_diff_data logs the full traceback and returns a concise
  type+message as the error detail.
- Bound the result/compare lru_caches; drop a duplicate warmup import;
  expand_to_tmp delegates to expand_into_dir.

New tests (heals_partial_expansion, diff_sessions_invalidated_on_buckaroo_restart)
confirmed failing before the fix.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: paddymul

src/tallyman_companion/app.py

@@ -210,12 +210,14 @@ def _compute_disk_usage(project: str) -> dict:
     }
 
 
-@functools.lru_cache(maxsize=None)
+@functools.lru_cache(maxsize=128)
 def _build_compare_expr(project: str, a_hash: str, b_hash: str, keys: tuple[str, ...]) -> tuple[Path, dict]:
     """Build and cache an xorq outer-join comparison expression for a diff pair.
 
     Keyed by (project, a_hash, b_hash, keys) — both entries are immutable
     content-addressed artifacts so the result is stable for the process lifetime.
+    Bounded so the cache can't grow without limit over a long-lived process;
+    eviction just rebuilds the (deterministic) comparison on the next view.
 
     Returns (build_path, column_config_overrides).
     """
@@ -235,11 +237,6 @@ def _build_compare_expr(project: str, a_hash: str, b_hash: str, keys: tuple[str,
     return build_path, overrides
 
 
-# Session IDs for which Buckaroo's /load_expr has already succeeded this process.
-# Skips the re-POST on repeat diff views for the same pair.
-_loaded_diff_sessions: set[str] = set()
-
-
 def _annotate_entries(project: str, entries: list[dict]) -> list[dict]:
     """Add alias/version fields and sort: named entries (by alias) first, then scratch."""
     aliases = load_aliases(project)
@@ -398,7 +395,6 @@ async def _warm_expr_cache():
         from starlette.concurrency import run_in_threadpool  # noqa: PLC0415
 
         from tallyman_core.paths import catalog_dir  # noqa: PLC0415
-        from tallyman_xorq.result_cache import cached_result_expr  # noqa: PLC0415
 
         project = _current_project()
         if project is None:
@@ -772,8 +768,6 @@ def diff_default_versions(project: str, alias: str):
     @app.get("/{project}/api/diff_data/{alias}/{va:int}/{vb:int}")
     def api_diff_data(project: str, alias: str, va: int, vb: int):
         """Diff data as JSON for the React SPA."""
-        import traceback as _tb  # noqa: PLC0415
-
         project = _validate_project(project)
         hashes = history_for(project, alias)
         if not hashes or va < 1 or vb < 1 or va > len(hashes) or vb > len(hashes):
@@ -801,8 +795,13 @@ def api_diff_data(project: str, alias: str, va: int, vb: int):
                 b_expr=b_expr,
                 keys=keys,
             )
-        except Exception:
-            raise HTTPException(500, detail=_tb.format_exc())
+        except Exception as exc:
+            # Full traceback to the server log; a concise type+message to the
+            # client (the diff UI renders `detail`). Avoids dumping internal
+            # paths/source if this endpoint ever sits behind anything but
+            # localhost, while still surfacing the real error.
+            log.exception("diff_data failed for %s/%s V%d→V%d", project, alias, va, vb)
+            raise HTTPException(500, detail=f"{type(exc).__name__}: {exc}") from exc
 
         compare_session = None
         buckaroo_ws_base_url = None
@@ -815,7 +814,7 @@ def api_diff_data(project: str, alias: str, va: int, vb: int):
                     build_path, overrides = _build_compare_expr(
                         project, a_hash, b_hash, tuple(keys)
                     )
-                    if session_id in _loaded_diff_sessions:
+                    if buckaroo.diff_session_is_loaded(session_id):
                         compare_session = session_id
                         buckaroo_ws_base_url = buckaroo.ws_base_url
                     else:
@@ -838,7 +837,7 @@ def api_diff_data(project: str, alias: str, va: int, vb: int):
                             timeout=30.0,
                         )
                         if resp.status_code == 200:
-                            _loaded_diff_sessions.add(session_id)
+                            buckaroo.mark_diff_session_loaded(session_id)
                             compare_session = session_id
                             buckaroo_ws_base_url = buckaroo.ws_base_url
                 except Exception as exc:

src/tallyman_companion/buckaroo_lifecycle.py

@@ -122,6 +122,10 @@ def __init__(
         self._client = httpx.Client(timeout=5.0)
         # Schema: {<content_hash>: {"session_id": str, "project": str}}.
         self._sessions: dict[str, dict] = {}
+        # Diff-compare session_ids (``diff-<a>-<b>``) Buckaroo has loaded this
+        # lifetime. Reset on a Buckaroo restart, same as ``_sessions`` — these
+        # sessions live only in the subprocess's RAM.
+        self._loaded_diff_sessions: set[str] = set()
         self._session_lock = threading.Lock()
         self._buckaroo_started_at: float | None = None
         # Tmp dirs we created by expanding ${TALLYMAN_PROJECT_ROOT} placeholders
@@ -224,17 +228,9 @@ def start(self) -> None:
                     started_at = health.get("started")
                     log.info("buckaroo ready on %s (started=%s)", self.base_url, started_at)
                     # If this Buckaroo started fresh (different start time from
-                    # what we last saw), the cached session_ids are stale and
-                    # we drop them so /load runs again on first hit.
-                    if started_at != self._buckaroo_started_at:
-                        if self._sessions:
-                            log.info(
-                                "buckaroo restart detected; invalidating %d cached sessions",
-                                len(self._sessions),
-                            )
-                        self._sessions = {}
-                        self._buckaroo_started_at = started_at
-                        self._persist_sessions()
+                    # what we last saw), its in-RAM sessions are gone — drop our
+                    # bookkeeping so /load_expr runs again on first hit.
+                    self._reset_session_bookkeeping_if_restarted(started_at)
                     threading.Thread(target=self._drain_stdout, daemon=True).start()
                     return
             except httpx.HTTPError:
@@ -282,6 +278,46 @@ def _cleanup_expanded_dirs(self) -> None:
             shutil.rmtree(p, ignore_errors=True)
         self._expanded_dirs.clear()
 
+    # ------------------------------------------------------------------
+    # diff-compare session bookkeeping
+    # ------------------------------------------------------------------
+
+    def diff_session_is_loaded(self, session_id: str) -> bool:
+        """True if Buckaroo has loaded this diff-compare session this lifetime."""
+        return session_id in self._loaded_diff_sessions
+
+    def mark_diff_session_loaded(self, session_id: str) -> None:
+        """Record a successful diff ``/load_expr``.
+
+        Dropped on a Buckaroo restart by ``_reset_session_bookkeeping_if_restarted``
+        — the session exists only in the subprocess's RAM, so a stale entry would
+        point a client at a session the restarted process never loaded.
+        """
+        self._loaded_diff_sessions.add(session_id)
+
+    def _reset_session_bookkeeping_if_restarted(self, started_at) -> None:
+        """Drop in-RAM session bookkeeping when a fresh Buckaroo is detected.
+
+        Buckaroo's ``/load_expr`` sessions — entry (``_sessions``) and
+        diff-compare (``_loaded_diff_sessions``) alike — live only in the
+        subprocess's memory, so a restart (a new ``started_at`` from ``/health``)
+        invalidates every session_id we've handed out. Clearing both forces a
+        re-POST on next access; that reload is cheap because the on-disk
+        stat/result cache survives the restart. Keeping a stale entry instead
+        would hand a client a dead session.
+        """
+        if started_at == self._buckaroo_started_at:
+            return
+        if self._sessions:
+            log.info(
+                "buckaroo restart detected; invalidating %d cached sessions",
+                len(self._sessions),
+            )
+        self._sessions = {}
+        self._loaded_diff_sessions.clear()
+        self._buckaroo_started_at = started_at
+        self._persist_sessions()
+
     # ------------------------------------------------------------------
     # session map (persisted to disk)
     # ------------------------------------------------------------------
@@ -486,10 +522,22 @@ def ensure_session(
             # Entries are content-addressed and immutable, so the expansion is
             # always correct once written.
             expanded = entry_dir(project, content_hash) / ".xorq_build_expanded"
-            if not expanded.exists():
-                expanded.mkdir(parents=True, exist_ok=True)
+            expanded_marker = entry_dir(project, content_hash) / ".xorq_build_expanded.complete"
+            if not expanded_marker.exists():
+                # Re-expand whenever the completion marker is absent: either this
+                # is the first load, or a previous expansion was interrupted
+                # (crash/kill/OOM) and left a partial dir. The marker is written
+                # last, so its presence is the only proof every file landed. An
+                # `exists()` check on the dir itself would treat a truncated
+                # expansion as complete — feeding buckaroo a build_dir missing
+                # files (zero rows), or raising FileExistsError out of the lock
+                # if a subdir landed half-copied — on every subsequent load.
+                if expanded.exists():
+                    shutil.rmtree(expanded)
+                expanded.mkdir(parents=True)
                 from tallyman_xorq.portable import expand_into_dir  # noqa: PLC0415
                 expand_into_dir(build_dir, project_dir(project), expanded)
+                expanded_marker.write_text("")
             stat_cache = entry_dir(project, content_hash) / ".buckaroo_stat_cache"
             stat_cache.mkdir(parents=True, exist_ok=True)
             payload: dict = {

src/tallyman_xorq/portable.py

@@ -66,17 +66,7 @@ def expand_into_dir(build_dir: Path, project_root: Path, target: Path) -> None:
 def expand_to_tmp(build_dir: Path, project_root: Path) -> Path:
     """Copy `build_dir` to a tmp dir, expanding placeholders. Caller must clean up."""
     target = Path(tempfile.mkdtemp(prefix="tallyman_load_"))
-    project_root_str = str(project_root)
-    for item in build_dir.iterdir():
-        if item.is_dir():
-            shutil.copytree(item, target / item.name)
-            continue
-        try:
-            text = item.read_text()
-        except UnicodeDecodeError:
-            shutil.copyfile(item, target / item.name)
-            continue
-        (target / item.name).write_text(text.replace(PLACEHOLDER, project_root_str))
+    expand_into_dir(build_dir, project_root, target)
     return target
 
 

src/tallyman_xorq/result_cache.py

@@ -86,12 +86,14 @@ def cache_worthy(project: str, content_hash: str) -> bool:
     return classify_build(entry_dir(project, content_hash) / "xorq_build")["worthy"]
 
 
-@functools.lru_cache(maxsize=None)
+@functools.lru_cache(maxsize=16)
 def result_cache(project: str):
     """A ParquetSnapshotCache rooted under the project's result_cache dir.
 
     One connection per project per process — lru_cache ensures xo.connect() is
     called once rather than on every cached_result_expr / ensure_result call.
+    Bounded (a process touches only a handful of projects); an evicted project
+    just reopens its connection on next use.
     """
     import xorq.api as xo
 
@@ -102,10 +104,14 @@ def result_cache(project: str):
     return xo.ParquetSnapshotCache.from_kwargs(source=xo.connect(), base_path=base)
 
 
-@functools.lru_cache(maxsize=None)
+@functools.lru_cache(maxsize=256)
 def cached_result_expr(project: str, content_hash: str):
     """The entry's result as an expression that resolves its own materialisation.
 
+    Bounded so the cache can't grow without limit across a long-lived process;
+    entries are content-addressed, so an evicted hash rebuilds an identical
+    expression on the next access.
+
     * Expensive entry → its expression wrapped in the snapshot cache: a hit
       reads the cached parquet, a miss recomputes from the build and stores.
     * Cheap entry → a read of its in-place ``result.parquet`` (the

tests/test_buckaroo.py

@@ -318,6 +318,76 @@ def json(self):
     assert expanded.is_dir()
 
 
+def test_unit_ensure_session_heals_partial_expansion(project: str, orders_parquet: Path, monkeypatch):
+    """A crash mid-expansion must not poison the stable expanded dir forever.
+
+    The expansion writes files into .xorq_build_expanded; a kill/OOM partway
+    leaves a half-written dir. The old ``if not expanded.exists()`` guard then
+    treated that truncated dir as complete and fed buckaroo a build_dir missing
+    files (zero rows), or — if a subdir landed half-copied — raised
+    FileExistsError out of the lock on every later load. ensure_session must key
+    off a completion marker written last, and re-expand when it's absent.
+    """
+    res = build_and_persist(project, _code(project))
+
+    mgr = BuckarooManager()
+    mgr.bound_port = 65000
+    mgr.proc = type("FakeProc", (), {"poll": staticmethod(lambda: None)})()
+
+    class FakeResponse:
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return {"session": "s"}
+
+    monkeypatch.setattr(mgr._client, "post", lambda *a, **kw: FakeResponse())
+
+    # Simulate an interrupted expansion: a dir with stray contents, no marker.
+    expanded = entry_dir(project, res.content_hash) / ".xorq_build_expanded"
+    marker = entry_dir(project, res.content_hash) / ".xorq_build_expanded.complete"
+    expanded.mkdir(parents=True)
+    (expanded / "stale_partial.txt").write_text("interrupted")
+    assert not marker.exists()
+
+    mgr.ensure_session(res.content_hash, project)
+
+    # Healed: the stray file is gone, the marker exists, and every file from the
+    # source build_dir was expanded in.
+    assert marker.exists()
+    assert not (expanded / "stale_partial.txt").exists()
+    build_dir = entry_dir(project, res.content_hash) / "xorq_build"
+    for item in build_dir.iterdir():
+        assert (expanded / item.name).exists(), item.name
+
+
+def test_unit_diff_sessions_invalidated_on_buckaroo_restart(project: str):
+    """A buckaroo restart must drop loaded diff-compare sessions.
+
+    Buckaroo's /load_expr sessions live only in the subprocess's RAM, so a
+    restart (fresh ``started_at`` from /health) invalidates every session_id —
+    entry sessions (``_sessions``) and diff-compare sessions alike. The diff
+    bookkeeping used to be a module global that nothing reset, so a post-restart
+    diff view skipped the re-POST and handed the client a session the new
+    process never loaded.
+    """
+    mgr = BuckarooManager()
+    mgr._buckaroo_started_at = 1000.0
+    mgr._sessions = {"h": {"session_id": "s", "project": project}}
+    mgr.mark_diff_session_loaded("diff-aaaa-bbbb")
+    assert mgr.diff_session_is_loaded("diff-aaaa-bbbb")
+
+    # Same started_at → not a restart → bookkeeping preserved.
+    mgr._reset_session_bookkeeping_if_restarted(1000.0)
+    assert mgr.diff_session_is_loaded("diff-aaaa-bbbb")
+    assert mgr._sessions
+
+    # Fresh started_at → restart → entry and diff sessions both dropped.
+    mgr._reset_session_bookkeeping_if_restarted(2000.0)
+    assert not mgr.diff_session_is_loaded("diff-aaaa-bbbb")
+    assert mgr._sessions == {}
+
+
 def test_unit_stop_cleans_many_expanded_dirs(project: str):
     """Leak guard: stop() must clean every tracked tmp dir, not just one.