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

@@ -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

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()

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})

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

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()
 
 
 # ---------------------------------------------------------------------------

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)