fix(cron): propagate pre-run script failure to job status

cron/scheduler.py

@@ -2168,10 +2168,13 @@ def _run_doc_header(job: dict, title: str, job_id: str, prompt: str) -> str:
 
 def _prepare_job_prompt(
     job: dict, job_id: str, job_name: str, extra_prompt: Optional[str], cancel_event,
-) -> tuple[Optional[_RunResult], Optional[str]]:
-    """Run every pre-agent gate and build the prompt. Returns ``(early_result, prompt)``: an early
-    result short-circuits ``run_job`` (no_agent job, empty payload, monitor gate, wake gate,
-    injection block, empty prompt); otherwise ``prompt`` is set."""
+) -> tuple[Optional[_RunResult], Optional[str], Optional[tuple[bool, Optional[str]]]]:
+    """Run every pre-agent gate and build the prompt. Returns ``(early_result, prompt,
+    script_failure)``: an early result short-circuits ``run_job`` (no_agent job, empty payload,
+    monitor gate, wake gate, injection block, empty prompt); otherwise ``prompt`` is set.
+    ``script_failure`` carries ``(failed, script_output)`` from the pre-run script so ``run_job``
+    can fail the run while still delivering the agent's report on the injected Script Error
+    block (#20301); it is ``None`` on every early-return path and when no script ran."""
     # Fail closed on a corrupt config.yaml: defaults would let auto-detection bill a provider the
     # user never chose. no_agent jobs are exempt. Escape hatch: HERMES_IGNORE_USER_CONFIG=1.
     if not job.get("no_agent"):
