fix(cron): keep a run's real outcome when an unknown-fenced delivery meets a claim miss

cron/scheduler.py

@@ -3430,11 +3430,18 @@ def _teardown_deferred() -> None:
             # left with its real error — so fall through to _finish_completed_run, whose owner-fenced
             # mark_job_run is authoritative either way. An explicit transport cancel stays fail-closed.
             transport_cancelled = fence.transport_cancelled()
-            if d.delivery_attempted and not d.delivery_error and not transport_cancelled:
+            # #132078: a delivery fenced ``unknown`` by the worker's wait budget is uncertain,
+            # not failed — the gateway's at-most-once send may still land — so it earns the same
+            # fall-through; the queue row is where that uncertainty lives, not the run's status.
+            uncertain_delivery = (
+                d.delivery_attempted and bool(job.get("_delivery_outcome_unknown")))
+            if (d.delivery_attempted and (not d.delivery_error or uncertain_delivery)
+                    and not transport_cancelled):
                 logger.warning(
-                    "Job '%s': fire claim ownership lost after completed delivery; "
+                    "Job '%s': fire claim ownership lost after %s delivery; "
                     "recording the delivered run's terminal status",
-                    job["id"])
+                    job["id"],
+                    "a completed" if not d.delivery_error else "an outcome-unknown")
             else:
                 if transport_cancelled:
                     logger.warning(

cron/scheduler_delivery.py

@@ -174,26 +174,6 @@ def _inchannel_seed_allowed(*, is_dm: bool, user_id: Optional[str]) -> bool:
     return bool(is_dm or user_id)
 
 
-def _redact_cron_payload(text: str, what: str) -> str:
-    """Fail-closed secret redaction for anything a cron job emits outward.
-
-    Every outward lane — chat message, session mirror, bot-chat turn — must apply the same policy,
-    so the policy lives in one place. ``force=True`` because this is a safety boundary, not
-    logging: the ``security.redact_secrets`` preference governs how much is scrubbed from the
-    user's own logs and must not be able to turn scrubbing off on the way out to a chat (same
-    reasoning as ``tools/delegation_live_log.py``). Empty input is returned as-is; any failure
-    inside the redactor replaces the payload entirely rather than letting an unscanned value out.
-    """
-    if not text:
-        return text
-    try:
-        from agent.redact import redact_sensitive_text
-        return redact_sensitive_text(text, force=True)
-    except Exception as e:
-        logger.warning("Failed to redact secrets from cron %s: %s", what, e)
-        return "[REDACTED - redaction failed]"
-
-
 def _cron_display_name(job: dict) -> str:
     """Job name/id as it appears in outward-facing text. The mirror sinks and the thread title
     splice the job *name* around the redacted payload, and the name is user-controlled config — a
@@ -1979,6 +1959,7 @@ def _deliver_result(
     ``failure_deliver`` override when present (NS-788). Returns None on success, else an error."""
     job.pop("_bot_chat_delivery_receipts", None)
     job.pop("_notification_all_targets_suppressed", None)
+    job.pop("_delivery_outcome_unknown", None)
     targets = _resolve_delivery_targets(job, for_failure=for_failure)
     if not targets:
         _record_delivery_verification(job, [])
@@ -2001,6 +1982,12 @@ def _deliver_result(
         delivery_status = get_status(external_execution)
         if delivery_status and delivery_status["status"] == "suppressed":
             job["_notification_all_targets_suppressed"] = True
+        if delivery_status and delivery_status["status"] == "unknown":
+            # The queue fenced the send uncertain (the worker's wait budget ran out mid-send);
+            # the gateway's at-most-once attempt may still land, so the returned error says
+            # "unknown", not "failed" — the scheduler must not read it as a delivery failure
+            # (#132078).
+            job["_delivery_outcome_unknown"] = True
         from cron.jobs import get_job
         refreshed = get_job(job["id"]) or {}
         job["last_delivery_queued"] = refreshed.get("last_delivery_queued")
@@ -2129,3 +2116,7 @@ def _deliver_result(
 from cron import scheduler_delivery_origin as _origin
 from cron import scheduler_preflight as _preflight
 from cron import scheduler_script as _script
+
+# Split out for the file-size ratchet; re-exported here so every existing caller (and any
+# monkeypatch of this module's name) keeps working unchanged.
+from cron.scheduler_delivery_redact import _redact_cron_payload as _redact_cron_payload

cron/scheduler_delivery_redact.py

@@ -0,0 +1,25 @@
+"""Fail-closed redaction for outward cron payloads (split from scheduler_delivery)."""
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def _redact_cron_payload(text: str, what: str) -> str:
+    """Fail-closed secret redaction for anything a cron job emits outward.
+
+    Every outward lane — chat message, session mirror, bot-chat turn — must apply the same policy,
+    so the policy lives in one place. ``force=True`` because this is a safety boundary, not
+    logging: the ``security.redact_secrets`` preference governs how much is scrubbed from the
+    user's own logs and must not be able to turn scrubbing off on the way out to a chat (same
+    reasoning as ``tools/delegation_live_log.py``). Empty input is returned as-is; any failure
+    inside the redactor replaces the payload entirely rather than letting an unscanned value out.
+    """
+    if not text:
+        return text
+    try:
+        from agent.redact import redact_sensitive_text
+        return redact_sensitive_text(text, force=True)
+    except Exception as e:
+        logger.warning("Failed to redact secrets from cron %s: %s", what, e)
+        return "[REDACTED - redaction failed]"

tests/cron/test_fire_claim_lost_after_delivery.py

@@ -264,3 +264,82 @@ def test_confirmed_claim_loss_mid_run_still_yields(temp_home, monkeypatch):
     assert record["last_status"] is None
     assert record["fire_claim"]["by"] == "replacement:deadbeef"
     assert "discarded" in get_execution(job["execution_id"])["error"]
+
+
+_UNKNOWN_DELIVERY_ERROR = (
+    "timed out while gateway delivery was in progress; outcome is unknown and was not retried"
+)
+
+
+@pytest.mark.parametrize(
+    "run_result, expected_status, expected_error, expected_streak",
+    [
+        ((True, "output text", "the report", None), "delivery_failed", None, 0),
+        ((False, "output text", "", _REAL_ERROR), "error", _REAL_ERROR, 1),
+    ],
+    ids=["unknown-delivered-ok", "unknown-delivered-failure-notice"],
+)
+def test_unknown_fenced_delivery_keeps_terminal_status_when_claim_sample_misses(
+    temp_home, monkeypatch, run_result, expected_status, expected_error, expected_streak,
+):
+    """#132078: the restart-safe worker's wait budget fences the send ``unknown`` while the
+    gateway is still mid-send; the message lands anyway, then the runner tail dies and the
+    post-delivery claim sample misses. The run must keep its real outcome — an ok run stays a
+    delivery-only blemish (streak untouched), a failed run keeps its real error — instead of
+    recording ``_OWNERSHIP_LOST_INTERRUPTED`` for a shutdown that never happened."""
+    from cron.jobs import get_job
+
+    sched, job, hb, delivered = _drive(
+        monkeypatch, run_result=run_result, samples_before_miss=2)
+
+    def fence_unknown(delivery_job, content, **kwargs):
+        delivered.append(content)
+        delivery_job["_delivery_outcome_unknown"] = True
+        return _UNKNOWN_DELIVERY_ERROR
+
+    monkeypatch.setattr(sched, "_deliver_result", fence_unknown)
+
+    assert sched.run_one_job(job) is True
+
+    assert len(delivered) == 1, "the notice had left the worker process"
+    assert hb.missed == 1, "exactly the post-delivery sample missed"
+    record = get_job(job["id"])
+    assert record["last_status"] == expected_status
+    assert record["last_error"] == expected_error
+    assert record["last_error"] != sched._OWNERSHIP_LOST_INTERRUPTED
+    assert record["failure_streak"] == expected_streak
+    assert record["last_delivery_error"] == _UNKNOWN_DELIVERY_ERROR
+
+
+def test_worker_wait_timeout_stamps_the_unknown_delivery_outcome(monkeypatch, tmp_path):
+    """The worker's queue branch marks a delivery the queue fenced ``unknown`` on the job, so the
+    post-delivery ownership check can tell "uncertain" from "failed"; a later confirmed outcome
+    re-derives the marker per attempt (#132078)."""
+    from cron import scheduler_delivery
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(
+        "cron.delivery_queue.enqueue_and_wait",
+        lambda execution_id, job, content, **kw: _UNKNOWN_DELIVERY_ERROR)
+    monkeypatch.setattr(
+        "cron.delivery_queue.get_status",
+        lambda execution_id: {"status": "unknown", "error": _UNKNOWN_DELIVERY_ERROR})
+    monkeypatch.setattr("cron.jobs.get_job", lambda job_id: {})
+    monkeypatch.setattr(
+        scheduler_delivery, "_resolve_delivery_targets",
+        lambda job, for_failure=False: [{"platform": "telegram", "chat_id": "123"}])
+    monkeypatch.setenv("_HERMES_CRON_EXTERNAL_WORKER", "exec-1")
+
+    job = {"id": "job-1", "execution_id": "exec-1", "deliver": "telegram:123"}
+    assert scheduler_delivery._deliver_result(
+        job, "done", adapters=None, loop=None) == _UNKNOWN_DELIVERY_ERROR
+    assert job.get("_delivery_outcome_unknown") is True
+
+    monkeypatch.setattr(
+        "cron.delivery_queue.enqueue_and_wait",
+        lambda execution_id, job, content, **kw: None)
+    monkeypatch.setattr(
+        "cron.delivery_queue.get_status",
+        lambda execution_id: {"status": "delivered", "error": None})
+    assert scheduler_delivery._deliver_result(job, "done", adapters=None, loop=None) is None
+    assert job.get("_delivery_outcome_unknown") is None