fix(cron): restore [SILENT] silence + suppress empty-turn explainer on Telegram

Scheduled jobs delivering to Telegram/etc. started posting a literal
'⚠️ No reply: the model returned empty content…' message instead of
staying silent. Two interacting causes:

1. The turn-completion explainer (#34452) replaces an empty model turn
   with a user-facing '⚠️ No reply…' string. In a cron context that is
   not a silence marker, so the scheduler delivered it — a regression
   from the previously-silent empty turn. run_job now detects the
   explainer text deterministically (via the same formatter that
   produced it) for abnormal-empty turn_exit_reasons and strips it to
   empty, so the existing empty-response suppression + soft-fail guard
   apply. The explainer is unchanged on CLI/gateway.

2. The cron suppression used a loose 'SILENT_MARKER in ...upper()'
   substring check. It leaked bracketless near-markers the model emits
   ('SILENT', 'NO_REPLY', 'NO REPLY' — #51438, #46917) and wrongly
   swallowed a real report that merely quoted '[SILENT]' mid-sentence.
   Replaced with _is_cron_silence_response(): suppresses a canonical
   token as the whole response, its own first/last line, or the
   documented bracketed '[SILENT] <note>' prefix — while a token buried
   mid-sentence in a genuine report is delivered. Preserves the
   intentional cron trailing/prefix tolerance (existing tests unchanged).

Tests: bracketless-variant suppression, mid-sentence-quote delivery,
direct matcher contract, and explainer-strip + defensive real-report
delivery.

cron/scheduler.py

@@ -243,6 +243,50 @@ from cron.jobs import get_due_jobs, mark_job_run, save_job_output, advance_next_
 # locally for audit.
 SILENT_MARKER = "[SILENT]"
 
+# Canonical silence tokens recognized in cron output.  Cron's contract is
+# intentionally looser than the gateway's exact-whole-response rule: the cron
+# system prompt *instructs* the agent to emit "[SILENT]", and real agents often
+# bracket it with a short note or trailing newline.  We therefore suppress when
+# a marker is the entire response OR appears as its own first/last line — but
+# NOT when a token merely appears mid-sentence in a genuine report (e.g.
+# "I considered staying [SILENT] but here is the summary…" must deliver).
+_CRON_SILENCE_TOKENS = frozenset({"[SILENT]", "SILENT", "NO_REPLY", "NO REPLY"})
+
+
+def _is_cron_silence_response(text: str) -> bool:
+    """Return True when a cron final response should suppress delivery.
+
+    Recognizes the bracketed ``[SILENT]`` sentinel (whole-response, first line,
+    or last line) plus the bracketless ``SILENT`` / ``NO_REPLY`` / ``NO REPLY``
+    variants the model emits when it drops the brackets (#51438, #46917).
+    Whitespace-trimmed and case-insensitive.  A token buried mid-sentence is
+    treated as real content and delivered.
+    """
+    if not isinstance(text, str):
+        return False
+    stripped = text.strip()
+    if not stripped:
+        return False
+
+    def _is_token(line: str) -> bool:
+        return " ".join(line.strip().upper().split()) in _CRON_SILENCE_TOKENS
+
+    # Whole response is exactly a token.
+    if _is_token(stripped):
+        return True
+    # Marker on its own first or last line (trailing/leading note on a
+    # separate line — e.g. "2 deals filtered\n\n[SILENT]").
+    lines = [ln for ln in stripped.splitlines() if ln.strip()]
+    if lines and (_is_token(lines[0]) or _is_token(lines[-1])):
+        return True
+    # Bracketed sentinel used as a same-line prefix — the documented cron
+    # pattern "[SILENT] No changes detected".  Restricted to the bracketed
+    # form so a bare word like "Silent retry succeeded" is NOT swallowed.
+    upper = stripped.upper()
+    if upper.startswith("[SILENT]"):
+        return True
+    return False
+
 # ---------------------------------------------------------------------------
 # Persistent thread pool for parallel cron jobs.
 # The tick function submits jobs here and returns immediately so the ticker
@@ -2582,6 +2626,27 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
         # Strip leaked placeholder text that upstream may inject on empty completions.
         if final_response.strip() == "(No response generated)":
             final_response = ""
+        # Cron silence on abnormal empty turns.  The turn-completion explainer
+        # (#34452) replaces a blank/empty model turn with a "⚠️ No reply: …"
+        # string so interactive surfaces (CLI/gateway) explain why the box is
+        # empty.  In a cron context that turns a previously-silent empty turn
+        # into a delivered warning (Manfredi's Telegram symptom).  Detect the
+        # explainer text deterministically (via the same formatter that
+        # produced it) and treat it as empty so the empty-response suppression
+        # and soft-failure marking below apply — restoring pre-#34452 silence
+        # for scheduled jobs without disabling the explainer everywhere.
+        if final_response.strip() and turn_exit_reason:
+            try:
+                _explainer_text = AIAgent._format_turn_completion_explanation(turn_exit_reason)
+            except Exception:
+                _explainer_text = ""
+            if _explainer_text and final_response.strip() == _explainer_text.strip():
+                logger.info(
+                    "Job '%s': abnormal empty turn (%s) — suppressing explainer for cron delivery",
+                    job_id,
+                    turn_exit_reason,
+                )
+                final_response = ""
         # Use a separate variable for log display; keep final_response clean
         # for delivery logic (empty response = no delivery).
         logged_response = final_response if final_response else "(No response generated)"
@@ -2710,7 +2775,13 @@ def run_one_job(job: dict, *, adapters=None, loop=None, verbose: bool = False) -
         # responses: do not deliver a blank message, and let the
         # empty-response guard below mark the run as a soft failure.
         should_deliver = bool(deliver_content.strip())
-        if should_deliver and success and SILENT_MARKER in deliver_content.strip().upper():
+        # 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).  Keeps the intentional bracketed-prefix / trailing-line
+        # tolerance the cron contract relies on.
+        if should_deliver and success and _is_cron_silence_response(deliver_content):
             logger.info("Job '%s': agent returned %s — skipping delivery", job["id"], SILENT_MARKER)
             should_deliver = False
 

tests/cron/test_scheduler.py

@@ -1000,6 +1000,88 @@ class TestRunJobSessionPersistence:
         fake_db.close.assert_called_once()
         mock_agent.close.assert_called_once()
 
+    def test_run_job_suppresses_empty_turn_explainer(self, tmp_path):
+        """An empty model turn becomes the '⚠️ No reply…' explainer (#34452).
+        For cron, that abnormal-empty explainer must be treated as empty so it
+        is suppressed instead of delivered (Manfredi's Telegram symptom)."""
+        from run_agent import AIAgent
+        explainer = AIAgent._format_turn_completion_explanation("empty_response_exhausted")
+        assert explainer  # sanity: the explainer text exists
+        job = {"id": "test-job", "name": "test", "prompt": "hello"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch(
+                 "hermes_cli.runtime_provider.resolve_runtime_provider",
+                 return_value={
+                     "api_key": "test-key",
+                     "base_url": "https://example.invalid/v1",
+                     "provider": "openrouter",
+                     "api_mode": "chat_completions",
+                 },
+             ), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent._format_turn_completion_explanation = (
+                AIAgent._format_turn_completion_explanation
+            )
+            mock_agent.run_conversation.return_value = {
+                "final_response": explainer,
+                "turn_exit_reason": "empty_response_exhausted",
+            }
+            mock_agent_cls.return_value = mock_agent
+            # Patch the class staticmethod the scheduler calls.
+            mock_agent_cls._format_turn_completion_explanation = (
+                AIAgent._format_turn_completion_explanation
+            )
+
+            success, output, final_response, error = run_job(job)
+
+        # The explainer is stripped to empty inside run_job; the downstream
+        # firing body (process_job) then suppresses delivery and marks the run
+        # a soft failure via its empty-response guard.  Here we assert the
+        # load-bearing transform: the "⚠️ No reply…" text never reaches delivery.
+        assert final_response == ""
+
+    def test_run_job_real_report_on_empty_reason_still_delivers(self, tmp_path):
+        """Defensive: a real report must NOT be suppressed even if the result
+        carries an abnormal turn_exit_reason — only the exact explainer text is."""
+        from run_agent import AIAgent
+        job = {"id": "test-job", "name": "test", "prompt": "hello"}
+        fake_db = MagicMock()
+
+        with patch("cron.scheduler._hermes_home", tmp_path), \
+             patch("cron.scheduler._resolve_origin", return_value=None), \
+             patch("dotenv.load_dotenv"), \
+             patch("hermes_state.SessionDB", return_value=fake_db), \
+             patch(
+                 "hermes_cli.runtime_provider.resolve_runtime_provider",
+                 return_value={
+                     "api_key": "test-key",
+                     "base_url": "https://example.invalid/v1",
+                     "provider": "openrouter",
+                     "api_mode": "chat_completions",
+                 },
+             ), \
+             patch("run_agent.AIAgent") as mock_agent_cls:
+            mock_agent = MagicMock()
+            mock_agent.run_conversation.return_value = {
+                "final_response": "Daily report: 4 PRs merged.",
+                "turn_exit_reason": "empty_response_exhausted",
+            }
+            mock_agent_cls.return_value = mock_agent
+            mock_agent_cls._format_turn_completion_explanation = (
+                AIAgent._format_turn_completion_explanation
+            )
+
+            success, output, final_response, error = run_job(job)
+
+        assert final_response == "Daily report: 4 PRs merged."
+        assert success is True
+
     def test_run_job_titles_cron_session_from_job_not_important_hint(self, tmp_path):
         # The cron session's first message is the injected "[IMPORTANT: …]"
         # hint, which used to surface as the sidebar/history row label. run_job
@@ -2296,6 +2378,52 @@ class TestSilentDelivery:
             tick(verbose=False)
         deliver_mock.assert_not_called()
 
+    def test_bracketless_silent_variants_suppress(self):
+        """Bracketless near-markers the model emits when it drops brackets
+        must still suppress delivery (#51438, #46917)."""
+        from cron.scheduler import tick
+        for marker in ("SILENT", "NO_REPLY", "NO REPLY", "no_reply"):
+            with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \
+                 patch("cron.scheduler.run_job", return_value=(True, "# output", marker, None)), \
+                 patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+                 patch("cron.scheduler._deliver_result") as deliver_mock, \
+                 patch("cron.scheduler.mark_job_run"):
+                tick(verbose=False)
+            deliver_mock.assert_not_called()
+
+    def test_report_quoting_marker_mid_sentence_still_delivers(self):
+        """A genuine report that merely mentions the token mid-sentence must
+        be delivered — the old substring check wrongly swallowed it."""
+        response = "I considered staying [SILENT] but here is the summary: 3 items merged."
+        with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \
+             patch("cron.scheduler.run_job", return_value=(True, "# output", response, None)), \
+             patch("cron.scheduler.save_job_output", return_value="/tmp/out.md"), \
+             patch("cron.scheduler._deliver_result") as deliver_mock, \
+             patch("cron.scheduler.mark_job_run"):
+            from cron.scheduler import tick
+            tick(verbose=False)
+        deliver_mock.assert_called_once()
+
+    def test_is_cron_silence_response_contract(self):
+        """Direct behavior contract for the cron silence matcher."""
+        from cron.scheduler import _is_cron_silence_response as sil
+        # Suppress: bare/bracketed/bracketless tokens, prefix, trailing-line.
+        assert sil("[SILENT]")
+        assert sil("[silent] nothing new")
+        assert sil("[SILENT] No changes detected")
+        assert sil("2 deals filtered.\n\n[SILENT]")
+        assert sil("SILENT")
+        assert sil("NO_REPLY")
+        assert sil("NO REPLY")
+        assert sil("Summary.\nSILENT")
+        # Deliver: real content, mid-sentence quotes, bare words, junk.
+        assert not sil("Daily report: 4 PRs merged.")
+        assert not sil("I stayed [SILENT] but here is the report: 3 items.")
+        assert not sil("Silent retry succeeded after 2 attempts.")
+        assert not sil("[SILENT")  # malformed open-bracket is not the sentinel
+        assert not sil("")
+        assert not sil("   \n\t ")
+
     def test_failed_job_always_delivers(self):
         """Failed jobs deliver regardless of [SILENT] in output."""
         with patch("cron.scheduler.get_due_jobs", return_value=[self._make_job()]), \