fix: dispatch cronjob(action='run') to the background like delegate_task

cron/scheduler.py

@@ -363,6 +363,35 @@ def get_running_job_ids() -> "frozenset[str]":
         return frozenset(_running_job_ids)
 
 
+def try_register_running_job(job_id: str) -> bool:
+    """Atomically add ``job_id`` to the in-flight running set.
+
+    Returns False (without registering) when the job is already mid-run —
+    the caller must skip the fire. This is the single dedupe owner shared by
+    the ticker's ``_submit_with_guard`` and manual runs
+    (``tools/cronjob_tools``): the fire claim alone cannot prevent a
+    double-fire because its TTL (300s) is routinely outlived by real jobs,
+    after which a manual ``cronjob(action='run')`` would claim successfully
+    and run the same job concurrently (idea from #53395 by @izumi0uu).
+
+    Registration also makes the run visible to ``get_running_job_ids`` (the
+    gateway shutdown drain, #60432) and ``mark_running_jobs_interrupted``.
+    Callers MUST pair a successful registration with
+    ``release_running_job`` in a ``finally`` block.
+    """
+    with _running_lock:
+        if job_id in _running_job_ids:
+            return False
+        _running_job_ids.add(job_id)
+        return True
+
+
+def release_running_job(job_id: str) -> None:
+    """Remove ``job_id`` from the in-flight running set (idempotent)."""
+    with _running_lock:
+        _running_job_ids.discard(job_id)
+
+
 def mark_running_jobs_interrupted(reason: str) -> list:
     """Best-effort: mark every currently in-flight cron job interrupted.
 
@@ -4288,11 +4317,9 @@ def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor):
                     job.get("name", job_id),
                 )
                 return None
-            with _running_lock:
-                if job_id in _running_job_ids:
-                    logger.info("Job '%s' already running — skipping", job.get("name", job_id))
-                    return None
-                _running_job_ids.add(job_id)
+            if not try_register_running_job(job_id):
+                logger.info("Job '%s' already running — skipping", job.get("name", job_id))
+                return None
             # Record the attempt before executor dispatch. Recovery classifies
             # abandoned records as unknown; it never automatically retries them.
             execution = create_execution(job_id, source="builtin")
@@ -4303,14 +4330,12 @@ def _run_and_release(j=dispatched_job, ctx=_ctx):
                 try:
                     return ctx.run(_process_job, j)
                 finally:
-                    with _running_lock:
-                        _running_job_ids.discard(j["id"])
+                    release_running_job(j["id"])
 
             try:
                 return pool.submit(_run_and_release)
             except Exception as submit_err:
-                with _running_lock:
-                    _running_job_ids.discard(job_id)
+                release_running_job(job_id)
                 finish_execution(
                     execution["id"],
                     success=False,

tests/tools/test_cronjob_run_background.py

@@ -0,0 +1,308 @@
+"""Tests for cronjob action='run' background dispatch.
+
+A manual `cronjob(action='run')` used to execute the job synchronously on the
+calling agent's tool thread — a full agent run (minutes to hours) inside ONE
+tool call, uninterruptible and serial. It now dispatches through the async
+delegation registry (same rail as delegate_task background mode): the tool
+returns immediately with a handle and the run's outcome re-enters the
+conversation as a type='async_delegation' completion event.
+
+Sync fallbacks preserved:
+  - no routable session (direct Python callers, `hermes cron run`)
+  - async delivery unsupported (one-shot runners, cron child sessions)
+  - dispatch pool at capacity (claim already taken — must not strand it)
+"""
+import json
+import threading
+from unittest.mock import patch
+
+from tools.cronjob_tools import (
+    _try_dispatch_background_run,
+    cronjob,
+)
+
+
+_JOB = {"id": "job-bg-1", "name": "bg run", "prompt": "hi",
+        "schedule": {"kind": "cron", "expr": "0 9 * * *"}}
+
+
+def _job(job_id):
+    """Per-test job dict with a UNIQUE id.
+
+    Background workers outlive their test (daemon executor) and hold the id
+    in the scheduler's shared running set until the run finishes; reusing one
+    id across tests makes the in-flight dedupe guard see a phantom
+    'already running' from a previous test's straggler worker.
+    """
+    return {"id": job_id, "name": f"bg run {job_id}", "prompt": "hi",
+            "schedule": {"kind": "cron", "expr": "0 9 * * *"}}
+
+
+def _bound_session_key(key="agent:main:telegram:dm:123"):
+    """Context manager binding the approval session key contextvar."""
+    import contextlib
+
+    from tools.approval import _approval_session_key
+
+    @contextlib.contextmanager
+    def _cm():
+        token = _approval_session_key.set(key)
+        try:
+            yield
+        finally:
+            _approval_session_key.reset(token)
+
+    return _cm()
+
+
+class TestBackgroundDispatch:
+    def test_dispatches_and_returns_handle_immediately(self):
+        """With a routable session, run claims sync then dispatches async."""
+        run_started = threading.Event()
+        run_release = threading.Event()
+
+        def slow_run_one_job(job):
+            run_started.set()
+            assert run_release.wait(timeout=5.0)
+            return True
+
+        with _bound_session_key():
+            with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True) as m_claim, \
+                 patch("cron.scheduler.run_one_job", side_effect=slow_run_one_job), \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={"last_status": "ok", "last_error": None}):
+                res = _try_dispatch_background_run(_job('job-bg-01'))
+
+        try:
+            # Returned BEFORE the job finished — that's the whole point.
+            assert res is not None
+            assert res["claimed"] is True
+            assert res["dispatched"] is True
+            assert res["delegation_id"]
+            m_claim.assert_called_once_with("job-bg-01")
+            # The job actually starts on the daemon executor.
+            assert run_started.wait(timeout=5.0), "job never started in background"
+        finally:
+            run_release.set()
+
+    def test_completion_event_reaches_shared_queue(self):
+        """The finished run pushes a type='async_delegation' event carrying
+        the job outcome onto process_registry.completion_queue."""
+        import time
+
+        from tools.process_registry import process_registry
+
+        # The runner executes on a daemon thread — the patches must stay
+        # active until the completion event lands, so poll INSIDE the blocks.
+        with _bound_session_key("agent:main:telegram:dm:777"):
+            with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+                 patch("cron.scheduler.run_one_job", return_value=True), \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={"last_status": "ok", "last_error": None,
+                                     "next_run_at": "2026-08-07T09:00:00"}):
+                res = _try_dispatch_background_run(_job('job-bg-02'))
+                assert res["dispatched"] is True
+
+                found = None
+                for _ in range(100):
+                    try:
+                        evt = process_registry.completion_queue.get_nowait()
+                    except Exception:
+                        time.sleep(0.05)
+                        continue
+                    if (evt.get("type") == "async_delegation"
+                            and evt.get("delegation_id") == res["delegation_id"]):
+                        found = evt
+                        break
+                    process_registry.completion_queue.put(evt)
+                    time.sleep(0.05)
+        assert found is not None, "completion event never reached the queue"
+        assert found["session_key"] == "agent:main:telegram:dm:777"
+        assert found["status"] == "completed"
+        assert "bg run" in (found.get("summary") or "")
+        assert "Next scheduled run" in found["summary"]
+
+    def test_failed_run_reports_error_status_in_event(self):
+        import time
+
+        from tools.process_registry import process_registry
+
+        with _bound_session_key("agent:main:telegram:dm:778"):
+            with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+                 patch("cron.scheduler.run_one_job", return_value=True), \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={"last_status": "error",
+                                     "last_error": "provider exploded"}):
+                res = _try_dispatch_background_run(_job('job-bg-03'))
+                assert res["dispatched"] is True
+
+                found = None
+                for _ in range(100):
+                    try:
+                        evt = process_registry.completion_queue.get_nowait()
+                    except Exception:
+                        time.sleep(0.05)
+                        continue
+                    if evt.get("delegation_id") == res["delegation_id"]:
+                        found = evt
+                        break
+                    process_registry.completion_queue.put(evt)
+                    time.sleep(0.05)
+        assert found is not None
+        assert found["status"] == "error"
+        assert "provider exploded" in (found.get("error") or "")
+
+    def test_claim_lost_reports_immediately_without_dispatch(self):
+        """Paused/already-firing jobs report in the tool response, not as a
+        delayed completion event."""
+        with _bound_session_key():
+            with patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={**_JOB, "enabled": False}), \
+                 patch("tools.async_delegation.dispatch_async_delegation") as m_disp:
+                res = _try_dispatch_background_run(_job('job-bg-04'))
+        assert res["claimed"] is False
+        assert "paused/disabled" in res["error"]
+        m_disp.assert_not_called()
+
+
+class TestSyncFallbacks:
+    def test_no_session_key_falls_back_to_sync(self):
+        """Direct Python callers (no agent session) keep the sync path."""
+        res = _try_dispatch_background_run(_job('job-bg-05'))
+        assert res is None
+
+    def test_async_delivery_unsupported_falls_back_to_sync(self):
+        """One-shot runtimes (hermes -z, cron child, Kanban) keep sync."""
+        with _bound_session_key():
+            with patch("gateway.session_context.async_delivery_supported",
+                       return_value=False):
+                res = _try_dispatch_background_run(_job('job-bg-06'))
+        assert res is None
+
+    def test_pool_at_capacity_runs_inline(self):
+        """A rejected dispatch must not strand the already-taken claim."""
+        with _bound_session_key():
+            with patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+                 patch("tools.async_delegation.dispatch_async_delegation",
+                       return_value={"status": "rejected", "error": "capacity"}), \
+                 patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={"last_status": "ok", "last_error": None}):
+                res = _try_dispatch_background_run(_job('job-bg-07'))
+        assert res["dispatched"] is False
+        assert res["success"] is True
+        m_run.assert_called_once()   # ran inline on this thread
+
+
+class TestInFlightDedupe:
+    """Manual runs must not double-fire a job that is already mid-run
+    (salvaged from #53395 by @izumi0uu): the fire claim's 300s TTL is
+    routinely outlived by real jobs, so the claim alone can't prevent it."""
+
+    def test_run_claimed_job_skips_when_already_running(self):
+        """The authoritative guard: _run_claimed_job refuses to fire a job
+        whose id is already registered in the scheduler running set."""
+        from cron import scheduler as sched
+        from tools.cronjob_tools import _run_claimed_job
+
+        assert sched.try_register_running_job("job-bg-08")   # simulate ticker mid-run
+        try:
+            with patch("cron.scheduler.run_one_job") as m_run:
+                res = _run_claimed_job(_job('job-bg-08'))
+            assert res["success"] is False
+            assert "already running" in res["error"]
+            m_run.assert_not_called()
+        finally:
+            sched.release_running_job("job-bg-08")
+
+    def test_run_claimed_job_registers_and_releases(self):
+        """A normal run holds the registration for run_one_job's duration and
+        releases it after — visible to get_running_job_ids mid-run."""
+        from cron import scheduler as sched
+        from tools.cronjob_tools import _run_claimed_job
+
+        seen_during_run = {}
+
+        def probe_run(job):
+            seen_during_run["registered"] = "job-bg-09" in sched.get_running_job_ids()
+            return True
+
+        with patch("cron.scheduler.run_one_job", side_effect=probe_run), \
+             patch("tools.cronjob_tools.get_job",
+                   return_value={"last_status": "ok", "last_error": None}):
+            res = _run_claimed_job(_job('job-bg-09'))
+
+        assert res["success"] is True
+        assert seen_during_run["registered"] is True
+        assert "job-bg-09" not in sched.get_running_job_ids()   # released after
+
+    def test_background_dispatch_reports_running_job_immediately(self):
+        """The dispatch path pre-checks the running set so a mid-run job
+        reports in the tool response, not as a delayed completion event."""
+        from cron import scheduler as sched
+
+        assert sched.try_register_running_job("job-bg-10")
+        try:
+            with _bound_session_key():
+                with patch("tools.cronjob_tools.claim_job_for_fire") as m_claim, \
+                     patch("tools.async_delegation.dispatch_async_delegation") as m_disp:
+                    res = _try_dispatch_background_run(_job('job-bg-10'))
+            assert res["claimed"] is False
+            assert "already running" in res["error"]
+            m_claim.assert_not_called()   # no claim consumed for a skipped run
+            m_disp.assert_not_called()
+        finally:
+            sched.release_running_job("job-bg-10")
+
+    def test_ticker_guard_uses_shared_helpers(self):
+        """The ticker's _submit_with_guard and manual runs share ONE dedupe
+        owner: registration through either side blocks the other."""
+        from cron import scheduler as sched
+
+        # Manual-run registration…
+        assert sched.try_register_running_job("job-shared-1")
+        try:
+            # …is exactly what the ticker-side helper consults.
+            assert not sched.try_register_running_job("job-shared-1")
+            assert "job-shared-1" in sched.get_running_job_ids()
+        finally:
+            sched.release_running_job("job-shared-1")
+        assert "job-shared-1" not in sched.get_running_job_ids()
+        # Idempotent release: never raises on a non-member.
+        sched.release_running_job("job-shared-1")
+
+
+class TestCronjobRunToolIntegration:
+    def test_run_action_returns_background_note(self):
+        """cronjob(action='run') surfaces the handle + do-not-wait note."""
+        with _bound_session_key():
+            with patch("tools.cronjob_tools.resolve_job_ref", return_value=_job('job-bg-12')), \
+                 patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \
+                 patch("cron.scheduler.run_one_job", return_value=True), \
+                 patch("tools.cronjob_tools.get_job",
+                       return_value={"id": "job-bg-12", "name": "bg run",
+                                     "last_status": "ok", "last_error": None}):
+                out = json.loads(cronjob(action="run", job_id="job-bg-12"))
+
+        assert out["success"] is True
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_mode"] == "background"
+        assert out["job"]["delegation_id"]
+        assert "background" in out["note"]
+
+    def test_run_action_sync_path_unchanged_without_session(self):
+        """No session context → the legacy synchronous behavior (executed +
+        execution_success populated from the completed run)."""
+        ran = {"job": "after-run", "last_status": "ok", "last_error": None}
+        with patch("tools.cronjob_tools.resolve_job_ref", return_value=_job('job-bg-13')), \
+             patch("tools.cronjob_tools.claim_job_for_fire", return_value=True) as m_claim, \
+             patch("cron.scheduler.run_one_job", return_value=True) as m_run, \
+             patch("tools.cronjob_tools.get_job", return_value=ran):
+            out = json.loads(cronjob(action="run", job_id="job-bg-13"))
+
+        assert out["success"] is True
+        assert out["job"]["executed"] is True
+        assert out["job"]["execution_success"] is True
+        m_claim.assert_called_once_with("job-bg-13")
+        m_run.assert_called_once()