fix: cron and local DMs reach an open Desktop Bot Chat

agent/monitoring/cron_health.py

@@ -23,7 +23,7 @@
 logger = logging.getLogger(__name__)
 _KNOWN_STATUSES = {"claimed", "running", "completed", "failed", "unknown"}
 _KNOWN_SOURCES = {"builtin", "direct", "external"}
-_KNOWN_DELIVERY_OUTCOMES = {"delivered", "failed", "suppressed", "suppressed_acked", "not_configured"}
+_KNOWN_DELIVERY_OUTCOMES = {"queued", "delivered", "failed", "suppressed", "suppressed_acked", "not_configured"}
 _TERMINAL_STATUSES = {"completed", "failed", "unknown"}
 
 

contributors/emails/falcon.ortiz11@gmail.com

@@ -0,0 +1,2 @@
+FalconOrtiz
+# PR #101564 live-owner ingress salvage

cron/executions.py

@@ -74,6 +74,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None:
         "CREATE INDEX IF NOT EXISTS idx_executions_status_claimed "
         "ON executions(status, claimed_at DESC, id DESC)"
     )
+    add_column_if_missing(conn, "executions", "delivery_outcome", "delivery_outcome TEXT")
     add_column_if_missing(conn, "executions", "scheduled_instant", "scheduled_instant TEXT")
     conn.execute(
         "CREATE INDEX IF NOT EXISTS idx_executions_occurrence "
@@ -247,10 +248,10 @@ def finish_execution(
         cur = conn.execute(
             """UPDATE executions
                SET status=?, finished_at=?, error=?, handoff_pending=0,
-                   handoff_started_at=NULL
+                   handoff_started_at=NULL, delivery_outcome=?
                WHERE id=? AND status IN ('claimed','running')
                  AND process_id=? AND pid=?""",
-            (status, now, detail, execution_id, _PROCESS_ID, os.getpid()),
+            (status, now, detail, delivery_outcome, execution_id, _PROCESS_ID, os.getpid()),
         )
         if cur.rowcount != 1:
             return None

cron/scheduler.py

@@ -2599,9 +2599,12 @@ def _record_fire_ownership_lost(job_id: str, fire_owner: Optional[str], executio
 def _classify_delivery_outcome(
     *, delivery_error, should_deliver: bool, unresolved_origin: bool,
     normalized_deliver: str, incident_acked: bool, success: bool,
+    delivery_queued=None,
 ) -> str:
     if delivery_error:
         return "failed"
+    if should_deliver and delivery_queued:
+        return "queued"
     if should_deliver and unresolved_origin:
         return "not_configured"
     if should_deliver and normalized_deliver != "local":
@@ -2803,7 +2806,13 @@ def _finish_interrupted_run(job: dict, execution_id: str, delivery_error: Option
 def _finish_completed_run(d: _RunDelivery, fire_owner: Optional[str], execution_id: str) -> bool:
     """mark_job_run (owner-fenced) + execution ledger row for a run that reached delivery."""
     job = d.job
+    if not d.should_deliver and job.get("last_delivery_queued"):
+        from cron.jobs import update_job
+        update_job(job["id"], {"last_delivery_queued": None})
+        job["last_delivery_queued"] = None
     mark_kwargs = {"delivery_error": d.delivery_error}
+    if d.success and not d.delivery_error and d.should_deliver and job.get("last_delivery_queued"):
+        mark_kwargs["status"] = "delivery_queued"
     if fire_owner is not None:
         mark_kwargs["expected_fire_owner"] = fire_owner
     if d.blocked_config:
@@ -2816,6 +2825,7 @@ def _finish_completed_run(d: _RunDelivery, fire_owner: Optional[str], execution_
         return True
     delivery_outcome = _classify_delivery_outcome(
         delivery_error=d.delivery_error,
+        delivery_queued=job.get("last_delivery_queued"),
         should_deliver=d.should_deliver,
         unresolved_origin=d.unresolved_origin,
         # Read the lane the notice was actually routed through (failure_deliver on failure).
@@ -2861,7 +2871,8 @@ def _deliver_crash_failure(
     )
     delivery_outcome = _classify_delivery_outcome(
         delivery_error=delivery_error, should_deliver=True, unresolved_origin=unresolved_origin,
-        normalized_deliver=normalized_deliver, incident_acked=False, success=False)
+        normalized_deliver=normalized_deliver, incident_acked=False, success=False,
+        delivery_queued=job.get("last_delivery_queued"))
     if delivery_outcome in ("delivered", "not_configured"):
         _mark_incident_alerted(failure_incident_id)
     return delivery_error, delivery_outcome

cron/scheduler_delivery.py

@@ -642,12 +642,67 @@ def _get_bot_chat_delivery_timeout() -> int:
 
 
 def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]:
-    """Deliver job output into a profile's canonical Bot Chat as a real inbound user turn, via
-    ``hermes [-p <profile>] chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file`` — the
-    Bot Mode agent-to-agent lane, so canonical-session rules apply and it is alternation-safe.
-    ``profile`` is ``""`` for the job's own profile. None on success, else an error string."""
+    """Hand output to the live Bot Chat owner, or use the legacy unowned CLI lane.
+
+    None means completed; a queued/claimed receipt returns an explicit unverified status
+    string so existing Optional[str] callers cannot misreport admission as delivery.
+    ``profile`` is ``""`` for the job's own profile.
+    """
+    import hashlib
+    import json
     import tempfile
+    import uuid
+    from hermes_constants import get_hermes_home
+    from hermes_cli.profiles import get_profile_dir
+    from tools.bot_live_delivery import (
+        deliver_to_live_owner, find_canonical_live_owner, read_delivery_result,
+    )
+
     job_id = job.get("id", "?")
+    profile_label = profile or "(own)"
+    message = (
+        f'[Cronjob "{job.get("name", job_id)}" output — scheduled job, not the user. '
+        f"Review it, act on anything that needs action, and summarize "
+        f"for the chat.]\n\n{content}"
+    )
+    try:
+        source_home = get_hermes_home().resolve()
+        home = (get_profile_dir(profile) if profile else source_home).resolve()
+        # run_one_job/claim_fire attach the durable execution id before delivery. The
+        # transient fallback supports direct helper callers, never deduping recurring
+        # runs by their (potentially identical) output or previous last_run timestamp.
+        run_id = job.get("execution_id")
+        if not run_id:
+            run_id = job.setdefault("_bot_chat_run_id", uuid.uuid4().hex)
+        key = hashlib.sha256(json.dumps(
+            [str(source_home), job_id, str(run_id), str(home)],
+            ensure_ascii=False, separators=(",", ":"),
+        ).encode("utf-8")).hexdigest()
+        # Read BEFORE discovery: the previous owner may have exited after accepting.
+        # No receipt state, including ambiguous/failed, authorizes a CLI replay.
+        receipt = read_delivery_result(home, key)
+        if receipt is None:
+            owner = find_canonical_live_owner(home)
+            if owner is not None:
+                receipt = deliver_to_live_owner(home, owner, message, delivery_id=key)
+        if receipt is not None:
+            if receipt["message"] != message:
+                raise ValueError("delivery id already belongs to a different payload")
+            status = receipt["status"]
+            target = f"bot-chat:{profile_label}"
+            receipts = job.setdefault("_bot_chat_delivery_receipts", {})
+            receipts[target] = {"status": status, "delivery_id": key}
+            logger.info("Job '%s': Bot Chat %s receipt=%s status=%s",
+                        job_id, profile_label, key, status)
+            if status == "settled":
+                return None
+            detail = ("completion unverified; do not resend" if status in ("queued", "claimed")
+                      else receipt.get("error") or receipt.get("reason") or "not completed")
+            return f"{target} {status} (receipt {key}): {detail}"
+    except Exception as exc:
+        # Discovery/admission uncertainty must never open a second-writer fallback.
+        return f"bot-chat delivery to profile '{profile_label}' unverified: {exc}"
+
     hermes_bin = shutil.which("hermes")
     if hermes_bin:
         argv = [hermes_bin]
@@ -671,14 +726,9 @@ def _fail(msg: str, **log_kwargs) -> str:
         argv += ["-p", profile]
         # -p owns profile resolution; this scheduler's HERMES_HOME must not shadow it.
         env.pop("HERMES_HOME", None)
-
-    # Prefix marks this as scheduled output, not the human (Bot Mode sender-attribution).
-    message = (
-        f'[Cronjob "{job.get("name", job_id)}" output — scheduled job, not the user. '
-        f"Review it, act on anything that needs action, and summarize "
-        f"for the chat.]\n\n{content}"
-    )
-    profile_label = profile or "(own)"
+    else:
+        # Multiplex workers carry the profile in a ContextVar, not os.environ.
+        env["HERMES_HOME"] = str(source_home)
 
     query_file = None
     try:
@@ -998,14 +1048,21 @@ def _cron_delivery_notify_enabled(cfg: Optional[dict]) -> bool:
 
 def _record_delivery_verification(job: dict, unverified_targets: list) -> None:
     """Persist ``last_delivery_unverified``: list of ``platform:chat_id`` targets acked with no
-    evidence, or None. Skips the write when unchanged; never raises (bookkeeping must not fail a
+    evidence, or None, alongside queued Bot Chat receipts. Never raises (bookkeeping must not fail a
     delivery)."""
     new_value = list(unverified_targets) or None
-    if (job.get("last_delivery_unverified") or None) == new_value:
+    queued = {target: receipt for target, receipt in
+              job.get("_bot_chat_delivery_receipts", {}).items()
+              if receipt["status"] in ("queued", "claimed")} or None
+    values = {key: value for key, value in {
+        "last_delivery_unverified": new_value, "last_delivery_queued": queued,
+    }.items() if (job.get(key) or None) != value}
+    if not values:
         return
+    job.update(values)
     try:
         from cron.jobs import update_job
-        update_job(job["id"], {"last_delivery_unverified": new_value})
+        update_job(job["id"], values)
     except Exception as exc:  # pragma: no cover - defensive
         logger.debug("Job '%s': could not record delivery verification: %s", job.get("id"), exc)
 
@@ -1599,8 +1656,10 @@ def _deliver_result(
     running) the live adapter is tried first (E2EE rooms can't use the standalone HTTP path), then
     standalone fallback. ``for_failure=True`` routes failure-category notices through the job's
     ``failure_deliver`` override when present (NS-788). Returns None on success, else an error."""
+    job.pop("_bot_chat_delivery_receipts", None)
     targets = _resolve_delivery_targets(job, for_failure=for_failure)
     if not targets:
+        _record_delivery_verification(job, [])
         return _unresolved_delivery_outcome(job, for_failure)
 
     # Restart-safe workers have no live gateway adapters: hand the send back through a durable
@@ -1610,10 +1669,16 @@ def _deliver_result(
     # and that nested delivery must not be keyed under the outer execution id.
     external_execution = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER", "")
     if (external_execution and adapters is None
-            and external_execution == str(job.get("execution_id") or "")):
+            and external_execution == str(job.get("execution_id") or "")
+            and any(target["platform"] != BOT_CHAT_PLATFORM for target in targets)):
         from cron.delivery_queue import enqueue_and_wait
 
-        return enqueue_and_wait(external_execution, job, content, for_failure=for_failure)
+        _record_delivery_verification(job, [])
+        error = enqueue_and_wait(external_execution, job, content, for_failure=for_failure)
+        from cron.jobs import get_job
+        refreshed = get_job(job["id"]) or {}
+        job["last_delivery_queued"] = refreshed.get("last_delivery_queued")
+        return error
 
     from gateway.config import load_gateway_config
 
@@ -1677,12 +1742,16 @@ def _deliver_result(
 
     delivery_errors = []
     for target in targets:
-        # bot-chat targets bypass gateway adapters: output becomes an inbound turn in the target
-        # profile's Bot Chat via the chat CLI lane. Must precede the Platform enum, which lacks it.
+        # Bot Chat owns admission; never concurrently resume a live owner's transcript.
         if target["platform"] == BOT_CHAT_PLATFORM:
             bot_chat_error = _deliver_to_bot_chat(job, content, target["chat_id"])
             if bot_chat_error:
-                delivery_errors.append(bot_chat_error)
+                receipt_target = f"bot-chat:{target['chat_id'] or '(own)'}"
+                receipt = job.get("_bot_chat_delivery_receipts", {}).get(receipt_target)
+                if not receipt or receipt["status"] not in ("queued", "claimed"):
+                    delivery_errors.append(bot_chat_error)
+                if receipt and receipt["status"] == "ambiguous":
+                    unverified_targets.append(bot_chat_error)
             continue
 
         t = _prepare_target_delivery(

hermes_cli/cron.py

@@ -167,6 +167,8 @@ def _last_run_display(job: Dict[str, Any]) -> str:
     last_status = job["last_status"]
     if last_status == "ok":
         return color("ok", Colors.GREEN)
+    if last_status == "delivery_queued":
+        return color("delivery_queued: completion unverified; do not resend", Colors.YELLOW)
     if last_status == "delivery_failed":
         # Agent succeeded but the result never reached the user — not green; last_error is None.
         return color(f"delivery_failed: {job.get('last_delivery_error') or '?'}", Colors.YELLOW)
@@ -216,6 +218,8 @@ def _job_rows(job: Dict[str, Any]) -> List[tuple[str, str]]:
 def _job_warnings(job: Dict[str, Any]) -> List[str]:
     """Delivery / fire warning lines for one job in ``cron list``."""
     lines = []
+    if queued := job.get("last_delivery_queued"):
+        lines.append(f"Delivery queued (completion unverified; do not resend): {queued}")
     if job.get("last_delivery_error"):
         lines.append(f"{color('⚠ Delivery failed:', Colors.YELLOW)} {job['last_delivery_error']}")
     # A live adapter acked the last send but returned no message_id / raw_response
@@ -491,7 +495,7 @@ def _cron_doctor_issues_for_job(job: Dict[str, Any]) -> List[str]:
     issues: List[str] = []
     last_status = str(job.get("last_status") or "").strip().lower()
     # "delivery_failed" = the agent run succeeded; the delivery issue below reports it.
-    if last_status and last_status not in {"ok", "delivery_failed"}:
+    if last_status and last_status not in {"ok", "delivery_failed", "delivery_queued"}:
         issues.append(f"last run failed: {str(job.get('last_error') or 'unknown error').strip()}")
     if delivery_err := str(job.get("last_delivery_error") or "").strip():
         issues.append(f"last delivery failed: {delivery_err}")

tests/cron/test_cron_live_bot_delivery.py

@@ -0,0 +1,70 @@
+"""Cron admission must not become a second writer or a completed-delivery claim."""
+from pathlib import Path
+from unittest.mock import Mock
+
+from cron import scheduler_delivery as delivery
+from tools import bot_live_delivery as mailbox
+
+
+def test_live_delivery_retry_keeps_receipt_across_owner_loss(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    source = tmp_path / "custom-home"
+    monkeypatch.setenv("HERMES_HOME", str(source))
+    subprocess_run = Mock(side_effect=AssertionError("live owner must not spawn CLI"))
+    monkeypatch.setattr(delivery.subprocess, "run", subprocess_run)
+    from hermes_cli.profiles import get_profile_dir
+
+    for profile, home in [("", source), ("research", get_profile_dir("research"))]:
+        owner = dict(profile_home=str(home.resolve()), session_id="bot", lease_id="lease",
+                     live_session_id="live")
+        discovery = Mock(return_value=owner)
+        monkeypatch.setattr(mailbox, "find_canonical_live_owner", discovery)
+        job = dict(id="digest", name="Digest", execution_id="first-run")
+        pending = delivery._deliver_to_bot_chat(job, "payload", profile)
+        assert pending and "queued" in pending
+        records = list((home / "runtime/bot_live_delivery").glob("*.json"))
+        assert len(records) == 1
+        key = records[0].stem
+        record = mailbox.read_delivery_result(home, key)
+        assert record and record["message"].endswith("payload")
+        discovery.side_effect = AssertionError("receipt must precede discovery")
+        assert delivery._deliver_to_bot_chat(dict(job), "payload", profile) == pending
+        mailbox.claim_pending_delivery(home, owner)
+        mailbox.complete_delivery(home, key, status="ambiguous", error="owner died")
+        outcome = delivery._deliver_to_bot_chat(dict(job), "payload", profile)
+        assert outcome and "ambiguous" in outcome
+        discovery.side_effect = None
+        next_job = dict(job, execution_id="next-run")
+        outcome = delivery._deliver_to_bot_chat(next_job, "payload", profile)
+        assert outcome and "queued" in outcome
+        assert len(list((home / "runtime/bot_live_delivery").glob("*.json"))) == 2
+    subprocess_run.assert_not_called()
+
+
+def test_result_records_pending_until_terminal_receipt(tmp_path, monkeypatch):
+    from cron import jobs
+    from gateway import config
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.delenv("_HERMES_CRON_EXTERNAL_WORKER", raising=False)
+    owner = dict(profile_home=str(tmp_path.resolve()), session_id="bot", lease_id="lease",
+                 live_session_id="live")
+    monkeypatch.setattr(mailbox, "find_canonical_live_owner", lambda home: owner)
+    monkeypatch.setattr(delivery._sched, "load_config", lambda: {})
+    monkeypatch.setattr(config, "load_gateway_config", lambda: None)
+    monkeypatch.setattr(delivery.subprocess, "run", Mock(side_effect=AssertionError("CLI")))
+    updates = []
+    monkeypatch.setattr(jobs, "update_job", lambda key, values: updates.append(values))
+    job = dict(id="digest", execution_id="run", deliver="bot-chat")
+    error = delivery._deliver_result(job, "payload")
+    assert error is None
+    queued = updates[-1]["last_delivery_queued"]
+    assert queued and next(iter(queued.values()))["status"] == "queued"
+    assert delivery._sched._classify_delivery_outcome(
+        delivery_error=error, delivery_queued=queued, should_deliver=True, unresolved_origin=False,
+        normalized_deliver="bot-chat", incident_acked=False, success=True) == "queued"
+    record = mailbox.claim_pending_delivery(tmp_path, owner)
+    assert record is not None
+    mailbox.complete_delivery(tmp_path, record["delivery_id"], status="settled", reply="done")
+    assert delivery._deliver_result(job, "payload") is None
+    assert updates[-1]["last_delivery_queued"] is None