fix(cron): add per-job allow_silent flag to control [SILENT] suppression

cron/jobs.py

@@ -1732,6 +1732,13 @@ def _normalize_reasoning_effort(value: Any) -> Optional[str]:
 
 # Normalizers for create_job (all fields) / update_job (present fields). Invalid values raise BEFORE
 # storing.
+def _normalize_allow_silent(value: Any) -> bool:
+    """``allow_silent`` defaults to True when unset: an absent/None value keeps the historical
+    ``[SILENT]`` suppression behaviour, so legacy jobs and agent calls that omit the field are
+    unchanged (#53230)."""
+    return True if value is None else bool(value)
+
+
 _CREATE_FIELD_NORMALIZERS: Dict[str, Callable[[Any], Any]] = {
     "model": _normalize_job_optional_text,
     "provider": _normalize_job_optional_text,
@@ -1745,13 +1752,15 @@ def _normalize_reasoning_effort(value: Any) -> Optional[str]:
     "context_from": _normalize_context_from,
     "failure_deliver": _normalize_failure_deliver,
     "interpreter": _normalize_job_optional_text,
+    "allow_silent": _normalize_allow_silent,
 }
 _UPDATE_FIELD_NORMALIZERS: Dict[str, Callable[[Any], Any]] = {
     "workdir": lambda v: None if v in {None, "", False} else _normalize_workdir(v),
     "monitor_script": _normalize_job_optional_text,
     "monitor_url": _normalize_job_optional_text,
     "interpreter": _normalize_job_optional_text,
     "reasoning_effort": _normalize_reasoning_effort,
+    "allow_silent": _normalize_allow_silent,
 }
 
 