@@ -2181,21 +2184,21 @@ def _prepare_job_prompt(
             require_parseable_user_config()
         except InvalidUserConfigError as exc:
             logger.error("Job '%s': refusing to run — %s", job_id, exc)
-            return (False, f"# Cron Job: {job_name}\n\nError: {exc}\n", "", str(exc)), None
+            return (False, f"# Cron Job: {job_name}\n\nError: {exc}\n", "", str(exc)), None, None
 
     # no_agent short-circuits BEFORE importing run_agent / opening SessionDB.
     if job.get("no_agent"):
-        return _run_no_agent_job(job, job_id, job_name, cancel_event), None
+        return _run_no_agent_job(job, job_id, job_name, cancel_event), None, None
 
     # Legacy / hand-edited job with nothing to run: pause it instead of waking the LLM every fire.
     from cron.jobs import EMPTY_PAYLOAD_ERROR, job_payload_is_empty
 
     if job_payload_is_empty(job):
-        return _block_and_pause_job(job_id, job_name, EMPTY_PAYLOAD_ERROR), None
+        return _block_and_pause_job(job_id, job_name, EMPTY_PAYLOAD_ERROR), None, None
 
     _early, extra_prompt, monitor_context = _apply_monitor_gate(job, job_id, job_name, extra_prompt)
     if _early is not None:
-        return _early, None
+        return _early, None, None
 
     # Wake-gate: run the pre-check script BEFORE building the prompt; its result is passed into
     # _build_job_prompt so the script runs only once.
@@ -2205,6 +2208,8 @@ def _prepare_job_prompt(
     # happens inside the main try, right before the agent is constructed — after every early-return path
     # (#96290).
     prerun_script = None
+    script_failed = False
+    script_error: Optional[str] = None
     script_path = job.get("script")
     if script_path:
         prerun_script = _run_job_script_with_claim_heartbeat(
@@ -2214,6 +2219,9 @@ def _prepare_job_prompt(
             cancel_event=cancel_event,
         )
         _ran_ok, _script_output = prerun_script
+        if not _ran_ok:
+            script_failed = True
+            script_error = _script_output
         if _ran_ok and not _parse_wake_gate(_script_output):
             logger.info("Job '%s' (ID: %s): wakeAgent=false, skipping agent run", job_name, job_id)
             silent_doc = (
@@ -2222,7 +2230,7 @@ def _prepare_job_prompt(
                 f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n\n"
                 "Script gate returned `wakeAgent=false` — agent skipped.\n"
             )
-            return (True, silent_doc, SILENT_MARKER, None), None
+            return (True, silent_doc, SILENT_MARKER, None), None, None
 
     try:
         prompt = _build_job_prompt(
@@ -2247,11 +2255,11 @@ def _prepare_job_prompt(
             "and the match is a false positive, rephrase the content to avoid "
             "the threat pattern (`tools/cronjob_tools.py::_CRON_THREAT_PATTERNS`)."
         )
-        return (False, blocked_doc, "", str(block_exc)), None
+        return (False, blocked_doc, "", str(block_exc)), None, None
     if prompt is None:
         logger.info("Job '%s': script produced no output, skipping AI call.", job_name)
-        return (True, "", SILENT_MARKER, None), None
-    return None, prompt
+        return (True, "", SILENT_MARKER, None), None, None
+    return None, prompt, (script_failed, script_error)
 
 
 _CRON_DELIVERY_VARS = (
@@ -2488,7 +2496,7 @@ def run_job(
     job_id = job["id"]
     job_name = str(job.get("name") or job.get("prompt") or job_id or "cron job")
 
-    early, prompt = _prepare_job_prompt(job, job_id, job_name, extra_prompt, cancel_event)
+    early, prompt, script_failure = _prepare_job_prompt(job, job_id, job_name, extra_prompt, cancel_event)
     if early is not None:
         return early
     from run_agent import AIAgent
@@ -2538,6 +2546,13 @@ def run_job(
         output = _run_doc_header(job, job_name, job_id, prompt) + f"## Response\n\n{logged_response}\n"
         logger.info("Job '%s' completed successfully", job_name)
         _audit.write(dict(result, response_silent=_is_cron_silence_response(final_response or "")), None)
+        if script_failure and script_failure[0]:
+            # The pre-run script failed but the agent still ran on the injected "Script Error"
+            # block. Fail the run itself so mark_job_run() (last_status / failure_streak /
+            # last_error) and the executions ledger record the broken data collection instead
+            # of "ok", while final_response keeps the agent's report deliverable (#20301).
+            logger.warning("Job '%s': pre-run script failed — recording run as failed", job_name)
+            return False, output, final_response, f"Pre-run script failed: {script_failure[1]}"
         return True, output, final_response, None
 
     except Exception as e:

tests/cron/test_cron_script.py

@@ -725,3 +725,76 @@ def is_live(pid):
                     psutil.Process(gpid).kill()
                 except psutil.NoSuchProcess:
                     pass
+
+
+class TestRunJobScriptFailureStatus:
+    """A failed pre-run script must fail the run itself (#20301).
+
+    The agent still runs on the injected "Script Error" block and its report stays
+    deliverable, but ``run_job`` must return ``success=False`` with a non-empty error —
+    the values that become ``last_status`` / ``last_error`` and the executions-ledger
+    row — instead of reporting "ok" over a broken data collection.
+    """
+
+    def _run_with_script(self, cron_env, script_text):
+        from unittest.mock import MagicMock, patch
+
+        from cron import scheduler as sched
+
+        fake_provider_key = "test" + "-provider-key"  # dummy fixture value, never a real secret
+        script = cron_env / "scripts" / "collect.py"
+        script.write_text(textwrap.dedent(script_text))
+        job = {
+            "id": "script-status",
+            "name": "script-status-test",
+            "prompt": "Report status.",
+            "schedule_display": "every 1h",
+            "script": str(script),
+        }
+        agent = MagicMock()
+        agent.run_conversation.return_value = {
+            "final_response": "The data-collection script failed; see the error block.",
+            "messages": [],
+            "failed": False,
+            "completed": True,
+        }
+        with patch("cron.scheduler._hermes_home", cron_env), \
+             patch("cron.scheduler_delivery._resolve_origin", return_value=None), \
+             patch("hermes_cli.env_loader.load_hermes_dotenv"), \
+             patch("hermes_cli.env_loader.reset_secret_source_cache"), \
+             patch("hermes_state_registry.acquire", return_value=MagicMock()), \
+             patch("tools.mcp_tool_discovery.discover_mcp_tools", return_value=[]), \
+             patch("hermes_cli.runtime_provider.resolve_runtime_provider", return_value={
+                 "api_key": fake_provider_key,
+                 "base_url": "https://example.invalid/v1",
+                 "provider": "openrouter",
+                 "api_mode": "chat_completions",
+             }), \
+             patch("run_agent.AIAgent", return_value=agent):
+            return sched.run_job(dict(job))
+
+    def test_failing_script_fails_the_run(self, cron_env):
+        """Non-zero exit ⇒ success=False and the error carries the script output."""
+        success, output, response, error = self._run_with_script(
+            cron_env,
+            """\
+            import sys
+            print("error: data source unavailable", file=sys.stderr)
+            sys.exit(1)
+            """,
+        )
+
+        assert success is False
+        assert error is not None
+        assert "Pre-run script failed" in error
+        assert "data source unavailable" in error
+        # The agent still ran on the injected Script Error block; its report survives.
+        assert response == "The data-collection script failed; see the error block."
+
+    def test_successful_script_keeps_run_ok(self, cron_env):
+        """Zero exit ⇒ success=True with no error."""
+        success, output, response, error = self._run_with_script(
+            cron_env, 'print("data collected")\n')
+
+        assert success is True
+        assert error is None