diff --git a/cron/scheduler.py b/cron/scheduler.py index ec865b75aed86..477189ef5041a 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2484,7 +2484,11 @@ def _parse_wake_gate(script_output: str) -> bool: return gate.get("wakeAgent", True) is not False -def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str: +def _build_job_prompt( + job: dict, + prerun_script: Optional[tuple] = None, + extra_prompt: Optional[str] = None, +) -> str: """Build the effective prompt for a cron job, optionally loading one or more skills first. Args: @@ -2494,8 +2498,14 @@ def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str: When provided, the script is not re-executed and the cached result is used for prompt injection. When omitted, the script (if any) runs inline as before. + extra_prompt: Optional per-run context (from ``cronjob(action='run')``, + #57331 — salvaged from #57342 by @liuhao1024). Appended to the + stored prompt under a ``## Run Context`` header for this single + fire only — never persisted to the job definition. """ user_prompt = str(job.get("prompt") or "") + if extra_prompt: + user_prompt = f"{user_prompt}\n\n## Run Context\n{extra_prompt}" prompt = user_prompt skills = job.get("skills") # True when runtime-collected DATA (script stdout, upstream-job output) @@ -2806,7 +2816,8 @@ def _guard_job_credential_exfil(job: dict) -> None: def run_job( - job: dict, *, defer_agent_teardown: Optional[list] = None + job: dict, *, defer_agent_teardown: Optional[list] = None, + extra_prompt: Optional[str] = None, ) -> tuple[bool, str, str, Optional[str]]: """ Execute a single cron job. @@ -2821,6 +2832,10 @@ def run_job( guard). When ``None`` (the default) teardown happens inline as before, so every existing caller is unchanged. + ``extra_prompt``: optional per-run context from ``cronjob(action='run', + prompt=...)`` (#57331). Appended to the stored prompt for this fire only — + never persisted to the job definition. + Returns: Tuple of (success, full_output_doc, final_response, error_message) """ @@ -3029,7 +3044,9 @@ def run_job( return True, silent_doc, SILENT_MARKER, None try: - prompt = _build_job_prompt(job, prerun_script=prerun_script) + prompt = _build_job_prompt( + job, prerun_script=prerun_script, extra_prompt=extra_prompt + ) except CronPromptInjectionBlocked as block_exc: # Assembled prompt (user prompt + loaded skill content) tripped the # injection scanner. Refuse to run the agent this tick and surface @@ -3956,7 +3973,10 @@ def _teardown_cron_agent(agent, job_id: str) -> None: logger.debug("Job '%s': failed to reap stale auxiliary clients: %s", job_id, e) -def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -> bool: +def run_one_job( + job: dict, *, adapters=None, loop=None, verbose: bool = False, + extra_prompt: Optional[str] = None, +) -> bool: """Run ONE due job end-to-end: execute → save output → deliver → mark. This is the shared firing body extracted from ``tick``'s per-job closure so @@ -4024,7 +4044,8 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) - _deferred_agents: list = [] try: success, output, final_response, error = run_job( - job, defer_agent_teardown=_deferred_agents + job, defer_agent_teardown=_deferred_agents, + extra_prompt=extra_prompt, ) except BaseException: # run_job's finally still hands back the agent when it raises; tear diff --git a/tests/cron/test_cron_kanban_env_isolation.py b/tests/cron/test_cron_kanban_env_isolation.py index 24be753e0c13b..dfa91babe372b 100644 --- a/tests/cron/test_cron_kanban_env_isolation.py +++ b/tests/cron/test_cron_kanban_env_isolation.py @@ -252,7 +252,7 @@ class TestRunJobKanbanIsolation: }, ) monkeypatch.setattr( - sched, "_build_job_prompt", lambda job, prerun_script=None: "hi" + sched, "_build_job_prompt", lambda job, prerun_script=None, **kw: "hi" ) monkeypatch.setattr(sched, "_resolve_origin", lambda job: None) monkeypatch.setattr(sched, "_resolve_delivery_target", lambda job: None) diff --git a/tests/cron/test_cron_workdir.py b/tests/cron/test_cron_workdir.py index 6fbab3bbb53fc..be2c2aced20aa 100644 --- a/tests/cron/test_cron_workdir.py +++ b/tests/cron/test_cron_workdir.py @@ -199,7 +199,7 @@ class TestRunJobTerminalCwd: ) # Stub scheduler helpers that would otherwise hit the filesystem / config. - monkeypatch.setattr(sched, "_build_job_prompt", lambda job, prerun_script=None: "hi") + monkeypatch.setattr(sched, "_build_job_prompt", lambda job, prerun_script=None, **kw: "hi") monkeypatch.setattr(sched, "_resolve_origin", lambda job: None) monkeypatch.setattr(sched, "_resolve_delivery_target", lambda job: None) monkeypatch.setattr(sched, "_resolve_cron_enabled_toolsets", lambda job, cfg: None) diff --git a/tests/cron/test_execution_ledger.py b/tests/cron/test_execution_ledger.py index 62478aa95be44..5f268c64192a2 100644 --- a/tests/cron/test_execution_ledger.py +++ b/tests/cron/test_execution_ledger.py @@ -215,7 +215,7 @@ def test_run_one_job_records_running_then_terminal(monkeypatch): monkeypatch.setattr( scheduler, "run_job", - lambda job, *, defer_agent_teardown=None: (True, "output", "response", None), + lambda job, *, defer_agent_teardown=None, **_kw: (True, "output", "response", None), ) monkeypatch.setattr(scheduler, "save_job_output", lambda *_args: None) monkeypatch.setattr(scheduler, "_deliver_result", lambda *_args, **_kwargs: None) diff --git a/tests/cron/test_parallel_pool.py b/tests/cron/test_parallel_pool.py index 67ba0a32bcd7c..4eca57463eab2 100644 --- a/tests/cron/test_parallel_pool.py +++ b/tests/cron/test_parallel_pool.py @@ -146,7 +146,7 @@ class TestSequentialPool: barrier = threading.Barrier(2, timeout=5) - def slow_run(j, *, defer_agent_teardown=None): + def slow_run(j, *, defer_agent_teardown=None, **_kw): barrier.wait() return True, "out", "resp", None diff --git a/tests/cron/test_run_one_job.py b/tests/cron/test_run_one_job.py index d147d7e717ac5..cf774e97a16ac 100644 --- a/tests/cron/test_run_one_job.py +++ b/tests/cron/test_run_one_job.py @@ -18,7 +18,7 @@ def _patch_pipeline(monkeypatch, *, success=True, output="out", final="final res """Patch the job pipeline primitives and record the call order.""" calls = [] - def fake_run_job(job, *, defer_agent_teardown=None): + def fake_run_job(job, *, defer_agent_teardown=None, **kw): calls.append(("run_job", job["id"])) fr = final if silent_marker_in is None else silent_marker_in return (success, output, fr, error) @@ -83,7 +83,7 @@ def test_run_one_job_installs_secret_scope_under_multiplex(monkeypatch, tmp_path scope_during_run = {} - def fake_run_job(job, *, defer_agent_teardown=None): + def fake_run_job(job, *, defer_agent_teardown=None, **kw): # This is where resolve_runtime_provider() would read a secret. Prove a # scope is installed and the profile's secret resolves without raising. scope_during_run["scope"] = ss.current_secret_scope() diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 288610165601c..357614a23e110 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -1268,7 +1268,7 @@ class TestParallelTick: barrier = threading.Barrier(2, timeout=5) call_order = [] - def mock_run_job(job, *, defer_agent_teardown=None): + def mock_run_job(job, *, defer_agent_teardown=None, **kw): """Each job hits a barrier — both must be active simultaneously.""" call_order.append(("start", job["id"])) barrier.wait() # blocks until both threads reach here @@ -1302,7 +1302,7 @@ class TestParallelTick: from gateway.session_context import get_session_env seen = {} - def mock_run_job(job, *, defer_agent_teardown=None): + def mock_run_job(job, *, defer_agent_teardown=None, **kw): origin = job.get("origin", {}) # run_job sets ContextVars — verify each job sees its own from gateway.session_context import set_session_vars, clear_session_vars diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 3ebd5c31480b6..ad24d02f257bd 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -69,6 +69,7 @@ class TestCronjobRunExecutesImmediately: _JOB, adapters=adapters, loop=gateway_loop, + extra_prompt=None, ) def test_execute_job_now_remains_standalone_without_gateway(self): @@ -82,7 +83,7 @@ class TestCronjobRunExecutesImmediately: res = _execute_job_now(dict(_JOB)) assert res["success"] is True - m_run.assert_called_once_with(_JOB, adapters=None, loop=None) + m_run.assert_called_once_with(_JOB, adapters=None, loop=None, extra_prompt=None) def test_execute_job_now_marks_failure_on_exception(self): """An exception during fire is captured, marked failed, not propagated.""" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 96fc1da7d6f68..3607288081f91 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -589,7 +589,9 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]: return result -def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: +def _execute_job_now( + job: Dict[str, Any], extra_prompt: Optional[str] = None +) -> Dict[str, Any]: """Execute a cron job immediately, outside the scheduler tick. Atomically claims the job first via ``claim_job_for_fire`` — the same @@ -629,10 +631,12 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: pass return {"claimed": True, "success": False, "error": str(e)} - return _run_claimed_job(job) + return _run_claimed_job(job, extra_prompt=extra_prompt) -def _run_claimed_job(job: Dict[str, Any]) -> Dict[str, Any]: +def _run_claimed_job( + job: Dict[str, Any], extra_prompt: Optional[str] = None +) -> 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 @@ -745,7 +749,10 @@ def _run_claimed_job(job: Dict[str, Any]) -> Dict[str, Any]: try: try: - processed = run_one_job(job, adapters=adapters, loop=gateway_loop) + processed = run_one_job( + job, adapters=adapters, loop=gateway_loop, + extra_prompt=extra_prompt, + ) finally: _heartbeat_stop.set() if _heartbeat_thread is not None: @@ -806,7 +813,8 @@ def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[s def _try_dispatch_background_run( - job: Dict[str, Any], session_id: Optional[str] = None + job: Dict[str, Any], session_id: Optional[str] = None, + extra_prompt: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Claim ``job`` now, then fire it on the async-delegation daemon executor. @@ -935,7 +943,7 @@ def _try_dispatch_background_run( "cronjob run: async delegation registry unavailable (%s); " "running job '%s' inline.", e, job_name, ) - result = _run_claimed_job(job) + result = _run_claimed_job(job, extra_prompt=extra_prompt) result["dispatched"] = False return result @@ -950,7 +958,7 @@ def _try_dispatch_background_run( deliver = job.get("deliver", "local") def _runner() -> Dict[str, Any]: - res = _run_claimed_job(job) + res = _run_claimed_job(job, extra_prompt=extra_prompt) duration = round(time.time() - started_at, 2) refreshed = get_job(job_id) or {} lines = [ @@ -1008,7 +1016,7 @@ def _try_dispatch_background_run( "cronjob run: background pool unavailable (%s); running job '%s' inline.", dispatch.get("error", "rejected"), job_name, ) - result = _run_claimed_job(job) + result = _run_claimed_job(job, extra_prompt=extra_prompt) result["dispatched"] = False return result @@ -1194,6 +1202,16 @@ def cronjob( return json.dumps({"success": True, "job": _format_job(updated)}, indent=2) if normalized in {"run", "run_now", "trigger"}: + # Per-run context (#57331, salvaged from #57342/@liuhao1024 and + # #57360/@ghedeselmabot): `prompt` on the run action is transient + # context appended to the stored prompt for THIS fire only, never + # persisted. It goes through the same strict injection scan as + # stored prompts before firing. + extra_prompt = prompt or None + if extra_prompt: + scan_error = _scan_cron_prompt(extra_prompt) + if scan_error: + return tool_error(scan_error, success=False) # 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 (taken @@ -1208,7 +1226,9 @@ def cronjob( # 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) + bg = _try_dispatch_background_run( + job, session_id=session_id, extra_prompt=extra_prompt + ) 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}) @@ -1231,7 +1251,10 @@ def cronjob( # 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) + exec_result = ( + bg if bg is not None + else _execute_job_now(job, extra_prompt=extra_prompt) + ) # 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 @@ -1371,7 +1394,7 @@ 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. +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. Optionally pass 'prompt' with action='run' to inject transient per-run context (appended to the job's stored prompt for that single fire only, never persisted). 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. @@ -1397,7 +1420,7 @@ Important safety rule: cron-run sessions should not recursively schedule more cr }, "prompt": { "type": "string", - "description": "For create: the full self-contained prompt. If skills are also provided, this becomes the task instruction paired with those skills." + "description": "For create: the full self-contained prompt. If skills are also provided, this becomes the task instruction paired with those skills. For run: optional transient context appended to the stored prompt for that single fire only (never persisted)." }, "schedule": { "type": "string",