fix(cron): the dead-owner sweep writes the cause its caller observed

cron/executions.py

@@ -312,17 +312,32 @@ def finish_execution(
     "Scheduler restarted after this execution's owner exited before a durable "
     "terminal state; whether side effects ran is unknown."
 )
+_PERIODIC_REAP_REASON = (
+    "Owner process died before reaching a durable terminal state; reclaimed by the periodic "
+    "dead-owner sweep, not a scheduler restart. Whether side effects ran is unknown."
+)
+_MANUAL_REAP_REASON = (
+    "Owner process died before reaching a durable terminal state; reclaimed ahead of a manual "
+    "run, not a scheduler restart. Whether side effects ran is unknown."
+)
 _OWNER_WEDGED_REASON = (
     "Owner process is still alive but the claim outlived the derived stale bound; "
     "treated as wedged (#115692). The process was not terminated; whether side effects "
     "ran is unknown."
 )
 
 
-def recover_interrupted_executions() -> int:
+def recover_interrupted_executions(*, reason: str) -> int:
     """Mark abandoned attempts unknown without scheduling retries: rows whose owner is provably
     dead, plus rows whose live owner holds a claim older than the derived stale bound (the
-    process is not killed)."""
+    process is not killed).
+
+    ``reason`` is the cause the CALLER observed, written to every row this sweep terminalizes.
+    It is required and has no default: the sweep itself knows only that the owner is gone, so a
+    default would let a caller silently inherit a cause it never saw (``_OWNER_GONE_REASON``
+    asserts a scheduler restart). A caller that observed nothing specific passes
+    ``_OWNER_GONE_REASON`` explicitly; rows judged live-but-wedged are overridden per row with
+    ``_OWNER_WEDGED_REASON``."""
     now = _hermes_now().isoformat()
     changed = 0
     recovered: List[Dict[str, Any]] = []
@@ -340,7 +355,7 @@ def recover_interrupted_executions() -> int:
         for row in rows:
             if row["process_id"] == _PROCESS_ID:
                 continue
-            reason = _OWNER_GONE_REASON
+            row_reason = reason
             if _owner_is_live(int(row["pid"]), row["process_started_at"]):
                 # A live owner is normally a legitimately running job. A worker permanently
                 # deadlocked (e.g. futex_wait behind a route/proxy flip, #115692) also passes
@@ -355,7 +370,7 @@ def recover_interrupted_executions() -> int:
                     stale_after_resolved = True
                 if stale_after is None or _claim_age_seconds(row["claimed_at"]) <= stale_after:
                     continue
-                reason = _OWNER_WEDGED_REASON
+                row_reason = _OWNER_WEDGED_REASON
             handoff_started_at = row["handoff_started_at"]
             if (
                 row["handoff_pending"]
@@ -371,7 +386,7 @@ def recover_interrupted_executions() -> int:
                    WHERE id=? AND status=? AND process_id=? AND pid=?
                      AND handoff_pending=?
                      AND handoff_started_at IS ?""",
-                (now, reason, row["id"], row["status"], row["process_id"], row["pid"],
+                (now, row_reason, row["id"], row["status"], row["process_id"], row["pid"],
                  row["handoff_pending"], row["handoff_started_at"]),
             )
             changed += cur.rowcount

cron/scheduler.py

@@ -3613,20 +3613,34 @@ def _is_terminal() -> bool:
         # instead: retire the claim and report the uncertain run, not a pre-dispatch failure.
         from cron.scheduler_worker_failure import external_worker_exited_reason
 
-        reason = external_worker_exited_reason(returncode)
+        # The re-read row above is the ONLY evidence of which phase the worker
+        # reached: 'running' means it won the claimed→running adoption gate,
+        # 'claimed' means it died before adopting (an import-time death cannot
+        # have adopted — adoption runs inside the payload path). Stamp the true
+        # phase into the cause; never assert "after adopting" without having seen
+        # the row run.
+        _phase = current.get("status") if current else None
+        adopted: bool | None = _phase == "running" if _phase in ("claimed", "running") else None
+        # Compose the stderr tail BEFORE the ledger write, not only into the raised
+        # message. This is the one branch where the worker's cause of death is in
+        # hand; an import-time death (no durable terminal state, no ack) left the
+        # row with just "exit 9" while the tail lived only in the exception the log
+        # consumed — the operator reading the ledger later saw no evidence at all.
+        # The tail is already bounded and redacted by external_worker_stderr_tail.
+        stderr_tail = ""
+        if stderr_path is not None:
+            from cron.scheduler_diagnostics import external_worker_stderr_tail
+
+            stderr_tail = external_worker_stderr_tail(stderr_path)
+        reason = f"{external_worker_exited_reason(returncode, adopted=adopted)}{stderr_tail}"
         if not terminalize_dead_owner(execution_id, reason=reason):