@@ -1823,6 +1832,7 @@ def create_job(
     paused_reason: Optional[str] = None,
     pinned: bool = False,
     interpreter: Optional[str] = None,
+    allow_silent: bool = True,
 ) -> Dict[str, Any]:
     """Create a new cron job and return the stored record.
 
@@ -1833,7 +1843,10 @@ def create_job(
     source run FIRST each tick; unchanged output suppresses the agent run (mutually exclusive,
     incompatible with ``no_agent``). reasoning_effort: per-job pin; capability NOT validated.
     interpreter: absolute/``~`` Python for ``.py`` script/monitor_script, validated at run time
-    (a venv can be rebuilt or moved after creation)."""
+    (a venv can be rebuilt or moved after creation). allow_silent: when True (default) an agent
+    response of ``[SILENT]`` suppresses delivery; when False the scheduler sends a short all-clear
+    instead, so a recurring briefing/report never goes quiet (#53230). Internal script-job silence
+    (no_agent empty stdout, wakeAgent=false) is unaffected either way."""
     if not isinstance(paused, bool):
         raise ValueError("paused must be a boolean.")
     if paused_reason is not None and not isinstance(paused_reason, str):
@@ -1912,6 +1925,9 @@ def create_job(
         "origin": origin,  # Tracks where job was created for "origin" delivery
         "enabled_toolsets": f["enabled_toolsets"],
         "workdir": f["workdir"],
+        # When False the scheduler replaces an agent [SILENT] response with a short all-clear
+        # instead of suppressing delivery — see _build_job_prompt / run_one_job (#53230).
+        "allow_silent": f["allow_silent"],
     }
     # Optional keys are persisted only when explicitly set: an absent key falls back to global
     # config (attach/reasoning) or to ``deliver`` (failure_deliver), byte-identical to pre-feature

cron/scheduler.py

@@ -548,6 +548,10 @@ def _resolve_job_reasoning_config(job: dict, cfg: dict, model: str) -> dict | No
 # Response marker that suppresses delivery (output is still saved locally for audit).
 SILENT_MARKER = "[SILENT]"
 
+# Delivered in place of an agent silence marker when the job opted out of silence with
+# ``allow_silent=False`` (recurring briefings that must always send an all-clear, #53230).
+CRON_ALL_CLEAR_MESSAGE = "Scheduled job completed with no reportable updates."
+
 # Agent-declared failure marker for cron runs. Unlike SILENT, it is deliberately strict so a
 # report that merely quotes the token cannot turn a healthy run into a failed one.
 CRON_FAILURE_MARKER = "[CRON_FAILURE]"
@@ -3010,8 +3014,16 @@ def _save_compose_deliver(
         # Cron silence suppression — see _is_cron_silence_response. Replaces the old `SILENT_MARKER in
         # ...upper()` substring check, which both leaked bracketless near-markers ("SILENT" / "NO_REPLY")
         # and wrongly swallowed a real report that merely quoted "[SILENT]" mid-sentence (#51438, #46917).
-        logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER)
-        d.should_deliver = False
+        if job.get("no_agent") or job.get("allow_silent", True):
+            logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER)
+            d.should_deliver = False
+        else:
+            # allow_silent=False: the job opted out of going quiet, so the user gets a real
+            # all-clear instead of the literal control marker the agent emitted (#53230).
+            logger.info(
+                "Job '%s': agent returned %s — delivering all-clear (allow_silent=False)",
+                job["id"], SILENT_MARKER)
+            deliver_content = CRON_ALL_CLEAR_MESSAGE
 
     if d.should_deliver and fence.lost():
         d.should_deliver = False

cron/scheduler_prompt.py

@@ -220,18 +220,32 @@ def _skip(msg: str, *args) -> None:
     return parts
 
 
-_CRON_HINT = (
+_CRON_HINT_HEAD = (
     "[IMPORTANT: You are running as a scheduled cron job. "
     "DELIVERY: Your final response will be automatically delivered "
     "to the user — do NOT use send_message or try to deliver "
     "the output yourself. Just produce your report/output as your "
     "final response and the system handles the rest. "
+)
+_CRON_HINT_SILENT = (
     "SILENT: If there is genuinely nothing new to report, respond "
     "with exactly \"[SILENT]\" (nothing else) to suppress delivery. "
     "[SILENT] is a literal ASCII control token — never translate or "
     "rephrase it, whatever language the rest of your answer uses. "
     "Never combine [SILENT] with content — either report your "
     "findings normally, or say [SILENT] and nothing more. "
+)
+# allow_silent=False: the silence escape hatch is unavailable, so the job is told to send an
+# all-clear instead. Deliberately does not name the marker tokens — the scheduler's matcher in
+# run_one_job still catches one the model emits anyway and substitutes the all-clear, and naming
+# them in an instruction ("do not say X") is a good way to get X (#53230).
+_CRON_HINT_ALWAYS_REPORT = (
+    "ALWAYS REPORT: this job must deliver a report on every run — it is a "
+    "recurring briefing the user expects to receive, so do not go quiet. "
+    "If there is genuinely nothing new to report, send a short all-clear "
+    "(e.g. \"No changes.\") rather than nothing. "
+)
+_CRON_HINT_TAIL = (
     "FAILURE: If a delegated child fails and this cron run must be "
     "recorded as failed, put [CRON_FAILURE] on the first line by itself, "
     "then explain the child failure on following lines. "
@@ -243,6 +257,21 @@ def _skip(msg: str, *args) -> None:
 )
 
 
+def _cron_hint(job: dict) -> str:
+    """The cron execution banner for ``job``.
+
+    The ``[SILENT]`` suppression guidance is injected only when the job allows silent delivery
+    (default, including legacy jobs with no ``allow_silent`` key). A job that opted out with
+    ``allow_silent=False`` gets the always-report instruction instead, so its prompt never
+    advertises a silence marker the scheduler is going to override (#53230).
+    """
+    if job.get("allow_silent", True):
+        body = _CRON_HINT_SILENT
+    else:
+        body = _CRON_HINT_ALWAYS_REPORT
+    return _CRON_HINT_HEAD + body + _CRON_HINT_TAIL
+
+
 def _build_job_prompt(
     job: dict, prerun_script: Optional[tuple] = None, extra_prompt: Optional[str] = None,
     runtime_data_prompt: Optional[str] = None,
@@ -296,7 +325,7 @@ def _build_job_prompt(
         prompt = f"{notepad_section}{prompt}"
         has_injected_data = True
 
-    prompt = _CRON_HINT + prompt
+    prompt = _cron_hint(job) + prompt
     skill_names = _job_skill_names(job)
     if not skill_names:
         return _scan_assembled_cron_prompt(

tests/cron/test_cron_allow_silent.py

@@ -0,0 +1,151 @@
+"""Per-job ``allow_silent`` policy (#53230).
+
+A recurring briefing that promises an all-clear must not silently skip a run
+because the model decided there was "nothing new". ``allow_silent=False`` on a
+job:
+
+1. replaces the ``[SILENT]`` suppression guidance in the prompt with an
+   always-report instruction, and
+2. makes the delivery path send a short all-clear when the model emits a
+   silence marker anyway, instead of suppressing the delivery or leaking the
+   literal marker text to the user.
+
+Default (``True``, including jobs stored before the field existed) keeps the
+historical behaviour byte-for-byte. Internal script-job silences (``no_agent``
+empty stdout, ``wakeAgent=false``) are scheduler signals rather than model
+decisions and stay silent either way.
+"""
+
+import pytest
+
+import cron.scheduler as s
+from cron.jobs import create_job, update_job
+from cron.scheduler import (
+    CRON_ALL_CLEAR_MESSAGE,
+    SILENT_MARKER,
+    _build_job_prompt,
+    _is_cron_silence_response,
+)
+
+
+class TestPromptInjection:
+    """The flag selects which delivery guidance the prompt carries."""
+
+    def test_default_job_injects_silent_guidance(self):
+        job = create_job(prompt="Daily report", schedule="0 9 * * *")
+        assert SILENT_MARKER in _build_job_prompt(job)
+
+    def test_allow_silent_true_injects_guidance(self):
+        job = create_job(prompt="Daily report", schedule="0 9 * * *", allow_silent=True)
+        assert SILENT_MARKER in _build_job_prompt(job)
+
+    def test_allow_silent_false_omits_silent_guidance(self):
+        job = create_job(
+            prompt="Daily report — always send an all-clear",
+            schedule="0 9 * * *",
+            allow_silent=False,
+        )
+        prompt = _build_job_prompt(job)
+        assert SILENT_MARKER not in prompt
+        # The delivery contract itself is not what is being turned off.
+        assert "scheduled cron job" in prompt
+        assert "DELIVERY:" in prompt
+        assert "ALWAYS REPORT" in prompt
+
+    def test_legacy_job_without_key_defaults_to_silent_guidance(self):
+        job = create_job(prompt="Legacy job", schedule="0 9 * * *")
+        job.pop("allow_silent", None)
+        assert SILENT_MARKER in _build_job_prompt(job)
+
+    def test_failure_and_recursion_markers_survive_the_switch(self):
+        for allow_silent in (True, False):
+            job = create_job(prompt="Test", schedule="0 9 * * *", allow_silent=allow_silent)
+            prompt = _build_job_prompt(job)
+            assert "[CRON_FAILURE]" in prompt
+            assert "RECURSION:" in prompt
+
+
+class TestJobField:
+    """The flag round-trips through the job store."""
+
+    def test_field_defaults_to_true(self):
+        assert create_job(prompt="Test", schedule="0 9 * * *").get("allow_silent") is True
+
+    def test_field_persists_false(self):
+        job = create_job(prompt="Test", schedule="0 9 * * *", allow_silent=False)
+        assert job.get("allow_silent") is False
+        assert isinstance(job.get("allow_silent"), bool)
+
+    def test_update_flips_the_field(self):
+        job = create_job(prompt="Test", schedule="0 9 * * *")
+        updated = update_job(job["id"], {"allow_silent": False})
+        assert updated is not None
+        assert updated.get("allow_silent") is False
+        assert SILENT_MARKER not in _build_job_prompt(updated)
+
+
+class TestSilenceClassification:
+    """``_is_cron_silence_response`` is a pure string check — the flag gates
+    whether the scheduler *acts* on it, not whether it is detectable."""
+
+    @pytest.mark.parametrize("marker", ["[SILENT]", "[SILENT]\n", "SILENT", "NO_REPLY", "NO REPLY"])
+    def test_detects_markers(self, marker):
+        assert _is_cron_silence_response(marker)
+
+    def test_does_not_detect_normal_text(self):
+        assert not _is_cron_silence_response("All systems normal")
+        assert not _is_cron_silence_response("Daily report: nothing changed.")
+
+
+@pytest.fixture
+def run_env(monkeypatch):
+    """Drive the real ``run_one_job`` delivery decision, captured at _deliver_result."""
+    delivered = []
+
+    monkeypatch.setattr(s, "create_execution", lambda *_a, **_kw: {"id": "exec-t"})
+    monkeypatch.setattr(s, "claim_dispatch", lambda _job_id: True)
+    monkeypatch.setattr(s, "mark_execution_running", lambda *_a, **_kw: {})
+    monkeypatch.setattr(s, "save_job_output", lambda jid, out: f"/tmp/{jid}.txt")
+    monkeypatch.setattr(s, "mark_job_run", lambda *_a, **_kw: True)
+    monkeypatch.setattr(s, "finish_execution", lambda *_a, **_kw: None)
+    monkeypatch.setattr(s, "_upsert_incident_for_failure", lambda *_a, **_kw: (False, None))
+    monkeypatch.setattr(s, "load_config", lambda: {})
+    monkeypatch.setattr(
+        s, "_deliver_result",
+        lambda job, content, **_kw: delivered.append(content) or None,
+    )
+    return delivered
+
+
+def _succeeding_run_job(final):
+    def _fake(job, **_kw):
+        return (True, "raw output", final, None)
+
+    return _fake
+
+
+@pytest.mark.parametrize(
+    ("no_agent", "allow_silent", "response", "expected"),
+    [
+        (False, True, "[SILENT]", None),
+        (False, None, "[SILENT]", None),  # legacy job: key absent
+        (False, False, "[SILENT]", CRON_ALL_CLEAR_MESSAGE),
+        (False, False, "NO_REPLY", CRON_ALL_CLEAR_MESSAGE),
+        (False, False, "NO REPLY", CRON_ALL_CLEAR_MESSAGE),
+        (False, False, "Daily report: normal", "Daily report: normal"),
+        (True, False, "[SILENT]", None),  # internal script silence, not a model decision
+    ],
+)
+def test_delivery_policy(monkeypatch, run_env, no_agent, allow_silent, response, expected):
+    job = {"id": "silence-policy", "name": "policy-test", "deliver": "local",
+           "no_agent": no_agent}
+    if allow_silent is not None:
+        job["allow_silent"] = allow_silent
+    monkeypatch.setattr(s, "run_job", _succeeding_run_job(response))
+
+    assert s.run_one_job(job) is True
+
+    if expected is None:
+        assert run_env == []
+    else:
+        assert run_env == [expected]

tools/cronjob_tools.py

@@ -688,7 +688,7 @@ def _action_create(a: Dict[str, Any]) -> str:
             monitor_url=_normalize_optional_job_value(a["monitor_url"]),
             # CLI-only lane: absent from CRONJOB_SCHEMA and the model dispatch (models don't pick models).
             reasoning_effort=a["reasoning_effort"], interpreter=a["interpreter"],
-            pinned=bool(a["pinned"]),
+            pinned=bool(a["pinned"]), allow_silent=a["allow_silent"],
             failure_deliver=_resolve_cron_context_deliver(_normalize_deliver_param(a["failure_deliver"])),
             **({"paused": a["paused"], "paused_reason": a["paused_reason"]}
                if a["paused"] is not False or a["paused_reason"] is not None else {}))
@@ -899,6 +899,10 @@ def _update_run_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str
         updates["enabled_toolsets"] = a["enabled_toolsets"] or None
     if a["attach_to_session"] is not None:
         updates["attach_to_session"] = bool(a["attach_to_session"])
+    if a["allow_silent"] is not None:
+        # Tri-state on the tool surface: only an explicit True/False edits the field, so a model
+        # re-sending the schema with type-default empties cannot flip a job's silence policy.
+        updates["allow_silent"] = bool(a["allow_silent"])
     if a["workdir"] is not None:
         # Empty string clears; otherwise update_job() validates/normalizes.
         updates["workdir"] = _normalize_optional_job_value(a["workdir"]) or None
@@ -1005,7 +1009,8 @@ def cronjob(
     paused: bool = False,
     paused_reason: Optional[str] = None,
     pinned: Optional[bool] = None,
-    interpreter: Optional[str] = None) -> str:
+    interpreter: Optional[str] = None,
+    allow_silent: Optional[bool] = None) -> str:
     """Unified cron job management tool."""
     a = dict(locals())
     del a["task_id"]  # unused but kept for handler signature compatibility
@@ -1133,6 +1138,10 @@ def _cronjob_schema_overrides() -> dict:
                 "type": "boolean",
                 "description": "True = the job's delivery is CONTINUABLE — the user can reply and the agent has the brief in context (threads on thread-capable platforms, mirrored into the DM elsewhere). Use for conversational recurring jobs (briefings); leave unset for fire-and-forget alerts. Scope: the job's own conversation only — the origin chat, the home-channel fallback when deliver='origin' captured no origin (script-created jobs), a user-written bare platform target (deliver='slack' — that platform's home channel), or the job's single explicit platform:chat target (this flag is the only way to attach an explicit target). Broadcast targets are never attached; no effect when deliver='local'."
             },
+            "allow_silent": {
+                "type": "boolean",
+                "description": "Default True: the agent may suppress delivery by responding with [SILENT] when there is nothing new to report. Set False for recurring briefings/heartbeats that must always send something — the [SILENT] guidance is replaced with an always-report instruction and, if the model emits a silence marker anyway, the scheduler delivers a short all-clear instead of the raw marker. Internal silences from no_agent script jobs (empty stdout, wakeAgent=false) stay silent regardless: this flag only governs an LLM agent's explicit silence response. Leave unset to keep the current behaviour."
+            },
         },
         "required": ["action"]
     }
@@ -1161,7 +1170,7 @@ def check_cronjob_requirements() -> bool:
 _HANDLER_FORWARDED_ARGS = (
     "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", "reason",
     "script", "context_from", "continuity", "enabled_toolsets", "workdir", "no_agent", "attach_to_session",
-    "paused_reason", "pinned")
+    "paused_reason", "pinned", "allow_silent")
 
 
 def _cronjob_handler(args, **kw):