fix(#89): address review findings 1-4 in auto-recalc surfaces

1. Companion POST /api/promote_diff double-committed: it self-checkpoints but
   was not in _checkpoint_exempt, so the dispatch middleware checkpointed it a
   second time (a trailing empty revision) — the same bug already fixed on the
   MCP side. Exempt the route so the promote + cascade land in exactly one
   revision, mirroring catalog_promote_diff.

2. Cascade failures were persisted with a hard-coded tool="catalog_recalc" even
   when driven by the revise auto-path. Thread the triggering tool through
   auto_recalc -> _replay_cone and the MCP/companion head-advance helpers so the
   forensic record names the tool that actually ran (catalog_revise,
   catalog_promote_diff, companion_revise, companion_promote_diff).

3. auto_recalc scanned staleness twice (once for verdicts, once inside the cone
   walk). _replay_cone now accepts the caller's verdicts so a revise scans once.
   classify_orphans reads the error log and alias heads once up front
   (errors_by_hash + _alias_heads_index) instead of once per orphan.

4. Corrected the stale backfill comment in the MCP dispatch decorator: only
   catalog_revise reaches it; catalog_promote_diff is exempt and stamps its own
   checkpoint step.

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

Author: paddymul

src/tallyman_companion/app.py

@@ -367,10 +367,11 @@ async def publish(event: dict):
         for q in list(subscribers):
             await q.put(event)
 
-    async def _auto_recalc_after_head_advance(project: str, name: str) -> dict | None:
+    async def _auto_recalc_after_head_advance(project: str, name: str, *, tool: str) -> dict | None:
         """Cascade-recompute *name*'s stale followers after an in-browser head
         advance (PUT /api/code, promote_diff), in the SAME revision the checkpoint
-        middleware commits — the walk takes no checkpoint of its own. Returns the
+        middleware commits — the walk takes no checkpoint of its own. ``tool`` names
+        the route so a persisted cascade failure is attributed to it. Returns the
         recalc sub-report (plus ``orphan_stale``), or ``None`` when the per-project
         ``auto_recalc`` switch is off. On a real cascade, invalidates the companion
         caches, reloads buckaroo sessions, and publishes the canonical recalc SSE
@@ -382,7 +383,7 @@ async def _auto_recalc_after_head_advance(project: str, name: str) -> dict | Non
 
         if not await run_in_threadpool(auto_recalc_enabled, project):
             return None
-        report = await run_in_threadpool(auto_recalc, project, name)
+        report = await run_in_threadpool(auto_recalc, project, name, tool=tool)
         if report.get("remap"):
             _invalidate_reset_caches()
             if buckaroo:
@@ -414,6 +415,8 @@ def _checkpoint_exempt(method: str, path: str) -> bool:
             return True  # reset moves `current`; it is not a new step
         if path.endswith("/api/recalc"):
             return True  # recalc self-checkpoints (one tx for the whole walk; dry-run takes none)
+        if "/api/promote_diff/" in path:
+            return True  # promote_diff self-checkpoints (the promote + cascade as one tx)
         return False
 
     @app.middleware("http")
@@ -931,7 +934,11 @@ def _resolve(v: int):
 
         # D5 parity: re-pointing an existing alias cascades to its followers in the
         # same revision (the walk runs before this route's checkpoint).
-        recalc_report = await _auto_recalc_after_head_advance(project, target_alias) if repointed else None
+        recalc_report = (
+            await _auto_recalc_after_head_advance(project, target_alias, tool="companion_promote_diff")
+            if repointed
+            else None
+        )
 
         checkpoint_catalog(project, f"tallyman: catalog_promote_diff {alias} V{a_idx}→V{b_idx}")
         await publish({"kind": "entry_added", "hash": result.content_hash})
@@ -1176,7 +1183,7 @@ async def put_code(project: str, alias: str, payload: dict):
         # followers in the SAME revision — the checkpoint middleware commits the
         # head advance + the cascade as one (the walk takes no checkpoint of its
         # own). Off → today's no-cascade behavior. Parity with catalog_revise.
-        rep = await _auto_recalc_after_head_advance(project, alias)
+        rep = await _auto_recalc_after_head_advance(project, alias, tool="companion_revise")
         if rep is not None:
             reply["recalc"] = rep
         return reply

src/tallyman_core/__init__.py

@@ -21,7 +21,7 @@
     write_display_klass,
 )
 from tallyman_core.entry_config import carry_forward_entry_config
-from tallyman_core.errors import clear_errors, error_for_hash, get_error, list_errors, record_error
+from tallyman_core.errors import clear_errors, error_for_hash, errors_by_hash, get_error, list_errors, record_error
 from tallyman_core.fsutil import atomic_write_text
 from tallyman_core.manifest import Manifest, read_manifest, write_manifest
 from tallyman_core.marimo_export import export_notebook_path, notebook_to_marimo
@@ -103,6 +103,7 @@
     "clear_errors",
     "data_dir",
     "error_for_hash",
+    "errors_by_hash",
     "read_config",
     "set_auto_recalc",
     "write_config",

src/tallyman_core/errors.py

@@ -73,17 +73,30 @@ def clear_errors(project: str) -> int:
     return n
 
 
+def errors_by_hash(project: str) -> dict[str, dict]:
+    """Map each entry hash to its most-recent recorded error, in ONE pass over the log.
+
+    The batch form of ``error_for_hash`` for the orphan-stale tie-back, which
+    classifies many stale entries at once: read ``errors.jsonl`` a single time and
+    keep the freshest error per hash, instead of re-scanning the whole log once per
+    entry (the classify pass is O(orphans) lookups).
+    """
+    out: dict[str, dict] = {}
+    for row in list_errors(project, limit=_UNBOUNDED):  # most-recent first
+        h = row.get("hash")
+        if h is not None and h not in out:  # first seen wins → freshest failure
+            out[h] = row
+    return out
+
+
 def error_for_hash(project: str, content_hash: str) -> dict | None:
     """The most recent recorded error carrying ``hash == content_hash``, or None.
 
     Powers the orphan-stale tie-back: an entry that is still stale because a prior
     recalc/build of it failed is "explained by" that recorded error, not an
     unexplained invariant break. Most-recent-first so the freshest failure wins.
     """
-    for row in list_errors(project, limit=_UNBOUNDED):
-        if row.get("hash") == content_hash:
-            return row
-    return None
+    return errors_by_hash(project).get(content_hash)
 
 
 def get_error(project: str, error_id: str) -> dict | None:

src/tallyman_mcp/server.py

@@ -168,11 +168,13 @@ def wrapped(*args, **kwargs):
             project = _resolve_active_project()
             if project:
                 step = checkpoint_catalog(project, f"tallyman: {fn.__name__}")
-                # An auto-path recalc sub-report (revise/promote_diff) is built by
-                # the checkpoint-free walk *before* this per-op checkpoint exists,
-                # so its checkpoint_step is None. This is the checkpoint that
-                # commits the head advance + cascade as one revision; backfill the
-                # real landed step the report couldn't yet know.
+                # catalog_revise's auto-path recalc sub-report is built by the
+                # checkpoint-free walk *before* this per-op checkpoint exists, so its
+                # checkpoint_step is None. This is the checkpoint that commits the head
+                # advance + cascade as one revision; backfill the real landed step the
+                # report couldn't yet know. (catalog_promote_diff is _NO_CHECKPOINT-
+                # exempt and self-checkpoints, so it stamps its own step and never
+                # reaches here.)
                 recalc_report = result.get("recalc") if isinstance(result, dict) else None
                 if isinstance(recalc_report, dict) and recalc_report.get("checkpoint_step") is None:
                     recalc_report["checkpoint_step"] = step
@@ -213,10 +215,11 @@ def _notify(kind: str, content_hash: str | None = None, **extra) -> None:
         print(f"[tallyman_mcp] notify failed ({kind}): {exc}", file=sys.stderr)
 
 
-def _auto_recalc_after_head_advance(project: str, name: str) -> dict | None:
+def _auto_recalc_after_head_advance(project: str, name: str, *, tool: str) -> dict | None:
     """Cascade-recompute *name*'s stale followers right after an existing alias
     head advanced — shared by every MCP path that re-points an existing alias
-    (``catalog_revise``, ``catalog_promote_diff``).
+    (``catalog_revise``, ``catalog_promote_diff``). ``tool`` names that surface so
+    a persisted cascade failure is attributed to the tool that actually ran.
 
     The walk is **checkpoint-free**: it leaves the re-pointed dependents in the
     working tree for the surrounding op's single checkpoint (the dispatch
@@ -230,7 +233,7 @@ def _auto_recalc_after_head_advance(project: str, name: str) -> dict | None:
         return None
     from tallyman_xorq.recalc import auto_recalc  # noqa: PLC0415
 
-    report = auto_recalc(project, name)
+    report = auto_recalc(project, name, tool=tool)
     if report.get("remap"):
         # Flat kwargs: _notify packs **extra into the wire `extra`, and the
         # companion handler reads extra.remap / extra.step. A literal extra={...}
@@ -621,7 +624,7 @@ def catalog_revise(name: str, code: str, prompt: str = "") -> dict:
     # Auto-recalc: cascade-recompute name's stale followers in this op's single
     # checkpoint (the dispatch decorator commits the head advance + the cascade as
     # one revision). Off → the explicit scan→recalc path. See plans/auto-recalc-on-revise.md.
-    recalc_report = _auto_recalc_after_head_advance(project, name)
+    recalc_report = _auto_recalc_after_head_advance(project, name, tool="catalog_revise")
     if recalc_report is not None:
         out["recalc"] = recalc_report
     return out
@@ -913,7 +916,9 @@ def _resolve(v):
     # helper. A fresh target_alias (expect_exists=False) has no followers; only
     # the re-point branch can. The checkpoint-free walk runs BEFORE this op's
     # checkpoint so that one revision sweeps the promote + the cascade together.
-    recalc_report = _auto_recalc_after_head_advance(project, target_alias) if repointed else None
+    recalc_report = (
+        _auto_recalc_after_head_advance(project, target_alias, tool="catalog_promote_diff") if repointed else None
+    )
 
     # promote_diff is checkpoint-exempt (self-checkpoints): THIS is the single
     # revision the promote + cascade land in. Stamp its step onto the recalc

src/tallyman_xorq/recalc.py

@@ -43,7 +43,7 @@
 
 from tallyman_core import aliases
 from tallyman_core.entry_config import carry_forward_entry_config
-from tallyman_core.errors import error_for_hash, record_error
+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
@@ -169,7 +169,13 @@ def recalc(project: str, roots: list[str], *, dry_run: bool = False) -> RecalcRe
     return report
 
 
-def _replay_cone(project: str, roots: list[str]) -> RecalcReport:
+def _replay_cone(
+    project: str,
+    roots: list[str],
+    *,
+    verdicts: dict[str, StaleVerdict] | None = None,
+    tool: str = "catalog_recalc",
+) -> RecalcReport:
     """Replay *roots* and their cone in dependency order, **without** taking a
     checkpoint — the reusable engine behind both the public ``recalc`` and the
     auto-recalc folded into ``catalog_revise``.
@@ -182,14 +188,20 @@ def _replay_cone(project: str, roots: list[str]) -> RecalcReport:
     for the caller's single checkpoint to commit. Returns a report with
     ``checkpoint_step=None`` — the *caller* owns the one checkpoint (``recalc``
     for an explicit run; the per-op decorator/middleware for an embedded one).
+
+    ``verdicts`` lets a caller that already scanned (``auto_recalc``) hand them in
+    so the walk doesn't re-scan; ``tool`` names the surface that triggered the
+    walk, recorded on each persisted failure so the forensic record names the tool
+    that actually ran (not always ``catalog_recalc``).
     """
     roots = [r for r in roots if entry_manifest_path(project, r).exists()]
     try:
         cone = descendant_cone(project, roots)
     except DependencyCycleError as exc:
         return RecalcReport(project, roots, [], [], False, "cycle", error=str(exc))
 
-    verdicts = scan(project)
+    if verdicts is None:
+        verdicts = scan(project)
     remap: dict[str, str] = {}
     entries: list[RecalcEntry] = []
     status = "ok"
@@ -205,7 +217,7 @@ def _replay_cone(project: str, roots: list[str]) -> RecalcReport:
                     project,
                     code=_recipe_code(project, old_hash),
                     message="skipped: an upstream build failure halted the recalc walk",
-                    tool="catalog_recalc",
+                    tool=tool,
                     hash=old_hash,
                 )
             entries.append(_entry(old_hash, "skipped", verdict, project))
