fix(cron): sanitize monitor runtime prompt data

cron/scheduler.py

@@ -1275,15 +1275,16 @@ def _run_no_agent_job(
 
 def _apply_monitor_gate(
     job: dict, job_id: str, job_name: str, extra_prompt: Optional[str],
-) -> tuple[Optional[tuple], Optional[str]]:
+) -> tuple[Optional[tuple], Optional[str], Optional[str]]:
     """Monitor gate (hash-suppressed change detection). Must run BEFORE any agent machinery so an
-    unchanged tick costs no LLM/delivery. Returns ``(early_result | None, extra_prompt)``; when
-    early_result is None, extra_prompt may carry the injected monitor context.
+    unchanged tick costs no LLM/delivery. Returns ``(early_result | None, extra_prompt,
+    monitor_context)``. Monitor context is runtime data and must remain distinct from a
+    user-authored ``extra_prompt`` so the prompt scanner keeps its strict user-input boundary.
     """
     from cron.monitor import check_monitor, job_has_monitor
 
     if not job_has_monitor(job):
-        return None, extra_prompt
+        return None, extra_prompt, None
     _mon = check_monitor(job)
     _mon_now = _hermes_now().strftime("%Y-%m-%d %H:%M:%S")
     header = _job_doc_header(job_name, job_id, _mon_now, "monitor")
@@ -1298,19 +1299,16 @@ def _apply_monitor_gate(
         )
         return (
             False, f"{header}**Status:** monitor source failed\n\n{_mon.error}\n", _mon_alert, _mon.error,
-        ), extra_prompt
+        ), extra_prompt, None
     if not _mon.changed:
         # Unchanged: silent no_change tick (ledger doc kept; SILENT_MARKER blocks delivery).
         logger.info("Job '%s': monitor output unchanged — suppressing agent run", job_id)
         return (
             True, f"{header}**Status:** no_change (agent run suppressed)\n", SILENT_MARKER, None,
-        ), extra_prompt
-    # Changed (or first run): inject monitor context via the per-run seam, then normal agent run.
-    if _mon.context_block:
-        extra_prompt = (
-            f"{_mon.context_block}\n\n{extra_prompt}" if extra_prompt else _mon.context_block
-        )
-    return None, extra_prompt
+        ), extra_prompt, None
+    # Changed (or first run): pass monitor output through the runtime-data seam. Keep any manual
+    # per-run prompt separate: it remains user input and is therefore still strict-scanned.
+    return None, extra_prompt, _mon.context_block
 
 
 @dataclass
@@ -1949,7 +1947,7 @@ def _prepare_job_prompt(
     if job_payload_is_empty(job):
         return _block_and_pause_job(job_id, job_name, EMPTY_PAYLOAD_ERROR), None
 
-    _early, extra_prompt = _apply_monitor_gate(job, job_id, job_name, extra_prompt)
+    _early, extra_prompt, monitor_context = _apply_monitor_gate(job, job_id, job_name, extra_prompt)
     if _early is not None:
         return _early, None
 
@@ -1981,7 +1979,10 @@ def _prepare_job_prompt(
             return (True, silent_doc, SILENT_MARKER, None), None
 
     try:
-        prompt = _build_job_prompt(job, prerun_script=prerun_script, extra_prompt=extra_prompt)
+        build_kwargs = {"prerun_script": prerun_script, "extra_prompt": extra_prompt}
+        if monitor_context:
+            build_kwargs["runtime_data_prompt"] = monitor_context
+        prompt = _build_job_prompt(job, **build_kwargs)
     except CronPromptInjectionBlocked as block_exc:
         # Injection scanner tripped: refuse this tick and tell the operator WHY.
         logger.warning(

cron/scheduler_prompt.py

@@ -210,11 +210,14 @@ _CRON_HINT = (
 
 
 def _build_job_prompt(
-    job: dict, prerun_script: Optional[tuple] = None, extra_prompt: Optional[str] = None) -> str:
+    job: dict, prerun_script: Optional[tuple] = None, extra_prompt: Optional[str] = None,
+    runtime_data_prompt: Optional[str] = None,
+) -> str:
     """Build the effective prompt for a cron job, optionally loading skills first.
     ``prerun_script``: cached ``(success, stdout)`` from a script the caller already ran (wake-gate
-    check) — skips re-execution. ``extra_prompt``: per-run ``## Run Context`` for this fire only,
-    never persisted to the job.
+    check) — skips re-execution. ``extra_prompt``: user-authored per-run ``## Run Context`` for this
+    fire only, never persisted to the job. ``runtime_data_prompt`` is operator-configured runtime
+    data (such as monitor output) and is scanned as injected data rather than user input.
 
     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
@@ -227,6 +230,9 @@ def _build_job_prompt(
     # Runtime DATA (script stdout, upstream output) legitimately quotes command-shape strings, so it
     # must not be scanned with the strict user-prompt set — see _scan_assembled_cron_prompt.
     has_injected_data = False
+    if runtime_data_prompt:
+        prompt = f"{prompt}\n\n## Run Context\n{runtime_data_prompt}"
+        has_injected_data = True
 
     script_path = job.get("script")
     if script_path:

tests/cron/test_cron_prompt_injection_skill.py

@@ -393,3 +393,25 @@ class TestScriptOutputNotStrictScanned:
         assert "\u200b" not in prompt
         assert "item oneitem two" in prompt
 
+
+class TestMonitorOutputIsRuntimeData:
+    """Monitor context is runtime data, not part of the stored user prompt."""
+
+    def test_bidi_monitor_output_is_sanitized_not_blocked(self, cron_env):
+        _, scheduler = cron_env
+        prompt = scheduler._build_job_prompt(
+            {"id": "job-monitor", "name": "WhatsApp", "prompt": "Summarize changes."},
+            runtime_data_prompt="## Monitor Baseline\n\nWhatsApp: Alice\u202a Work",
+        )
+        assert prompt is not None
+        assert "\u202a" not in prompt
+        assert "WhatsApp: Alice Work" in prompt
+
+    def test_stored_user_prompt_remains_strict_with_monitor_data(self, cron_env):
+        _, scheduler = cron_env
+        with pytest.raises(scheduler.CronPromptInjectionBlocked) as exc_info:
+            scheduler._build_job_prompt(
+                {"id": "job-monitor", "name": "legacy", "prompt": "normal\u202atext"},
+                runtime_data_prompt="## Monitor Baseline\n\nordinary monitor output",
+            )
+        assert "invisible unicode" in str(exc_info.value)

tests/cron/test_monitor_kind.py

@@ -293,6 +293,23 @@ def test_first_run_always_runs_agent(hermes_env, monkeypatch):
     assert "state A" in observed["prompts"][0]
 
 
+def test_bidi_monitor_output_is_sanitized_before_agent(hermes_env, monkeypatch):
+    """Runtime monitor data with a WhatsApp-style bidi marker must not block the job."""
+    from cron.scheduler import run_job
+
+    job = _make_monitor_job(hermes_env, "printf 'Alice\\342\\200\\252 Work\\n'\n")
+    observed: dict = {}
+    _install_agent_stubs(monkeypatch, observed)
+
+    success, _, _, error = run_job(job)
+
+    assert success is True
+    assert error is None
+    assert observed["agent_runs"] == 1
+    assert "\u202a" not in observed["prompts"][0]
+    assert "Alice Work" in observed["prompts"][0]
+
+
 def test_unchanged_output_suppresses_agent_run(hermes_env, monkeypatch):
     from cron.jobs import get_job
     from cron.scheduler import SILENT_MARKER, run_job