-            recover_interrupted_executions()
+            recover_interrupted_executions(reason=reason)
             # A concurrent sweep may have won the unknown transition. Still
             # surface this waiter's exit code and stderr instead of discarding them.
             current = get_execution(execution_id)
             if current and current.get("status") in ("completed", "failed"):
                 return True
-        stderr_tail = ""
-        if stderr_path is not None:
-            from cron.scheduler_diagnostics import external_worker_stderr_tail
-
-            stderr_tail = external_worker_stderr_tail(stderr_path)
-        raise RuntimeError(f"{reason}{stderr_tail}")
+        raise RuntimeError(reason)
 
 
 class _ExternalWorkerPostHandoffError(RuntimeError):
@@ -3662,6 +3676,118 @@ def _wait_for_external_cron_worker(
                 pass
 
 
+# Cron worker interpreter resolver.  Some installs start Hermes through a
+# bootstrap shim (``-I`` plus a sys.path insert) with an interpreter binary
+# whose own site-packages carries none of Hermes' dependencies: the parent
+# works because the bootstrap fixed its path, but a child spawned as
+# ``[sys.executable, "-m", "cron.scheduler"]`` does not inherit the shim and
+# dies on its first repo import.  Probe candidate interpreters by BEHAVIOUR,
+# never by name or path — a binary is healthy iff it can actually run the
+# worker's first imports under the child's real launch conditions — and let
+# the operator pin one explicitly with HERMES_CRON_PYTHON.
+#
+# A rejected candidate must never go silent (see _log_rejected_worker_interpreters):
+# an operator-set override that fails its probe and vanishes without a word is
+# indistinguishable from an override that ran, and an instrument that never
+# ran leaves no evidence behind.
+_CRON_WORKER_PYTHON_OK: list[str] = []
+
+
+def _cron_repo_root() -> str:
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+
+def _interpreter_is_healthy(candidate: str) -> bool:
+    """Probe under the child's REAL launch conditions: cwd=repo root, stripped
+    env (no PYTHONPATH/PYTHONHOME), importing what the worker imports first."""
+    try:
+        child_env = dict(os.environ)
+        child_env.pop("PYTHONPATH", None)
+        child_env.pop("PYTHONHOME", None)
+        r = subprocess.run(
+            [candidate, "-c", "import cron, ruamel.yaml, httpx"],
+            capture_output=True,
+            timeout=20,
+            cwd=_cron_repo_root(),
+            env=child_env,
+        )
+        return r.returncode == 0
+    except (OSError, subprocess.TimeoutExpired):
+        return False
+
+
+def _log_rejected_worker_interpreters(
+    rejected: list[tuple[str, str, bool]], chosen: Optional[str]
+) -> None:
+    """Say the truth about every candidate the resolver dropped.
+
+    ``rejected`` is ``[(path, reason, was_operator_override), ...]`` in
+    consultation order; ``chosen`` is the interpreter that passed, or ``None``
+    when none did.  The silent-drop case this closes: an earlier candidate
+    passes and the rejected ones — possibly HERMES_CRON_PYTHON, the operator's
+    own pin — are discarded without a word, so an instrument the operator
+    deployed reads as "ran, found nothing" when the truth is "never ran."
+    Silence is not evidence; the drop itself is the event worth logging.
+    """
+    for path, reason, is_override in rejected:
+        if is_override:
+            logger.warning(
+                "cron worker interpreter: operator override HERMES_CRON_PYTHON=%s "
+                "%s — the override is NOT in effect%s",
+                path,
+                reason,
+                f"; worker will run on {chosen}" if chosen else " (no candidate passed)",
+            )
+        else:
+            logger.warning(
+                "cron worker interpreter: candidate %s %s%s",
+                path,
+                reason,
+                f" (chosen: {chosen})" if chosen else "",
+            )
+
+
+def _cron_worker_python() -> str:
+    if _CRON_WORKER_PYTHON_OK:
+        return _CRON_WORKER_PYTHON_OK[0]
+    override = os.environ.get("HERMES_CRON_PYTHON", "").strip()
+    candidates: list[str] = []
+    if override:
+        candidates.append(override)
+    candidates.append(sys.executable)
+    candidates.append(os.path.join(_cron_repo_root(), "venv", "bin", "python"))
+    seen: set[str] = set()
+    rejected: list[tuple[str, str, bool]] = []
+    for cand in candidates:
+        # A duplicate (override naming a later candidate) was consulted under
+        # its own entry — dropping it silently is honest, not an erasure.
+        if not cand or cand in seen:
+            continue
+        seen.add(cand)
+        is_override = cand == override
+        if not os.path.exists(cand):
+            rejected.append((cand, "does not exist", is_override))
+            continue
+        if _interpreter_is_healthy(cand):
+            _CRON_WORKER_PYTHON_OK.append(cand)
+            # A candidate that passed while earlier ones were dropped is the
+            # silent-drop case: the old code returned here without speaking.
+            # The all-failed raise below was always loud; the warning belongs
+            # on the SUCCESS path.  Log before returning so every drop is
+            # named even when resolution succeeds.
+            _log_rejected_worker_interpreters(rejected, chosen=cand)
+            return cand
+        rejected.append((cand, "failed the health probe", is_override))
+    _log_rejected_worker_interpreters(rejected, chosen=None)
+    raise RuntimeError(
+        "cron worker launch refused: no healthy interpreter among ["
+        + ", ".join(f"{path} ({reason})" for path, reason, _ in rejected)
+        + "] (probe: import cron, ruamel.yaml, httpx under cwd=repo root). "
+        "Install is broken; set HERMES_CRON_PYTHON to a Python carrying "
+        "Hermes' dependencies."
+    )
+
+
 def _launch_external_cron_worker(job: dict) -> bool:
     """Launch *job* outside the managed gateway process when required.
 
@@ -3684,8 +3810,9 @@ def _launch_external_cron_worker(job: dict) -> bool:
     ack_path = handoff_dir / f"{execution_id}.ready"
     # Captured so a worker that dies before its acknowledgement can name the cause (#112729).
     stderr_path = handoff_dir / f"{execution_id}.stderr"
+    worker_python = _cron_worker_python()
     command = [
-        sys.executable,
+        worker_python,
         "-m",
         "cron.scheduler",
         "--external-worker-file",
@@ -4219,9 +4346,9 @@ def _maybe_reap_dead_owners() -> None:
         return
     _last_dead_owner_reap_at[_reap_key] = _reap_now
     try:
-        from cron.executions import recover_interrupted_executions
+        from cron.executions import _PERIODIC_REAP_REASON, recover_interrupted_executions
 
-        _reclaimed = recover_interrupted_executions()
+        _reclaimed = recover_interrupted_executions(reason=_PERIODIC_REAP_REASON)
         if _reclaimed:
             logger.warning(
                 "Reclaimed %d cron execution(s) whose owner process died "

cron/scheduler_provider.py

@@ -171,9 +171,10 @@ def register_job(self, job: dict[str, Any]) -> None:
 
     def recover_interrupted(self) -> int:
         """Run profile-local attempt recovery for every provider lifecycle."""
-        from cron.executions import recover_interrupted_executions
+        from cron.executions import _OWNER_GONE_REASON, recover_interrupted_executions
 
-        return recover_interrupted_executions()
+        # Every caller of this wrapper is a ticker-startup recovery: the restart text is true here.
+        return recover_interrupted_executions(reason=_OWNER_GONE_REASON)
 
     @property
     def supports_force_fire(self) -> bool:

cron/scheduler_worker_failure.py

@@ -45,17 +45,44 @@ def record_unknown_worker_outcome(
     return True
 
 
-def external_worker_exited_reason(returncode: int | None) -> str:
+def external_worker_exited_reason(returncode: int | None, *, adopted: bool | None) -> str:
     """Cause for an attempt whose restart-safe worker this process watched exit.
 
     The generic dead-owner sweep can only say the owner vanished, so it names a
-    scheduler restart — but this waiter never lost its scheduler: it handed the
+    scheduler restart -- but this waiter never lost its scheduler: it handed the
     attempt to the external worker and then saw that worker exit with this status.
     The run is still ``unknown`` rather than ``failed`` (whether side effects ran is
     still unknown); only the cause becomes truthful (#128509).
+
+    ``adopted`` is REQUIRED and has no default: it is a factual claim about a
+    phase the caller must have observed, and a default would let a caller
+    silently assert a claim it cannot substantiate. Required keyword-only mirrors
+    the seam #128509 set with ``terminalize_dead_owner(..., reason=...)`` and
+    ``recover_interrupted_executions(reason=...)``.
+
+    ``adopted`` is the phase read from the ledger row this waiter re-checked after
+    the worker exited — never a guess. ``running`` means the worker won the
+    claimed→running adoption gate, so "after adopting" is true of it. ``claimed``
+    means it died BEFORE that gate (the observed field case: an import-time death,
+    e.g. ModuleNotFoundError in the interpreter it was handed) and cannot be said
+    to have adopted. Anything else (already terminalized by a concurrent sweep,
+    row gone) has no observable phase; the wording then asserts no phase at all
+    rather than defaulting to the post-adoption claim the evidence does not support.
     """
+    if adopted:
+        return (
+            f"Restart-safe cron worker exited with status {returncode} after adopting this "
+            "execution, without writing a durable terminal state; whether side effects ran "
+            "is unknown."
+        )
+    if adopted is None:
+        return (
+            f"Restart-safe cron worker exited with status {returncode} without writing a "
+            "durable terminal state and its adoption phase was not observable from this "
+            "waiter; whether side effects ran is unknown."
+        )
     return (
-        f"Restart-safe cron worker exited with status {returncode} after adopting this "
-        "execution, without writing a durable terminal state; whether side effects ran "
-        "is unknown."
+        f"Restart-safe cron worker exited with status {returncode} before adopting this "
+        "execution: it never won the claimed→running adoption gate, which is the contract "
+        "seam before any payload side effect may run."
     )

tests/_fixtures/env_filter.py

@@ -160,6 +160,13 @@ def _looks_like_credential(name: str) -> bool:
     "HERMES_SESSION_KEY",
     "HERMES_GATEWAY_SESSION",
     "HERMES_CRON_SESSION",
+    # The operator's cron-worker interpreter pin. Set in the gateway (and in any
+    # shell launched under it), it changes what the worker-interpreter resolver
+    # probes — and a leaked pin pointing into the REAL hermes home makes the
+    # resolver's existence probe trip the home-I/O guard in every test that
+    # reaches _launch_external_cron_worker. Tests that exercise the override do
+    # so explicitly via monkeypatch.setenv.
+    "HERMES_CRON_PYTHON",
     "_HERMES_GATEWAY",
     "HERMES_PLATFORM",
     "HERMES_MODEL",

tests/cron/test_dead_owner_claim_reclaim.py

@@ -120,7 +120,7 @@ def test_reap_is_throttled_between_ticks(self, monkeypatch, executions):
         calls = []
         monkeypatch.setattr(
             "cron.executions.recover_interrupted_executions",
-            lambda: calls.append(1) or 0,
+            lambda **_kwargs: calls.append(1) or 0,
         )
 
         _run_tick()
@@ -137,7 +137,7 @@ def test_reap_is_throttled_between_ticks(self, monkeypatch, executions):
         assert len(calls) == 2, "an expired throttle window must reap again"
 
     def test_reap_failure_does_not_break_the_tick(self, monkeypatch):
-        def _boom():
+        def _boom(**_kwargs):
             raise RuntimeError("ledger unavailable")
 
         monkeypatch.setattr(
@@ -215,7 +215,7 @@ def test_reap_throttle_is_profile_scoped(monkeypatch, tmp_path):
 
     calls = []
     monkeypatch.setattr(
-        executions_mod, "recover_interrupted_executions", lambda: calls.append(1) or 0
+        executions_mod, "recover_interrupted_executions", lambda **_kwargs: calls.append(1) or 0
     )
     home_a = tmp_path / "profile-a"
     home_b = tmp_path / "profile-b"