diff --git a/tests/tools/test_cronjob_run_background.py b/tests/tools/test_cronjob_run_background.py new file mode 100644 index 0000000000000..cdba7ed13c02f --- /dev/null +++ b/tests/tools/test_cronjob_run_background.py @@ -0,0 +1,218 @@ +"""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 _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(dict(_JOB)) + + 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-1") + # 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(dict(_JOB)) + 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(dict(_JOB)) + 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(dict(_JOB)) + 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(dict(_JOB)) + 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(dict(_JOB)) + 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(dict(_JOB)) + assert res["dispatched"] is False + assert res["success"] is True + m_run.assert_called_once() # ran inline on this thread + + +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=dict(_JOB)), \ + 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-1", "name": "bg run", + "last_status": "ok", "last_error": None}): + out = json.loads(cronjob(action="run", job_id="job-bg-1")) + + 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=dict(_JOB)), \ + 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-1")) + + 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-1") + m_run.assert_called_once() diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 5f1a56d258054..f86db92c1a995 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -607,8 +607,6 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: """ job_id = job["id"] try: - from cron.scheduler import run_one_job - # At-most-once claim: bail without running if a tick/other fire owns it. if not claim_job_for_fire(job_id): # claim_job_for_fire returns False for paused/disabled/missing @@ -623,6 +621,30 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: else: reason = "Job is already being fired by the scheduler; not run again." return {"claimed": False, "success": False, "error": reason} + except Exception as e: + logger.error("Failed to claim cron job %s for immediate run: %s", job_id, e) + try: + mark_job_run(job_id, False, str(e)) + except Exception: + pass + return {"claimed": True, "success": False, "error": str(e)} + + return _run_claimed_job(job) + + +def _run_claimed_job(job: Dict[str, Any]) -> Dict[str, Any]: + """Fire an already-claimed job through the shared ``run_one_job`` body. + + Split out of ``_execute_job_now`` so the background dispatch path + (``_try_dispatch_background_run``) can take the claim synchronously — so + the tool response can report "paused"/"already firing" immediately — and + hand the actual run to a daemon worker. + + Returns {"claimed": True, "success": bool, "error": str|None}. + """ + job_id = job["id"] + try: + from cron.scheduler import run_one_job # run_one_job records last_run_at/last_status via mark_job_run (which # also clears the fire claim) and returns True iff it processed the job. @@ -707,6 +729,219 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: return {"claimed": True, "success": False, "error": str(e)} +def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[str]: + """Best-effort excerpt of the job's most recent saved output file. + + Included in the background-run completion block so the parent agent sees + what the job actually produced without having to dig through + ``~/.hermes/cron/output/``. Never raises. + """ + try: + from cron.jobs import get_cron_output_dir + + out_dir = get_cron_output_dir() / job_id + files = sorted(out_dir.glob("*.md")) + if not files: + return None + text = files[-1].read_text(encoding="utf-8", errors="replace").strip() + if not text: + return None + if len(text) > max_chars: + text = text[:max_chars] + f"\n… (truncated; full output: {files[-1]})" + return text + except Exception: + return None + + +def _try_dispatch_background_run( + job: Dict[str, Any], session_id: Optional[str] = None +) -> Optional[Dict[str, Any]]: + """Claim ``job`` now, then fire it on the async-delegation daemon executor. + + A manual ``cronjob(action='run')`` used to execute the job synchronously + on the calling agent's tool thread. A cron job is a full agent run that + routinely takes minutes-to-hours, so the parent turn sat inside ONE tool + call the whole time: uninterruptible (the interrupt flag is only checked + between loop iterations) and serial (a batch of runs executed one by one). + + This dispatches the run like ``delegate_task``'s background mode: the tool + returns immediately with a handle, the run executes on the shared async + daemon executor, and a ``type="async_delegation"`` completion event + re-enters the conversation as a fresh turn when the job finishes — riding + the existing completion-queue rail (CLI drain + gateway watcher), which + keeps message-role alternation legal and the prompt cache intact. + + The at-most-once claim is taken SYNCHRONOUSLY before dispatch so + unrunnable jobs (paused / missing / already firing) report in the tool + response immediately instead of as a delayed completion event. + + Returns + ------- + None + Background delivery unavailable on this session runtime (one-shot + ``hermes -z``, stateless HTTP, Kanban worker, nested cron run). + Caller falls back to the synchronous path unchanged. + dict + ``{"claimed": False, "success": False, "error": ...}`` — claim lost; + same shape as ``_execute_job_now`` so the caller's existing response + formatting applies. + ``{"claimed": True, "dispatched": True, "delegation_id": ...}`` — + run is executing in the background. + ``{"claimed": True, "dispatched": False, "success": ..., "error": ...}`` + — dispatch pool was at capacity; the run executed inline (the claim + was already taken and must not be stranded). + """ + # Finite sessions cannot route a detached result back after the turn + # ends — mirror delegate_task's gate and fall back to sync execution. + try: + from gateway.session_context import async_delivery_supported + + if not async_delivery_supported(): + return None + except Exception: + pass + + job_id = job["id"] + job_name = str(job.get("name") or job_id) + + # ---- routing capture (on THIS thread; contextvars don't cross the pool) ---- + # Resolved BEFORE the claim: with no routable session there is no durable + # consumer for a detached completion, so we must not claim-and-dispatch. + try: + from tools.approval import get_current_session_key + + session_key = get_current_session_key(default="") + except Exception: + session_key = "" + if not session_key and session_id: + # CLI path: the approval contextvar is only bound during gateway/TUI + # turns. The CLI drain filters completions by the durable agent + # session id (#64240), so stamp it as the key — an empty key would + # fail closed and the completion could never be claimed. + session_key = str(session_id) + if not session_key: + # Direct Python callers (`hermes cron run`, tests) have no agent + # session to deliver a completion to — the process exits right after + # the tool returns. Run synchronously. + return None + + # ---- synchronous claim (same semantics as _execute_job_now) ---- + try: + if not claim_job_for_fire(job_id): + refreshed = get_job(job_id) + if refreshed is None: + reason = "Job no longer exists; nothing to run." + elif not refreshed.get("enabled", True) or refreshed.get("state") == "paused": + reason = "Job is paused/disabled; resume it before running." + else: + reason = "Job is already being fired by the scheduler; not run again." + return {"claimed": False, "success": False, "error": reason} + except Exception as e: + logger.error("Failed to claim cron job %s for background run: %s", job_id, e) + try: + mark_job_run(job_id, False, str(e)) + except Exception: + pass + return {"claimed": True, "dispatched": False, "success": False, "error": str(e)} + + origin_ui_session_id = "" + try: + from gateway.session_context import get_session_env + + origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "") or "" + except Exception: + pass + + try: + from tools.async_delegation import ( + _current_origin_session_id, + dispatch_async_delegation, + ) + + origin_session_id = _current_origin_session_id() + except Exception as e: + logger.warning( + "cronjob run: async delegation registry unavailable (%s); " + "running job '%s' inline.", e, job_name, + ) + result = _run_claimed_job(job) + result["dispatched"] = False + return result + + try: + from tools.delegate_tool import _get_max_async_children + + max_async = _get_max_async_children() + except Exception: + max_async = 3 + + started_at = time.time() + deliver = job.get("deliver", "local") + + def _runner() -> Dict[str, Any]: + res = _run_claimed_job(job) + duration = round(time.time() - started_at, 2) + refreshed = get_job(job_id) or {} + lines = [ + f"Cron job '{job_name}' ({job_id}) finished its manual run.", + f"Result: {'ok' if res.get('success') else 'FAILED'}" + + (f" — {res.get('error')}" if res.get("error") else ""), + f"Delivery target: {deliver}" + + ( + " (output was delivered there by the job itself)" + if deliver != "local" + else " (output saved locally only)" + ), + ] + if refreshed.get("next_run_at"): + lines.append(f"Next scheduled run: {refreshed['next_run_at']}") + excerpt = _latest_job_output_excerpt(job_id) + if excerpt: + lines.append("--- JOB OUTPUT ---") + lines.append(excerpt) + return { + "status": "completed" if res.get("success") else "error", + "summary": "\n".join(lines), + "error": res.get("error"), + "api_calls": 0, + "duration_seconds": duration, + } + + dispatch = dispatch_async_delegation( + goal=f"Manual run of cron job '{job_name}' ({job_id})", + context=( + "Triggered via cronjob(action='run'). The job executed in its own " + "fresh cron session; this block reports its outcome." + ), + toolsets=None, + role="cron_run", + model=job.get("model"), + session_key=session_key, + parent_session_id=str(session_id) if session_id else None, + runner=_runner, + origin_ui_session_id=origin_ui_session_id, + origin_session_id=origin_session_id, + max_async_children=max_async, + ) + + if dispatch.get("status") == "dispatched": + return { + "claimed": True, + "dispatched": True, + "delegation_id": dispatch.get("delegation_id"), + } + + # Pool at capacity (or submit failure): the claim is already taken and + # must not be stranded — run inline exactly as the legacy path did. + logger.info( + "cronjob run: background pool unavailable (%s); running job '%s' inline.", + dispatch.get("error", "rejected"), job_name, + ) + result = _run_claimed_job(job) + result["dispatched"] = False + return result + + def cronjob( action: str, job_id: Optional[str] = None, @@ -729,6 +964,7 @@ def cronjob( no_agent: Optional[bool] = None, attach_to_session: Optional[bool] = None, task_id: str = None, + session_id: Optional[str] = None, ) -> str: """Unified cron job management tool.""" del task_id # unused but kept for handler signature compatibility @@ -889,10 +1125,42 @@ def cronjob( if normalized in {"run", "run_now", "trigger"}: # Execute the job immediately rather than only scheduling it for the # next scheduler tick — a manual `run` should actually run, even when - # no gateway/ticker is active (the #41037 case). The claim inside - # _execute_job_now advances next_run_at and blocks a concurrent tick - # from double-firing. - exec_result = _execute_job_now(job) + # no gateway/ticker is active (the #41037 case). The claim (taken + # inside both paths below) advances next_run_at and blocks a + # concurrent tick from double-firing. + # + # Preferred path: dispatch the run to the background like + # delegate_task — the tool returns a handle immediately and the + # job's outcome re-enters the conversation as a completion event. + # A cron job is a full agent run (minutes to hours); executing it + # inline made the parent turn uninterruptible and serialized + # batches of manual runs (#80xxx — the "stuck Telegram session" + # incident). Falls back to inline execution when the session + # runtime can't receive detached completions. + bg = _try_dispatch_background_run(job, session_id=session_id) + if bg is not None and bg.get("dispatched"): + _notify_provider_jobs_changed_safe() + result = _format_job(get_job(job_id) or {"id": job_id}) + result["executed"] = True + result["execution_mode"] = "background" + result["delegation_id"] = bg.get("delegation_id") + return json.dumps( + { + "success": True, + "job": result, + "note": ( + "The job is running in the background. You and the " + "user can keep working; its outcome re-enters the " + "conversation as a new message when it finishes. " + "Do not wait or poll — just continue." + ), + }, + indent=2, + ) + # bg carries a terminal result (claim lost, or inline fallback + # after pool rejection); None means background delivery is + # unsupported here — run synchronously as before. + exec_result = bg if bg is not None else _execute_job_now(job) # A claimed direct run advances next_run_at and may race the # external one-shot for the same occurrence. If Chronos loses that # claim, its consumed fire cannot re-arm itself; reconcile from the @@ -1032,6 +1300,8 @@ Use action='create' to schedule a new job from a prompt or one or more skills. Use action='list' to inspect jobs. Use action='update', 'pause', 'resume', 'remove', or 'run' to manage an existing job. +action='run' fires the job immediately in the BACKGROUND (like delegate_task): the call returns at once with a handle and the job's outcome re-enters the conversation as a new message when it finishes. Do not wait or poll after triggering a run — just continue. + To stop a job the user no longer wants: first action='list' to find the job_id, then action='remove' with that job_id. Never guess job IDs — always list first. Jobs run in a fresh session with no current-chat context, so prompts must be self-contained. @@ -1185,6 +1455,7 @@ registry.register( workdir=args.get("workdir"), no_agent=args.get("no_agent"), task_id=kw.get("task_id"), + session_id=kw.get("session_id"), ), check_fn=check_cronjob_requirements, emoji="⏰",