fix: dispatch cronjob(action='run') to the background like delegate_task
A manual cronjob run executed 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 manual runs executed one by one). A Telegram session that kicked off dozens of new jobs 'right now' was wedged for hours ignoring every interrupt. action='run' now rides the async-delegation rail delegate_task background mode uses: the at-most-once claim is taken synchronously (so paused/missing/already-firing jobs still report immediately), the run executes on the shared daemon executor, the tool returns at once with a delegation handle, and the job's outcome re-enters the conversation as a type='async_delegation' completion event through the existing completion-queue drains (CLI + gateway) — preserving message-role alternation and the prompt cache. Sync fallbacks preserved: - no routable session (direct Python callers, hermes cron run) - async delivery unsupported (hermes -z, cron child sessions, Kanban workers, stateless HTTP) - dispatch pool at capacity (claim already taken — runs inline rather than stranding it) The completion block reports ok/failure, delivery target, next scheduled run, and an excerpt of the job's saved output.
This commit is contained in:
Teknium
parent
32e7fb07a0
commit
7ab42dda60
|
|
@ -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()
|
||||
|
|
@ -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="⏰",
|
||||
|
|
|
|||
Loading…
Reference in New Issue