diff --git a/cron/scheduler.py b/cron/scheduler.py index 624efc48f9..b67853200c 100644 --- a/cron/scheduler.py +++ b/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 diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index a383e52610..4120ab62c3 100644 --- a/hermes_cli/cron.py +++ b/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 diff --git a/tests/cron/test_dead_owner_claim_reclaim.py b/tests/cron/test_dead_owner_claim_reclaim.py new file mode 100644 index 0000000000..89dfee820e --- /dev/null +++ b/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 `` 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