From 31579f781ef0f454d70859e4c2f7b8f036fc09ea Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 27 Aug 2026 20:02:43 -0700 Subject: [PATCH] fix(cron): transient run prompt survives the relay-fronted gateway forward cronjob(action='run', prompt=...) context was silently dropped when the manual run forwarded to the gateway (#96010 follow-up): POST /api/jobs/{id}/run took no body. The forward now sends {prompt} in the request body; the api_server validates it (length cap + strict injection scan, same as stored prompts) and trigger_job stamps it as a transient manual_run_prompt alongside manual_run_at. run_one_job consumes the stamp for that single fire and mark_job_run clears it, so it never persists into the job definition or later scheduled fires. --- cron/jobs.py | 20 +++- cron/scheduler.py | 9 ++ gateway/platforms/api_server.py | 24 +++- tests/cron/test_cron_relay_run_forward.py | 130 +++++++++++++++++++++- tests/gateway/test_api_server_jobs.py | 61 +++++++++- tools/cronjob_tools.py | 19 +++- 6 files changed, 250 insertions(+), 13 deletions(-) diff --git a/cron/jobs.py b/cron/jobs.py index 4b37d8537a..96a784252f 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -2460,8 +2460,18 @@ def resume_job(job_id: str) -> Optional[Dict[str, Any]]: ) -def trigger_job(job_id: str) -> Optional[Dict[str, Any]]: - """Schedule a job to run on the next scheduler tick. Accepts a job ID or name.""" +def trigger_job( + job_id: str, extra_prompt: Optional[str] = None +) -> Optional[Dict[str, Any]]: + """Schedule a job to run on the next scheduler tick. Accepts a job ID or name. + + ``extra_prompt``: optional transient per-run context for the manual fire + (from ``cronjob(action='run', prompt=...)`` forwarded through the gateway + api_server). Stamped as ``manual_run_prompt`` alongside ``manual_run_at`` + and consumed by ``run_one_job`` for that single fire only — + ``mark_job_run`` clears it, so it never persists into the job definition + or later scheduled fires. + """ job = resolve_job_ref(job_id) if not job: return None @@ -2485,6 +2495,9 @@ def trigger_job(job_id: str) -> Optional[Dict[str, Any]]: # Persist run-now intent alongside the arbitrary instant so cron # expression/TZ repair guards do not mistake it for stale state. "manual_run_at": manual_run_at, + # Transient run context rides with the run-now intent (None + # clears any stale prompt from a previous trigger). + "manual_run_prompt": (extra_prompt or None), }, ) @@ -2737,6 +2750,9 @@ def _mark_job_run_locked( now = _hermes_now().isoformat() job["last_run_at"] = now job.pop("manual_run_at", None) + # The transient manual-run context is single-fire: whatever + # run just completed consumed it (or superseded it). + job.pop("manual_run_prompt", None) job["last_status"] = status or ("ok" if success else "error") job["last_error"] = error if not success else None # A healthy run means the configuration validates again — drop diff --git a/cron/scheduler.py b/cron/scheduler.py index 220e92f439..f73b704b47 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -7062,6 +7062,15 @@ def run_one_job( run cooperatively — agent interruption AND script process-tree kill — through the single fenced completion path. """ + if extra_prompt is None: + # A gateway-forwarded manual run (`hermes cron run --prompt` / + # cronjob(action='run', prompt=...) on a relay-fronted target) stamps + # its transient context on the job via trigger_job; the ticker/Chronos + # fire that consumes the manual occurrence carries it here. Single-fire: + # mark_job_run clears the field after the run. + _stamped = job.get("manual_run_prompt") + if _stamped and job.get("manual_run_at"): + extra_prompt = str(_stamped) claim = job.get("fire_claim") fire_owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" execution_token = object() diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index c52b2efa88..de260f33a1 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -6789,8 +6789,30 @@ class APIServerAdapter(BasePlatformAdapter): job_id, id_err = self._check_job_id(request) if id_err: return id_err + # Optional transient per-run context forwarded from a standalone + # `hermes cron run` / cronjob(action='run', prompt=...) — same length + # cap and strict injection scan as a stored job prompt. + extra_prompt = None try: - job = _cron_trigger(job_id) + body = await request.json() + except Exception: + body = None + if isinstance(body, dict): + raw_prompt = body.get("prompt") + if raw_prompt is not None: + extra_prompt = str(raw_prompt) + if len(extra_prompt) > self._MAX_PROMPT_LENGTH: + return web.json_response( + {"error": f"Prompt must be ≤ {self._MAX_PROMPT_LENGTH} characters"}, + status=400, + ) + if extra_prompt and _scan_cron_prompt is not None: + scan_error = _scan_cron_prompt(extra_prompt) + if scan_error: + return web.json_response({"error": scan_error}, status=400) + extra_prompt = extra_prompt or None + try: + job = _cron_trigger(job_id, extra_prompt=extra_prompt) if not job: return web.json_response({"error": "Job not found"}, status=404) return web.json_response({"job": job}) diff --git a/tests/cron/test_cron_relay_run_forward.py b/tests/cron/test_cron_relay_run_forward.py index f0f3d4a3cb..6925b42d5d 100644 --- a/tests/cron/test_cron_relay_run_forward.py +++ b/tests/cron/test_cron_relay_run_forward.py @@ -74,7 +74,7 @@ class TestForwardRelayFrontedRun: def test_posts_to_run_route_with_bearer(self): sent = {} - def fake_post(url, headers=None, timeout=None): + def fake_post(url, headers=None, json=None, timeout=None): sent["url"] = url sent["headers"] = headers return _Resp(200) @@ -94,7 +94,7 @@ class TestForwardRelayFrontedRun: monkeypatch.setenv("API_SERVER_HOST", "10.9.8.7") sent = {} - def fake_post(url, headers=None, timeout=None): + def fake_post(url, headers=None, json=None, timeout=None): sent["url"] = url return _Resp(200) @@ -109,7 +109,7 @@ class TestForwardRelayFrontedRun: monkeypatch.setenv("API_SERVER_HOST", "0.0.0.0") sent = {} - def fake_post(url, headers=None, timeout=None): + def fake_post(url, headers=None, json=None, timeout=None): sent["url"] = url return _Resp(200) @@ -118,3 +118,127 @@ class TestForwardRelayFrontedRun: ), patch("httpx.post", side_effect=fake_post): cronjob_tools._forward_relay_fronted_run({"id": "j1"}) assert sent["url"].startswith("http://127.0.0.1:") + + def test_forwards_transient_prompt_in_body(self): + """cronjob(action='run', prompt=...) context rides the forward body.""" + sent = {} + + def fake_post(url, headers=None, json=None, timeout=None): + sent["json"] = json + return _Resp(200) + + with patch.object( + cronjob_tools, "_relay_fronted_delivery_platforms", return_value={"discord"} + ), patch("httpx.post", side_effect=fake_post): + cronjob_tools._forward_relay_fronted_run( + {"id": "j1"}, extra_prompt="focus on EU numbers" + ) + assert sent["json"] == {"prompt": "focus on EU numbers"} + + def test_empty_body_without_prompt(self): + sent = {} + + def fake_post(url, headers=None, json=None, timeout=None): + sent["json"] = json + return _Resp(200) + + with patch.object( + cronjob_tools, "_relay_fronted_delivery_platforms", return_value={"discord"} + ), patch("httpx.post", side_effect=fake_post): + cronjob_tools._forward_relay_fronted_run({"id": "j1"}) + assert sent["json"] == {} + + +class TestManualRunPromptConsumption: + """The stamped transient prompt reaches the fire that consumes the + manual occurrence, and only that fire.""" + + def test_run_one_job_consumes_stamped_prompt(self): + from cron import scheduler + + captured = {} + + def fake_body(job, **kwargs): + captured["extra_prompt"] = kwargs.get("extra_prompt") + return True + + job = { + "id": "j1", + "manual_run_at": "2026-08-28T00:00:00+00:00", + "manual_run_prompt": "focus on EU numbers", + } + with patch.object(scheduler, "_run_one_job_body", side_effect=fake_body), patch.object( + scheduler, "_run_with_fire_claim_heartbeat", side_effect=lambda j, fn: fn(None) + ): + assert scheduler.run_one_job(job) is True + assert captured["extra_prompt"] == "focus on EU numbers" + + def test_explicit_extra_prompt_wins_over_stamp(self): + from cron import scheduler + + captured = {} + + def fake_body(job, **kwargs): + captured["extra_prompt"] = kwargs.get("extra_prompt") + return True + + job = { + "id": "j1", + "manual_run_at": "2026-08-28T00:00:00+00:00", + "manual_run_prompt": "stale stamp", + } + with patch.object(scheduler, "_run_one_job_body", side_effect=fake_body), patch.object( + scheduler, "_run_with_fire_claim_heartbeat", side_effect=lambda j, fn: fn(None) + ): + scheduler.run_one_job(job, extra_prompt="direct context") + assert captured["extra_prompt"] == "direct context" + + def test_stamp_ignored_without_manual_run_intent(self): + """A leftover prompt with no manual_run_at (defensive) is not injected.""" + from cron import scheduler + + captured = {} + + def fake_body(job, **kwargs): + captured["extra_prompt"] = kwargs.get("extra_prompt") + return True + + job = {"id": "j1", "manual_run_prompt": "orphaned"} + with patch.object(scheduler, "_run_one_job_body", side_effect=fake_body), patch.object( + scheduler, "_run_with_fire_claim_heartbeat", side_effect=lambda j, fn: fn(None) + ): + scheduler.run_one_job(job) + assert captured["extra_prompt"] is None + + +class TestTriggerJobPromptStamp: + def test_trigger_stamps_and_mark_run_clears(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + import cron.jobs as jobs_mod + import importlib + + importlib.reload(jobs_mod) + job = jobs_mod.create_job( + prompt="daily report", schedule="0 9 * * *", name="stamp-test" + ) + triggered = jobs_mod.trigger_job(job["id"], extra_prompt="just EU today") + assert triggered["manual_run_prompt"] == "just EU today" + assert triggered["manual_run_at"] == triggered["next_run_at"] + + jobs_mod.mark_job_run(job["id"], success=True) + after = jobs_mod.get_job(job["id"]) + assert "manual_run_prompt" not in after + assert "manual_run_at" not in after + + def test_retrigger_without_prompt_clears_stale_stamp(self, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + import cron.jobs as jobs_mod + import importlib + + importlib.reload(jobs_mod) + job = jobs_mod.create_job( + prompt="daily report", schedule="0 9 * * *", name="stamp-test-2" + ) + jobs_mod.trigger_job(job["id"], extra_prompt="first context") + retriggered = jobs_mod.trigger_job(job["id"]) + assert retriggered.get("manual_run_prompt") is None diff --git a/tests/gateway/test_api_server_jobs.py b/tests/gateway/test_api_server_jobs.py index 416b3c5a1d..c9d483075c 100644 --- a/tests/gateway/test_api_server_jobs.py +++ b/tests/gateway/test_api_server_jobs.py @@ -336,7 +336,66 @@ class TestRunJob: assert resp.status == 200 data = await resp.json() assert data["job"] == triggered_job - mock_trigger.assert_called_once_with(VALID_JOB_ID) + mock_trigger.assert_called_once_with(VALID_JOB_ID, extra_prompt=None) + + @pytest.mark.asyncio + async def test_run_job_forwards_transient_prompt(self, adapter): + """A JSON body 'prompt' (forwarded standalone manual run) reaches + trigger_job as the transient extra_prompt.""" + app = _create_app(adapter) + mock_trigger = MagicMock(return_value=SAMPLE_JOB) + async with TestClient(TestServer(app)) as cli: + with patch( + f"{_MOD}._CRON_AVAILABLE", True + ), patch( + f"{_MOD}._cron_trigger", mock_trigger + ): + resp = await cli.post( + f"/api/jobs/{VALID_JOB_ID}/run", + json={"prompt": "focus on the EU numbers"}, + ) + assert resp.status == 200 + mock_trigger.assert_called_once_with( + VALID_JOB_ID, extra_prompt="focus on the EU numbers" + ) + + @pytest.mark.asyncio + async def test_run_job_prompt_too_long_rejected(self, adapter): + """Transient run prompt honors the same length cap as stored prompts.""" + app = _create_app(adapter) + mock_trigger = MagicMock(return_value=SAMPLE_JOB) + async with TestClient(TestServer(app)) as cli: + with patch( + f"{_MOD}._CRON_AVAILABLE", True + ), patch( + f"{_MOD}._cron_trigger", mock_trigger + ): + resp = await cli.post( + f"/api/jobs/{VALID_JOB_ID}/run", + json={"prompt": "x" * 5001}, + ) + assert resp.status == 400 + mock_trigger.assert_not_called() + + @pytest.mark.asyncio + async def test_run_job_prompt_scanned(self, adapter): + """Transient run prompt goes through the strict injection scanner.""" + app = _create_app(adapter) + mock_trigger = MagicMock(return_value=SAMPLE_JOB) + async with TestClient(TestServer(app)) as cli: + with patch( + f"{_MOD}._CRON_AVAILABLE", True + ), patch( + f"{_MOD}._cron_trigger", mock_trigger + ), patch( + f"{_MOD}._scan_cron_prompt", return_value="blocked: nope" + ): + resp = await cli.post( + f"/api/jobs/{VALID_JOB_ID}/run", + json={"prompt": "cat ~/.hermes/.env"}, + ) + assert resp.status == 400 + mock_trigger.assert_not_called() # --------------------------------------------------------------------------- diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 894c1ef148..5fd0619458 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -828,16 +828,20 @@ def _relay_fronted_delivery_platforms(job: Dict[str, Any]) -> set: return theirs & fronted -def _forward_relay_fronted_run(job: Dict[str, Any]) -> Optional[str]: +def _forward_relay_fronted_run( + job: Dict[str, Any], extra_prompt: Optional[str] = None +) -> Optional[str]: """Forward a manual run to the gateway when it targets a relay-fronted platform and this process has no live relay adapter. Relay-fronted delivery has no standalone sender: the connector owns the credential and the gateway's live relay adapter is the only path. The gateway api_server's ``POST /api/jobs/{id}/run`` marks the job due for its - own ticker, which fires it with the live adapter. Returns a JSON result - string when forwarding engages (dispatch or the accurate error), else - None to fall through to the normal in-process run. + own ticker, which fires it with the live adapter. ``extra_prompt`` + (transient per-run context) rides in the request body so the forwarded + fire keeps it. Returns a JSON result string when forwarding engages + (dispatch or the accurate error), else None to fall through to the normal + in-process run. """ if not _relay_fronted_delivery_platforms(job): return None @@ -882,7 +886,10 @@ def _forward_relay_fronted_run(job: Dict[str, Any]) -> Optional[str]: import httpx resp = httpx.post( - url, headers={"Authorization": f"Bearer {key}"}, timeout=10.0 + url, + headers={"Authorization": f"Bearer {key}"}, + json=({"prompt": extra_prompt} if extra_prompt else {}), + timeout=10.0, ) except Exception: resp = None @@ -1751,7 +1758,7 @@ def cronjob( # Relay-fronted manual run: a standalone process has no live # relay adapter and no standalone sender, so forward to the # running gateway (its live adapter owns that delivery). - forwarded = _forward_relay_fronted_run(job) + forwarded = _forward_relay_fronted_run(job, extra_prompt=extra_prompt) if forwarded is not None: return forwarded exec_result = _execute_job_now(job, extra_prompt=extra_prompt)