fix(cron): stop one-shot CLI cron run from orphaning the job; reap dead-owner claims on tick

`hermes cron run <job_id>` from a one-shot CLI invocation could
background-dispatch the run onto a daemon thread of the calling process
(when the CLI inherited a gateway/desktop session env and resolved a
session key). The CLI printed "Triggered job: ..." and exited instantly,
killing the runner mid-LLM-call: the async delegation died with
state='unknown' and the job's row in cron/executions.db stayed
status='claimed' forever, blocking every subsequent run of that job.

Two-part fix:

1. hermes_cli/cron.py: `_job_action("run", ...)` declares the delivery
   channel stateless (scoped ContextVar set/reset around the call) before
   invoking the cron API, so `async_delivery_supported()` gates off
   `_try_dispatch_background_run` and the run executes synchronously to
   completion in the CLI process — the same behavior `hermes -z` already
   gets via declare_stateless_channel().

2. cron/scheduler.py: tick() now periodically invokes
   recover_interrupted_executions() (previously only run at scheduler
   startup), so execution rows whose exact owner process is provably dead
   (pid + process start time check in _owner_is_live) are reaped to
   'unknown' by the long-lived gateway ticker without a restart.
   Throttled to once per 300s so idle 60s ticks don't pay a ledger
   connection every cycle.

Tests: tests/cron/test_dead_owner_claim_reclaim.py covers the dead-owner
reap (real dead pid via a finished subprocess), live-owner rows surviving
the reap, throttle behavior, reap-failure isolation, the CLI stateless
gate (including restoration after the call), and the end-to-end refusal
of background dispatch under a stateless channel.

Fixes #86721

cron/scheduler.py

@@ -6303,6 +6303,14 @@ def create_job_with_scheduler_registration(**kwargs) -> dict:
     return job
 
 
+# Dead-owner claim reclaim throttle (#86721): recover_interrupted_executions
+# opens the executions ledger, so the per-tick reap is rate-limited rather
+# than run on every idle 60s cycle. Tests may reset _last_dead_owner_reap_at
+# to None to force a reap on the next tick.
+_DEAD_OWNER_REAP_INTERVAL_SECONDS = 300.0
+_last_dead_owner_reap_at: Optional[float] = None
+
+
 def tick(
     verbose: bool = True,
     adapters=None,
@@ -6360,6 +6368,37 @@ def tick(
             logger.debug("Cron dispatch paused while gateway drains existing work")
             return 0
 
+        # Dead-owner claim reclaim (#86721): execution rows carry their owner
+        # pid + process start time, but recovery previously ran only at
+        # scheduler STARTUP. A one-shot `hermes cron run` that claimed a job
+        # and died mid-run (its runner thread lived in the exiting CLI
+        # process) left the row 'claimed' forever while the long-lived
+        # gateway ticker kept running — blocking every future run of that
+        # job. Reap provably-dead owners periodically so stale claims
+        # auto-clear without a gateway restart. Only rows whose exact owner
+        # process is proved gone are touched (see _owner_is_live), so live
+        # runs in other processes are never rewritten. Throttled so idle
+        # 60s ticks don't pay a ledger connection every cycle (#33612).
+        global _last_dead_owner_reap_at
+        _reap_now = time.monotonic()
+        if (
+            _last_dead_owner_reap_at is None
+            or _reap_now - _last_dead_owner_reap_at >= _DEAD_OWNER_REAP_INTERVAL_SECONDS
+        ):
+            _last_dead_owner_reap_at = _reap_now
+            try:
+                from cron.executions import recover_interrupted_executions
+
+                _reclaimed = recover_interrupted_executions()
+                if _reclaimed:
+                    logger.warning(
+                        "Reclaimed %d cron execution(s) whose owner process died "
+                        "before reaching a terminal state (marked unknown)",
+                        _reclaimed,
+                    )
+            except Exception as _reap_exc:
+                logger.debug("Dead-owner execution reclaim failed: %s", _reap_exc)
+
         due_jobs = get_due_jobs()
 
         # Bound the in-flight set BEFORE the dedup guard is consulted, so a

hermes_cli/cron.py

@@ -462,7 +462,32 @@ def cron_edit(args):
 
 
 def _job_action(action: str, job_id: str, success_verb: str) -> int:
-    result = _cron_api(action=action, job_id=job_id)
+    _stateless_reset = None
+    if action == "run":
+        # One-shot CLI: this process exits as soon as the command returns, so
+        # a background-dispatched run (daemon thread of THIS process) would be
+        # orphaned mid-LLM-call — the delegation dies 'unknown' and the job's
+        # execution row is stuck 'claimed', blocking future runs (#86721).
+        # The background path in ``_try_dispatch_background_run`` triggers when
+        # the CLI inherits a gateway/desktop session env (HERMES_SESSION_KEY);
+        # declare the channel stateless so ``async_delivery_supported()`` gates
+        # it off and the run executes synchronously to completion instead.
+        # The declaration is scoped to this call (token reset in ``finally``)
+        # so in-process callers (tests, embedding apps) are not tainted.
+        try:
+            from gateway.session_context import _SESSION_ASYNC_DELIVERY
+
+            _stateless_token = _SESSION_ASYNC_DELIVERY.set(False)
+
+            def _stateless_reset() -> None:
+                _SESSION_ASYNC_DELIVERY.reset(_stateless_token)
+        except Exception:
+            _stateless_reset = None
+    try:
+        result = _cron_api(action=action, job_id=job_id)
+    finally:
+        if _stateless_reset is not None:
+            _stateless_reset()
     if not result.get("success"):
         print(color(f"Failed to {action} job: {result.get('error', 'unknown error')}", Colors.RED))
         return 1

tests/cron/test_dead_owner_claim_reclaim.py

@@ -0,0 +1,205 @@
+"""Dead-owner cron claim reclaim + one-shot CLI `cron run` sync gate (#86721).
+
+A one-shot ``hermes cron run <job_id>`` used to background-dispatch the run
+onto a daemon thread of the calling process when the CLI inherited a
+gateway/desktop session env. The process exited immediately, the runner died
+mid-LLM-call, and the job's execution row stayed ``claimed`` forever —
+blocking every future run.
+
+Two-part fix under test here:
+
+1. ``hermes_cli.cron._job_action("run", ...)`` declares the channel stateless
+   before invoking the cron API, so the background-dispatch path is gated off
+   and the run executes synchronously to completion in the CLI process.
+2. ``cron.scheduler.tick`` periodically reaps execution rows whose owner
+   process is provably dead (``recover_interrupted_executions``), so a stale
+   ``claimed`` row from a crashed/exited owner auto-clears without a gateway
+   restart.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import time
+from unittest.mock import patch
+
+import pytest
+
+import cron.scheduler as scheduler_mod
+
+
+@pytest.fixture()
+def executions(monkeypatch, tmp_path):
+    import cron.executions as executions_mod
+
+    monkeypatch.setattr(
+        executions_mod, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db"
+    )
+    return executions_mod
+
+
+@pytest.fixture(autouse=True)
+def _fresh_reap_window(monkeypatch):
+    """Each test starts with the reap throttle open."""
+    monkeypatch.setattr(scheduler_mod, "_last_dead_owner_reap_at", None)
+
+
+def _dead_pid() -> int:
+    """PID of a real process that has already exited."""
+    proc = subprocess.run(
+        [sys.executable, "-c", "import os; print(os.getpid())"],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    return int(proc.stdout.strip())
+
+
+def _orphan_claimed_row(executions, job_id: str) -> str:
+    """Persist a claimed execution owned by a process that no longer exists.
+
+    Mirrors what a one-shot ``hermes cron run`` leaves behind: a row stuck in
+    ``claimed`` whose owner pid is dead.
+    """
+    record = executions.create_execution(job_id, source="direct")
+    with executions._transaction() as conn:
+        conn.execute(
+            "UPDATE executions SET process_id='dead-cli-process', pid=?, "
+            "process_started_at=NULL WHERE id=?",
+            (_dead_pid(), record["id"]),
+        )
+    return record["id"]
+
+
+def _run_tick():
+    with (
+        patch.object(scheduler_mod, "get_due_jobs", return_value=[]),
+        patch("tools.mcp_tool._kill_orphaned_mcp_children", lambda: None),
+    ):
+        return scheduler_mod.tick(verbose=False)
+
+
+class TestTickReapsDeadOwnerClaims:
+    def test_stale_claimed_row_from_dead_owner_is_cleared_by_tick(self, executions):
+        """The exact #86721 wedge: dead-owner 'claimed' row unblocks on tick."""
+        execution_id = _orphan_claimed_row(executions, "orphaned-job")
+
+        assert _run_tick() == 0
+
+        record = executions.latest_execution("orphaned-job")
+        assert record["id"] == execution_id
+        assert record["status"] == "unknown"
+        assert record["finished_at"]
+
+    def test_running_row_from_dead_owner_is_also_reclaimed(self, executions):
+        record = executions.create_execution("orphaned-running", source="direct")
+        executions.mark_execution_running(record["id"])
+        with executions._transaction() as conn:
+            conn.execute(
+                "UPDATE executions SET process_id='dead-cli-process', pid=?, "
+                "process_started_at=NULL WHERE id=?",
+                (_dead_pid(), record["id"]),
+            )
+
+        _run_tick()
+
+        assert executions.latest_execution("orphaned-running")["status"] == "unknown"
+
+    def test_live_owner_claim_is_never_rewritten(self, executions):
+        """A claim owned by a live process (this one) must survive the reap."""
+        record = executions.create_execution("live-job", source="builtin")
+        executions.mark_execution_running(record["id"])
+
+        _run_tick()
+
+        assert executions.latest_execution("live-job")["status"] == "running"
+
+    def test_reap_is_throttled_between_ticks(self, monkeypatch, executions):
+        calls = []
+        monkeypatch.setattr(
+            "cron.executions.recover_interrupted_executions",
+            lambda: calls.append(1) or 0,
+        )
+
+        _run_tick()
+        _run_tick()
+        assert len(calls) == 1, "back-to-back ticks must not reap twice"
+
+        monkeypatch.setattr(
+            scheduler_mod,
+            "_last_dead_owner_reap_at",
+            time.monotonic() - scheduler_mod._DEAD_OWNER_REAP_INTERVAL_SECONDS - 1,
+        )
+        _run_tick()
+        assert len(calls) == 2, "an expired throttle window must reap again"
+
+    def test_reap_failure_does_not_break_the_tick(self, monkeypatch):
+        def _boom():
+            raise RuntimeError("ledger unavailable")
+
+        monkeypatch.setattr(
+            "cron.executions.recover_interrupted_executions", _boom
+        )
+
+        assert _run_tick() == 0
+
+
+class TestOneShotCliRunIsSynchronous:
+    @pytest.fixture(autouse=True)
+    def _restore_async_delivery_flag(self):
+        from gateway.session_context import _SESSION_ASYNC_DELIVERY, _UNSET
+
+        token = _SESSION_ASYNC_DELIVERY.set(_UNSET)
+        yield
+        _SESSION_ASYNC_DELIVERY.reset(token)
+
+    def test_cli_run_declares_stateless_channel_before_dispatch(self, monkeypatch):
+        """`hermes cron run` must gate off async delivery so the run executes
+        synchronously in the CLI process instead of on a doomed daemon thread."""
+        from gateway.session_context import async_delivery_supported
+        from hermes_cli import cron as cron_cli
+
+        observed = {}
+
+        def _fake_cron_api(**kwargs):
+            observed["async_delivery"] = async_delivery_supported()
+            return {"success": True, "job": {"executed": True, "execution_success": True}}
+
+        monkeypatch.setattr(cron_cli, "_cron_api", _fake_cron_api)
+
+        assert cron_cli._job_action("run", "job-123", "Triggered") == 0
+        assert observed["async_delivery"] is False
+        # Scoped declaration: the capability must be restored after the call
+        # so in-process callers (tests, embedding apps) are not tainted.
+        assert async_delivery_supported() is True
+
+    def test_non_run_actions_leave_channel_capability_alone(self, monkeypatch):
+        from gateway.session_context import async_delivery_supported
+        from hermes_cli import cron as cron_cli
+
+        observed = {}
+
+        def _fake_cron_api(**kwargs):
+            observed["async_delivery"] = async_delivery_supported()
+            return {"success": True, "job": {"name": "j"}}
+
+        monkeypatch.setattr(cron_cli, "_cron_api", _fake_cron_api)
+
+        cron_cli._job_action("pause", "job-123", "Paused")
+        assert observed["async_delivery"] is True
+
+    def test_background_dispatch_refused_when_channel_stateless(self, monkeypatch):
+        """End-to-end gate: with the stateless declaration active, the cron
+        tool's background dispatcher must fall back to synchronous execution
+        (return None) even when a session key is inherited from a gateway env."""
+        from gateway.session_context import declare_stateless_channel
+        from tools.cronjob_tools import _try_dispatch_background_run
+
+        declare_stateless_channel()
+        monkeypatch.setenv("HERMES_SESSION_KEY", "inherited-gateway-session")
+
+        result = _try_dispatch_background_run(
+            {"id": "job-x", "name": "job-x"}, session_id="sess-1"
+        )
+        assert result is None