@@ -222,7 +234,7 @@ def _replay_cone(project: str, roots: list[str]) -> RecalcReport:
                 project,
                 code=_recipe_code(project, old_hash),
                 message=str(exc),
-                tool="catalog_recalc",
+                tool=tool,
                 hash=old_hash,
             )
             entries.append(
@@ -272,22 +284,41 @@ def followers_of(project: str, name: str, verdicts: dict[str, StaleVerdict] | No
     )
 
 
-def _classify_orphan(project: str, content_hash: str, verdict: StaleVerdict) -> dict:
+def _alias_heads_index(project: str) -> dict[str, list[str]]:
+    """``{content_hash: [alias, ...]}`` over the current alias heads, built once.
+
+    The reverse of ``_aliases_heading`` for a batch caller (``classify_orphans``)
+    so classifying N orphans loads ``aliases.jsonl`` a single time, not N times."""
+    index: dict[str, list[str]] = {}
+    for name, latest in aliases.load_aliases(project).items():
+        index.setdefault(latest, []).append(name)
+    for names in index.values():
+        names.sort()
+    return index
+
+
+def _classify_orphan(
+    content_hash: str,
+    verdict: StaleVerdict,
+    errors_index: dict[str, dict],
+    alias_index: dict[str, list[str]],
+) -> dict:
     """Classify a stale entry an auto-recalc did *not* recompute: a self-referential
     pin (expected), explained by a recorded failure, or UNEXPLAINED.
 
     Self-reference is checked first and only when it is the SOLE cause (every reason
     is the entry following its own head): an entry that is also source- or
-    alias-drifted for a real reason still classifies normally."""
+    alias-drifted for a real reason still classifies normally. ``errors_index`` /
+    ``alias_index`` are the prebuilt one-pass lookups (see ``classify_orphans``)."""
     if verdict.reasons and all(_is_self_ref(content_hash, r) for r in verdict.reasons):
         explanation = "self-referential revise pin (follows its own alias head); permanently stale by design — expected"
-    elif (err := error_for_hash(project, content_hash)) is not None:
+    elif (err := errors_index.get(content_hash)) is not None:
         explanation = f"explained by error {err['id']}"
     else:
         explanation = "UNEXPLAINED stale entry — staleness with no recorded recalc error; file a bug."
     return {
         "hash": content_hash,
-        "alias": (_aliases_heading(project, content_hash) or [None])[0],
+        "alias": (alias_index.get(content_hash) or [None])[0],
         "reasons": [asdict(r) for r in verdict.reasons],
         "explanation": explanation,
     }
@@ -303,12 +334,20 @@ def classify_orphans(
     rebuilt as *recomputed*, so only the leftovers are classified. The
     scan-on-load surfaces pass nothing: at scan time no cascade ran, so every
     directly-stale entry is an orphan to be accounted for (the D5 backstop —
-    a project load surfaces orphans even when no revise has happened since)."""
+    a project load surfaces orphans even when no revise has happened since).
+
+    The error store and alias heads are read once up front (``errors_by_hash`` /
+    ``_alias_heads_index``) and shared across all orphans, so the pass is one log
+    read + one aliases read regardless of orphan count."""
     orphans = sorted(h for h, v in verdicts.items() if v.stale and h not in recomputed)
-    return [_classify_orphan(project, h, verdicts[h]) for h in orphans]
+    if not orphans:
+        return []
+    errors_index = errors_by_hash(project)
+    alias_index = _alias_heads_index(project)
+    return [_classify_orphan(h, verdicts[h], errors_index, alias_index) for h in orphans]
 
 
-def auto_recalc(project: str, name: str) -> dict:
+def auto_recalc(project: str, name: str, *, tool: str = "catalog_revise") -> dict:
     """Recompute the followers a revise of alias *name* made stale, **without**
     taking a checkpoint, and account for every other stale entry it did not touch.
 
@@ -318,14 +357,20 @@ def auto_recalc(project: str, name: str) -> dict:
     yet recalced, or an invariant break) — left untouched, logged at WARNING, and
     returned in ``orphan_stale`` classified against the error store (#6).
 
+    ``tool`` names the surface that triggered this auto-recalc (``catalog_revise``,
+    ``catalog_promote_diff``, the companion routes); it is recorded on any persisted
+    cascade failure so the forensic record names the tool that actually ran.
+
     Returns the recalc report (``catalog_recalc``'s shape) plus ``orphan_stale``.
     The caller embeds this in its own transaction: it owns the single checkpoint
     and the cross-process notify / SSE publish.
     """
     verdicts = scan(project)
     roots = followers_of(project, name, verdicts)
 
-    report = _replay_cone(project, roots)  # checkpoint-free; caller commits
+    # Reuse the verdicts just scanned (one scan per revise, not two) and tag any
+    # persisted failure with the triggering tool.
+    report = _replay_cone(project, roots, verdicts=verdicts, tool=tool)  # checkpoint-free; caller commits
 
     orphan_stale = classify_orphans(project, verdicts, recomputed=frozenset(report.cone))
     for o in orphan_stale: