fix(cron): degraded-delivery marker is recognised by its record, not its text

The recursion guard for the timeout marker matched 'DELIVERY DEGRADED' in the
payload, so a job whose own output contained that phrase would silently lose its
notice. The deferred record now carries an explicit degraded flag and the guard
reads it. The marker's id is a fresh sha256 digest of '<key>:degraded' instead of
'<key>-degraded': deferred ids double as live-owner delivery ids, which
tools.bot_live_delivery._delivery_id requires to be 32-64 hex characters.

test_bot_chat_pending documents the deliberate extra turn: a timeout drains one
short marker turn, whose own timeout queues nothing more.

cron/bot_chat_delivery.py

@@ -57,7 +57,9 @@ def _records(root: Path) -> list[tuple[Path, dict]]:
 
 
 def defer(key: str, job: dict, content: str, profile: str, home: Path, *,
-          for_failure: bool = False, suppressed: bool = False) -> dict:
+          for_failure: bool = False, suppressed: bool = False, degraded: bool = False) -> dict:
+    """``degraded`` marks the short notice queued after a CLI-lane turn timed out; the record
+    carries it so the consumer recognizes the marker by the record, never by its text."""
     root = _root()
     root.mkdir(parents=True, exist_ok=True, mode=0o700)
     with _FileLock(root / ".lock"):
@@ -75,6 +77,8 @@ def defer(key: str, job: dict, content: str, profile: str, home: Path, *,
                       profile=profile, home=str(home), sequence=sequence)
         if for_failure:
             record["for_failure"] = True
+        if degraded:
+            record["degraded"] = True
         atomic_json_write(root / f"{key}.json", record, fsync_dir=True, mode=0o600)
         return record
 

cron/scheduler_delivery.py

@@ -989,9 +989,11 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str, *, deferred: Opt
         # knew to look). Fail-safe middle: queue a SHORT degraded-delivery marker through
         # the deferred lane — it references the saved output instead of repeating it, so a
         # late-completing original turn cannot duplicate content. One marker per execution
-        # (stable key); a marker that itself times out is never re-marked (recursion guard).
+        # (stable key); a marker that itself times out is never re-marked: the guard reads the
+        # deferred record's ``degraded`` flag, never the payload text (a job whose output
+        # happens to contain the marker phrase must still get its notice).
         marker_queued = False
-        if "DELIVERY DEGRADED" not in content:
+        if not (deferred or {}).get("degraded"):
             marker = (
                 f"[Cronjob \"{job.get('name', job_id)}\" — DELIVERY DEGRADED, scheduled job, "
                 f"not the user. This alert's bot-chat turn timed out after "
@@ -1001,8 +1003,12 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str, *, deferred: Opt
             )
             try:
                 from cron.bot_chat_delivery import defer as _defer_marker
-                _defer_marker(f"{key}-degraded", dict(job), marker, profile, home,
-                              for_failure=for_failure)
+                # Deferred ids double as live-owner delivery ids, which must be 32-64 hex
+                # chars (tools.bot_live_delivery._delivery_id) — so the marker's id is a
+                # fresh digest derived from the execution key, not a suffixed one.
+                marker_key = hashlib.sha256(f"{key}:degraded".encode("utf-8")).hexdigest()
+                _defer_marker(marker_key, dict(job), marker, profile, home,
+                              for_failure=for_failure, degraded=True)
                 marker_queued = True
             except Exception as defer_exc:
                 logger.warning(

tests/cron/test_bot_chat_pending.py

@@ -36,7 +36,11 @@ def test_cli_owner_deferral_and_attempt_fence(tmp_path, monkeypatch, error):
         assert queue.read_pending(key)["status"] == expected
         queue.drain()
         delivery._deliver_to_bot_chat(job, "output", "")
-        assert run.call_count == 1
+        # The failed turn is never retried; a timeout additionally drains ONE short
+        # degraded-delivery marker (its own turn), and its timeout queues nothing more.
+        assert run.call_count == (2 if error else 1)
+        markers = [r for _, r in queue._records(queue._root()) if r.get("degraded")]
+        assert len(markers) == (1 if error else 0)
     finally:
         lease.release()
         db.close()