From e5d3c231aa7e13dadf23b1c325d32f0197f67c3d Mon Sep 17 00:00:00 2001 From: FalconOrtiz Date: Mon, 7 Sep 2026 16:41:12 -0700 Subject: [PATCH 1/2] fix: admit Bot Chat deliveries through the live session owner Adapt FalconOrtiz's owner-mailbox proposal from #101564 onto the current notification poller and topical modules. The durable mailbox is cross-process ingress only: the existing owner admits its normal prompt turn after the current turn and human FIFO clear. Retain immutable receipts, stable admission identities, capability and lease fencing, compression lineage, and disable blind recovery replay of imported turns. The original stale server hunks and expiring receipt protocol were rebuilt rather than cherry-picked: the current facade decomposition and durable busy admission contract differ. Credit the earlier owner-mailbox work in #100544 and durable producer work in #100319. Co-authored-by: fangliquanflq Co-authored-by: 686f6c61 --- contributors/emails/falcon.ortiz11@gmail.com | 2 + tests/tools/test_bot_live_owner_delivery.py | 95 ++++++++ tests/tui_gateway/test_auto_continue.py | 4 +- .../test_bot_live_owner_delivery.py | 89 +++++++ tools/bot_live_delivery.py | 223 ++++++++++++++++++ tui_gateway/prompt_turn.py | 12 +- tui_gateway/session_auto_continue.py | 2 + tui_gateway/session_lifecycle.py | 5 +- tui_gateway/session_notifications.py | 55 +++++ tui_gateway/turn_marker.py | 9 +- 10 files changed, 486 insertions(+), 10 deletions(-) create mode 100644 contributors/emails/falcon.ortiz11@gmail.com create mode 100644 tests/tools/test_bot_live_owner_delivery.py create mode 100644 tests/tui_gateway/test_bot_live_owner_delivery.py create mode 100644 tools/bot_live_delivery.py diff --git a/contributors/emails/falcon.ortiz11@gmail.com b/contributors/emails/falcon.ortiz11@gmail.com new file mode 100644 index 0000000000000..caffee26b4970 --- /dev/null +++ b/contributors/emails/falcon.ortiz11@gmail.com @@ -0,0 +1,2 @@ +FalconOrtiz +# PR #101564 live-owner ingress salvage diff --git a/tests/tools/test_bot_live_owner_delivery.py b/tests/tools/test_bot_live_owner_delivery.py new file mode 100644 index 0000000000000..eacecbc72fd5b --- /dev/null +++ b/tests/tools/test_bot_live_owner_delivery.py @@ -0,0 +1,95 @@ +"""Durable mailbox invariants, using real disk and exec boundaries.""" +import json +import os +import subprocess +import sys + +import pytest + + +@pytest.mark.parametrize("terminal_status", ["settled", "failed", "cancelled"]) +def test_delivery_is_idempotent_fenced_and_permanent(tmp_path, terminal_status): + from tools import bot_live_delivery as mailbox + + owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat", + lease_id="lease", live_session_id="live") + delivery_id = "a" * 32 + queued = mailbox.deliver_to_live_owner(tmp_path, owner, "hello", delivery_id=delivery_id) + assert queued["status"] == "queued" + assert mailbox.deliver_to_live_owner(tmp_path, owner, "hello", delivery_id=delivery_id) == queued + with pytest.raises(ValueError): + mailbox.deliver_to_live_owner(tmp_path, owner, "different", delivery_id=delivery_id) + assert mailbox.claim_pending_delivery(tmp_path, dict(owner, lease_id="other")) is None + assert mailbox.claim_pending_delivery(tmp_path, dict(owner, live_session_id="other")) is None + script = ( + "import json,sys; from tools.bot_live_delivery import claim_pending_delivery; " + "print(json.dumps(claim_pending_delivery(sys.argv[1],json.loads(sys.argv[2]))))" + ) + children = [subprocess.Popen([sys.executable, "-c", script, str(tmp_path), json.dumps(owner)], + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) for _ in range(2)] + results = [] + for child in children: + out, err = child.communicate(timeout=30) + assert child.returncode == 0, err + results.append(json.loads(out)) + claims = [r for r in results if r is not None] + assert len(claims) == 1 and claims[0]["message"] == "hello" + assert mailbox.read_delivery_result(tmp_path, delivery_id)["status"] == "claimed" + assert mailbox.claim_pending_delivery(tmp_path, owner) is None + receipt = mailbox.complete_delivery(tmp_path, delivery_id, status=terminal_status, reply="answer") + assert mailbox.read_delivery_result(tmp_path, delivery_id) == receipt + assert mailbox.complete_delivery(tmp_path, delivery_id, status=terminal_status, reply="answer") == receipt + with pytest.raises(ValueError): + mailbox.complete_delivery(tmp_path, delivery_id, status=terminal_status, reply="rewrite") + assert mailbox.deliver_to_live_owner(tmp_path, owner, "hello", delivery_id=delivery_id) == receipt + assert mailbox.claim_pending_delivery(tmp_path, owner) is None + if os.name != "nt": + for path in (tmp_path / "runtime" / mailbox.DELIVERY_DIR_NAME).iterdir(): + assert path.stat().st_mode & 0o077 == 0 + + +def test_fifo_survives_clock_rollback(tmp_path, monkeypatch): + from tools import bot_live_delivery as mailbox + + owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat", + lease_id="lease", live_session_id="live") + for timestamp, message in ((100, "first"), (90, "second")): + monkeypatch.setattr(mailbox.time, "time_ns", lambda: timestamp) + mailbox.deliver_to_live_owner(tmp_path, owner, message) + assert mailbox.claim_pending_delivery(tmp_path, owner)["message"] == "first" + assert mailbox.claim_pending_delivery(tmp_path, owner)["message"] == "second" + + +@pytest.mark.parametrize("capable", [True, False]) +def test_only_canonical_capable_owner_receives_across_compression(tmp_path, capable): + from hermes_state import SessionDB + from hermes_cli.active_sessions import try_acquire_active_session, transfer_active_session + from tools import bot_live_delivery as mailbox + + db = SessionDB(db_path=tmp_path / "state.db") + db.create_session(session_id="chat", source="cli") + db.set_session_title("chat", "Bot Chat") + meta = dict(live_session_id="live", bot_live_delivery_consumer=capable) + lease, refusal = try_acquire_active_session(session_id="chat", surface="desktop", config={}, + registry_home=tmp_path, metadata=meta) + assert refusal is None + try: + owner = mailbox.find_canonical_live_owner(tmp_path) + if not capable: + assert owner is None + return + assert owner["lease_id"] == lease.lease_id + queued = mailbox.deliver_to_live_owner(tmp_path, owner, "before compression") + db.end_session("chat", "compression") + db.create_session(session_id="tip", source="cli", parent_session_id="chat") + assert transfer_active_session(lease, session_id="tip", metadata=meta) + current = mailbox.find_canonical_live_owner(tmp_path) + assert current["session_id"] == "tip" + claim = mailbox.claim_pending_delivery(tmp_path, current) + assert claim["delivery_id"] == queued["delivery_id"] + assert claim["session_id"] == "chat" + assert mailbox.claim_pending_delivery(tmp_path, current) is None + finally: + lease.release() + db.close() diff --git a/tests/tui_gateway/test_auto_continue.py b/tests/tui_gateway/test_auto_continue.py index 922cd629610fe..213a925db483c 100644 --- a/tests/tui_gateway/test_auto_continue.py +++ b/tests/tui_gateway/test_auto_continue.py @@ -181,12 +181,12 @@ def test_interrupt_racing_marker_write_cannot_leave_recovery_state( session = _session(agent=agent, running=True) _patch_local_interrupt(monkeypatch, session) - def write_after_stop(home, key, prompt, *, attempts=0): + def write_after_stop(home, key, prompt, *, attempts=0, auto_continue=True): response = server._methods["session.interrupt"]( "stop-during-write", {"session_id": "runtime-race"} ) assert response["result"]["status"] == "interrupted" - record_turn_start(home, key, prompt, attempts=attempts) + record_turn_start(home, key, prompt, attempts=attempts, auto_continue=auto_continue) monkeypatch.setattr(server, "record_turn_start", write_after_stop) diff --git a/tests/tui_gateway/test_bot_live_owner_delivery.py b/tests/tui_gateway/test_bot_live_owner_delivery.py new file mode 100644 index 0000000000000..9484d49dc5e73 --- /dev/null +++ b/tests/tui_gateway/test_bot_live_owner_delivery.py @@ -0,0 +1,89 @@ +"""Imported turns retain their receipt and cannot bypass the local FIFO.""" +import threading +from types import SimpleNamespace + +from tui_gateway.method_ctx import rebind +from tui_gateway import session_notifications, session_auto_continue +from tui_gateway.turn_marker import record_turn_start, read_turn_marker + + +def test_refused_input_commits_failed_mailbox_receipt(tmp_path): + import contextvars + import logging + import time + from tui_gateway import prompt_turn + from tools import bot_live_delivery as mailbox + + owner = dict(profile_home=str(tmp_path.resolve()), session_id="chat", + lease_id="lease", live_session_id="live") + queued = mailbox.deliver_to_live_owner(tmp_path, owner, "refused input") + mailbox.claim_pending_delivery(tmp_path, owner) + agent = SimpleNamespace(session_id="chat") + session = dict(agent=agent, session_key="chat", history_lock=threading.RLock(), running=True) + retired = [] + noop = lambda *args, **kwargs: None + submit = rebind(prompt_turn._run_prompt_submit, { + "threading": threading, "time": time, "logger": logging.getLogger(__name__), + "_sessions_lock": threading.RLock(), "_sessions": {}, + "_admit_prompt_turn": lambda *args: ([], agent), + "_emit": noop, "bind_transport": noop, "reset_transport": noop, + "_current_runtime_session_record": contextvars.ContextVar("refused_turn"), + "_TurnRun": prompt_turn._TurnRun, + "_record_turn_marker": lambda *args, **kwargs: "marker", + "_prepare_turn_input": lambda *args: None, + "_finish_turn": noop, "_clear_inflight_turn": noop, + "_retire_turn_marker": lambda *args: retired.append(args), + "_emit_settled_session_info": noop, + }) + def terminal(outcome): + mailbox.complete_delivery(tmp_path, queued["id"], status=outcome["status"], + error=outcome.get("error", "")) + assert submit(None, "live", session, "refused input", terminal_callback=terminal) + session["_run_thread"].join(timeout=5) + assert not session["_run_thread"].is_alive() + assert mailbox.read_delivery_result(tmp_path, queued["id"])["status"] == "failed" + assert retired and session["running"] is False + + +def test_imported_crash_marker_never_autocontinues(tmp_path): + record_turn_start(tmp_path, "chat", "imported", auto_continue=False) + marker = read_turn_marker(tmp_path, "chat") + assert marker["auto_continue"] is False + schedule = rebind(session_auto_continue._maybe_schedule_auto_continue, { + "_session_home": lambda session: tmp_path, + "read_turn_marker": read_turn_marker, + }) + assert schedule("live", {}, "chat") is None + + +def test_local_work_blocks_mailbox_claim_without_consuming_envelope(monkeypatch, tmp_path): + import tools.bot_live_delivery as mailbox + owner = {"lease_id": "lease", "live_session_id": "live", "session_id": "chat"} + pending = [{"id": "receipt", "message": "imported"}] + monkeypatch.setattr(mailbox, "find_canonical_live_owner", lambda home: owner) + monkeypatch.setattr(mailbox, "claim_pending_delivery", lambda home, pinned: pending.pop(0)) + receipts = [] + monkeypatch.setattr(mailbox, "complete_delivery", lambda *args, **kwargs: receipts.append((args, kwargs))) + submitted = [] + def submit(rid, sid, session, text, **kwargs): + submitted.append(text) + kwargs["terminal_callback"]({"status": "settled", "text": "reply"}) + return True + poll = rebind(session_notifications._poll_bot_live_delivery_once, { + "_session_home": lambda session: tmp_path, + "_run_prompt_submit": submit, + "_notif_release_turn": lambda session: session.update(running=False), + }) + session = {"history_lock": threading.RLock(), "agent": object(), "session_key": "chat", + "active_session_lease": SimpleNamespace(lease_id="lease", released=False)} + for blocker in ("running", "queued_prompt", "queued_prompts", "_auto_continue_scheduled"): + session[blocker] = True + assert poll("live", session) is False + assert pending and not submitted + session.pop(blocker) + assert poll("other-live", session) is False + assert pending + assert poll("live", session) is True + assert submitted == ["imported"] and not pending + assert receipts[0][0][1] == "receipt" + assert receipts[0][1]["reply"] == "reply" diff --git a/tools/bot_live_delivery.py b/tools/bot_live_delivery.py new file mode 100644 index 0000000000000..7a195377900e7 --- /dev/null +++ b/tools/bot_live_delivery.py @@ -0,0 +1,223 @@ +"""Durable, at-most-once handoff to an existing Bot Chat owner. + +Adapted from FalconOrtiz's live-owner mailbox (#101564). A single private +record advances queued -> claimed -> terminal under a process-shared lock. +Claims never expire: a crashed consumer leaves an inspectable unknown outcome, +not permission to execute the same input again. Receipts are permanent. +""" +from __future__ import annotations + +import json +import os +import re +import tempfile +import time +import uuid +from contextlib import contextmanager +from pathlib import Path +from typing import Any + +from hermes_cli.active_sessions import _FileLock + +DELIVERY_DIR_NAME = "bot_live_delivery" +_OWNER_KEYS = ("profile_home", "session_id", "lease_id", "live_session_id") +_TERMINAL = frozenset({"settled", "failed", "cancelled", "ambiguous"}) + + +def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None: + """Resolve exact Bot Chat's compression tip without creating/migrating its DB. + + Capability advertisement is mandatory; old Desktop/TUI processes must not + receive work they cannot consume. Registry errors propagate, failing closed. + """ + from hermes_cli.active_sessions import active_session_registry_snapshot + from hermes_state import SessionDB + + home = Path(profile_home).resolve() + if not (home / "state.db").is_file(): + return None + db = SessionDB(db_path=home / "state.db", read_only=True) + try: + row = db.get_session_by_title("Bot Chat") + session_id = db.get_compression_tip(row["id"]) if row else None + finally: + db.close() + if not session_id: + return None + for entry in active_session_registry_snapshot(registry_home=home): + meta = entry.get("metadata") or {} + if (entry["session_id"] == session_id + and meta.get("bot_live_delivery_consumer") is True + and meta.get("live_session_id")): + return dict(profile_home=str(home), session_id=session_id, + lease_id=entry["lease_id"], live_session_id=meta["live_session_id"]) + return None + + +def _owner(home: Path | str, owner: dict[str, Any]) -> dict[str, str]: + pinned = {key: owner.get(key) for key in _OWNER_KEYS} + if not all(isinstance(value, str) and value for value in pinned.values()): + raise ValueError("owner requires profile_home, session_id, lease_id and live_session_id") + if pinned["profile_home"] != str(Path(home).resolve()): + raise ValueError("owner belongs to a different profile home") + return pinned + + +def _delivery_id(value: str) -> str: + if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{32,64}", value) is None: + raise ValueError("delivery id must be 32 to 64 lowercase hex characters") + return value + + +def _root(home: Path | str) -> Path: + return Path(home).resolve() / "runtime" / DELIVERY_DIR_NAME + + +def _fsync_dir(path: Path) -> None: + # Windows cannot open directories with os.open; file fsync still applies. + if os.name == "nt": + return + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +@contextmanager +def _locked(home: Path | str): + root = _root(home) + root.parent.mkdir(parents=True, exist_ok=True) + root.mkdir(mode=0o700, exist_ok=True) + root.chmod(0o700) + _fsync_dir(root.parent) + _fsync_dir(root.parent.parent) + lock = root / ".lock" + fd = os.open(lock, os.O_CREAT | os.O_WRONLY, 0o600) + os.close(fd) + with _FileLock(lock): + yield root + + +def _read(path: Path) -> dict[str, Any] | None: + try: + return json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + + +def _write(path: Path, record: dict[str, Any]) -> None: + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".delivery-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(record, stream, ensure_ascii=False, sort_keys=True) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + _fsync_dir(path.parent) + finally: + Path(temporary).unlink(missing_ok=True) + + +def deliver_to_live_owner( + profile_home: Path | str, owner: dict[str, Any], message: str, + *, delivery_id: str | None = None, +) -> dict[str, Any]: + """Return durable admission immediately, without waiting for the owner. + + Retry with the same id AND pinned owner/message to inspect the existing + state. Reusing an id with a different payload is an error, never an overwrite. + """ + pinned = _owner(profile_home, owner) + if not isinstance(message, str): + raise ValueError("message must be a string") + key = _delivery_id(delivery_id if delivery_id is not None else uuid.uuid4().hex) + with _locked(profile_home) as root: + path = root / f"{key}.json" + existing = _read(path) + if existing is not None: + if existing["owner"] != pinned or existing["message"] != message: + raise ValueError("delivery id already belongs to a different payload") + return existing + # Wall time can roll back. Permanent receipts retain the admission + # high-water mark, allocated while holding the cross-process lock. + sequence = max((record.get("sequence", record["created_at"]) + for candidate in root.glob("*.json") + if (record := _read(candidate)) is not None), default=0) + 1 + record = dict(delivery_id=key, id=key, owner=pinned, **pinned, + message=message, status="queued", created_at=time.time_ns(), + sequence=sequence) + _write(path, record) + return record + + +def _matches(home: Path | str, record: dict, owner: dict) -> bool: + pinned = record["owner"] + if any(pinned[key] != owner[key] for key in ("profile_home", "lease_id", "live_session_id")): + return False + if pinned["session_id"] == owner["session_id"]: + return True + from hermes_state import SessionDB + + db = SessionDB(db_path=Path(home) / "state.db", read_only=True) + try: + return db.get_compression_tip(pinned["session_id"]) == owner["session_id"] + finally: + db.close() + + +def claim_pending_delivery( + profile_home: Path | str, owner: dict[str, Any], +) -> dict[str, Any] | None: + """Claim oldest matching input exactly once; caller supplies its current lease. + + A lease transfer across compression is accepted only along the original + stored session's compression chain. A new lease/live session cannot steal it. + Caller must hold its normal turn-admission guard before invoking this. + """ + current = _owner(profile_home, owner) + if not _root(profile_home).is_dir(): + return None + with _locked(profile_home) as root: + pending = [] + for path in root.glob("*.json"): + record = _read(path) + if record is not None and record["status"] == "queued" and _matches(profile_home, record, current): + pending.append(record) + if not pending: + return None + record = min(pending, key=lambda item: ( + item.get("sequence", item["created_at"]), item["delivery_id"])) + record.update(status="claimed", claimed_at=time.time_ns()) + _write(root / f"{record['delivery_id']}.json", record) + return record + + +def complete_delivery( + profile_home: Path | str, delivery_id: str, *, status: str, + reply: str = "", error: str = "", reason: str = "", +) -> dict[str, Any]: + """Persist an immutable terminal receipt; duplicate identical completion is safe.""" + key = _delivery_id(delivery_id) + if status not in _TERMINAL: + raise ValueError("invalid terminal delivery status") + outcome = dict(status=status, reply=reply, error=error, reason=reason) + with _locked(profile_home) as root: + path = root / f"{key}.json" + record = _read(path) + if record is None: + raise FileNotFoundError(f"delivery not found: {key}") + if record["status"] in _TERMINAL: + if any(record.get(k) != v for k, v in outcome.items()): + raise ValueError("delivery already has a different terminal receipt") + return record + if record["status"] != "claimed": + raise ValueError("delivery must be claimed before completion") + record.update(outcome, completed_at=time.time_ns()) + _write(path, record) + return record + + +def read_delivery_result(profile_home: Path | str, delivery_id: str) -> dict[str, Any] | None: + """Read admission/claim/terminal state without waiting or deleting its receipt.""" + return _read(_root(profile_home) / f"{_delivery_id(delivery_id)}.json") diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index 0b82e37ef642b..987aa66ebdd53 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -115,7 +115,7 @@ def _admit_prompt_turn( return images, agent -def _record_turn_marker(session: dict, text: Any) -> str: +def _record_turn_marker(session: dict, text: Any, *, auto_continue: bool = True) -> str: """Write the durable crash marker; returns the session key it was written under (compression can rotate session_key mid-turn). A surviving marker means the process died mid-turn. The key is published before the disk write so an interrupt racing startup can retire @@ -127,7 +127,8 @@ def _record_turn_marker(session: dict, text: Any) -> str: if isinstance(marker_text, str) and marker_text.strip(): with session["history_lock"]: session["_active_turn_marker_key"] = marker_key - record_turn_start(marker_home, marker_key, marker_text, attempts=marker_attempt) + record_turn_start(marker_home, marker_key, marker_text, attempts=marker_attempt, + auto_continue=auto_continue) with session["history_lock"]: marker_cancelled = bool(session.get("_turn_cancel_requested")) if marker_cancelled: @@ -779,11 +780,16 @@ def run(): st = _TurnRun( session["agent"], session.pop("one_turn_model_restore", None), terminal_callback, receipt_committed=terminal_callback is None) - st.marker_key = _record_turn_marker(session, text) + st.marker_key = _record_turn_marker(session, text, auto_continue=terminal_callback is None) goal_followup = None try: prepared = _prepare_turn_input(sid, session, st, text, images) if prepared is None: + if st.terminal_callback is not None and not st.receipt_attempted: + st.receipt_attempted = True + st.terminal_callback({ + "status": "failed", "text": "", "error": "Context injection refused."}) + st.receipt_committed = True return prompt, run_message, cols, streamer = prepared _invoke_agent( diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py index 291d494cd0c73..f31bdf0febecb 100644 --- a/tui_gateway/session_auto_continue.py +++ b/tui_gateway/session_auto_continue.py @@ -65,6 +65,8 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> home = _session_home(session) if (marker := read_turn_marker(home, session_key)) is None: return None + if not marker.get("auto_continue", True): + return None # The mailbox owns recovery and receipt identity for imported turns. enabled, freshness_secs, max_attempts = _auto_continue_config() age = time.time() - marker["started_at"] if not enabled or age > freshness_secs or marker["attempts"] >= max_attempts: diff --git a/tui_gateway/session_lifecycle.py b/tui_gateway/session_lifecycle.py index 6800e1cd1a2a2..90bbf31783556 100644 --- a/tui_gateway/session_lifecycle.py +++ b/tui_gateway/session_lifecycle.py @@ -31,7 +31,7 @@ def _claim_active_session_slot( from hermes_cli.active_sessions import try_acquire_active_session return try_acquire_active_session( session_id=session_key, surface=surface, config=_load_cfg(), registry_home=profile_home, - metadata={"live_session_id": live_session_id}, + metadata={"live_session_id": live_session_id, "bot_live_delivery_consumer": True}, track_liveness=str(surface or "").strip().lower() == "desktop") except Exception as exc: logger.warning("Failed to claim active session slot: %s", exc) @@ -138,7 +138,8 @@ def _transfer_active_session_slot(sid: str, session: dict, *, new_session_id: st return True try: from hermes_cli.active_sessions import transfer_active_session - if transfer_active_session(lease, session_id=new_session_id, metadata={"live_session_id": sid}): + if transfer_active_session(lease, session_id=new_session_id, metadata={ + "live_session_id": sid, "bot_live_delivery_consumer": True}): return True except Exception: logger.debug("Failed to transfer active session slot", exc_info=True) diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 246880691c30f..8f3bc45e52f6b 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -489,6 +489,57 @@ def _notif_handle_ready(sid, session, events, emitted, registry, fmt, deferred, _notif_dispatch_completions(sid, session, completions, registry, deferred) +def _poll_bot_live_delivery_once(sid: str, session: dict) -> bool: + """Run one durable envelope only after local FIFO/continuations yield the idle boundary.""" + from tools.bot_live_delivery import claim_pending_delivery, complete_delivery, find_canonical_live_owner + + home = _session_home(session) + with session["history_lock"]: + if any(session.get(key) for key in ( + "running", "_closing", "_finalized", "queued_prompt", "queued_prompts", + "_auto_continue_scheduled")) or session.get("agent") is None: + return False + lease = session.get("active_session_lease") + if lease is None or getattr(lease, "released", False): + return False + owner = find_canonical_live_owner(home) + if (not owner or owner.get("lease_id") != lease.lease_id + or owner.get("live_session_id") != sid + or owner.get("session_id") != session.get("session_key")): + return False + # The mailbox matches each envelope to this pinned lease/live id and compression lineage. + claimed = claim_pending_delivery(home, owner) + if claimed is None: + return False + session["running"] = True + + delivery_id = str(claimed["id"]) + + def terminal_receipt(terminal: dict) -> None: + status = str(terminal.get("status") or "failed") + error = str(terminal.get("error") or "") + reason = "cancelled" if status == "cancelled" else "" + if status not in {"settled", "cancelled"}: + from tools.bot_failure_reasons import classify_agent_error + reason = classify_agent_error(error) + # Let a failed write propagate: the turn must not retire its crash marker without its receipt. + complete_delivery(home, delivery_id, status=status, + reply=str(terminal.get("text") or "") if status == "settled" else "", + error=error, reason=reason) + + try: + started = _run_prompt_submit(f"__bot_dm__{delivery_id}", sid, session, claimed["message"], + image_paths=[], terminal_callback=terminal_receipt) + except Exception as exc: + _notif_release_turn(session) + terminal_receipt({"status": "failed", "error": str(exc)}) + raise + if not started: + _notif_release_turn(session) + terminal_receipt({"status": "failed", "error": "live session owner could not start the delivery turn"}) + return started + + def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None: """Daemon thread (started by _init_session()) that drains the process-global completion_queue for this session (ownership routing: _notif_handle_event) and polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` — the @@ -507,6 +558,10 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di last_kanban_poll = last_loop_poll = 0.0 while not stop_event.is_set() and not session.get("_finalized"): now = time.monotonic() + try: + _poll_bot_live_delivery_once(sid, session) + except Exception: + logger.warning("Bot live-owner delivery poll failed", exc_info=True) # /loop and /heartbeat wakeup drivers: fire a due tick for THIS session while idle (same claim-under-lock # as kanban dispatch). An active non-parked /goal owns the idle boundary and defers the loop tick. if now - last_loop_poll >= _LOOP_POLL_SECONDS: diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py index a5cc237412c9b..e17891feed085 100644 --- a/tui_gateway/turn_marker.py +++ b/tui_gateway/turn_marker.py @@ -83,13 +83,15 @@ def _update(home: Path | str, session_key: str, mutate, what: str) -> None: logger.debug("failed to %s turn marker for %s", what, session_key, exc_info=True) -def record_turn_start(home: Path | str, session_key: str, prompt: str, *, attempts: int = 0) -> None: +def record_turn_start(home: Path | str, session_key: str, prompt: str, *, attempts: int = 0, + auto_continue: bool = True) -> None: """Persist the marker for a turn that is about to run. ``attempts`` = how many auto-continues led to this run (0 for a user-initiated turn); the crash-loop breaker reads it back on the next resume.""" if not session_key or not prompt: return now = time.time() - entry = {"attempts": max(0, int(attempts)), "prompt": prompt[:_MAX_PROMPT_CHARS], "started_at": now} + entry = {"attempts": max(0, int(attempts)), "prompt": prompt[:_MAX_PROMPT_CHARS], "started_at": now, + "auto_continue": bool(auto_continue)} _update(home, session_key, lambda entries: {**_prune(entries, now), session_key: entry}, "record") @@ -109,6 +111,7 @@ def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | Non prompt = str(entry.get("prompt") or "") if isinstance(entry, dict) else "" if not prompt.strip(): return None - return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry)} + return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry), + "auto_continue": bool(entry.get("auto_continue", True))} except Exception: return None From e8125a532d4b4f1631307fefbe8d0622726c3b38 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:41:12 -0700 Subject: [PATCH 2/2] fix: cron and local DMs reach an open Desktop Bot Chat Route local producers to durable owner ingress before attempting the unowned CLI lane. Preserve per-run/per-message IDs and receipt-first retry handling; never fall back after ambiguous admission. Report cron admission as queued, not completed or failed, in job status, the execution ledger and CLI/tool UX. Native isolated Electron validation reproduces SESSION_NOT_OWNED on main for both idle and busy owners. Fixed owner consumes idle cron, busy cron, local DM and mounted-chat cron exactly once, keeps its lease, yields to queued human input, and preserves the prior model-request prefix and tool schema. Inference alone used a deterministic loopback wire stub; no paid model call. --- agent/monitoring/cron_health.py | 2 +- cron/executions.py | 5 +- cron/scheduler.py | 13 +- cron/scheduler_delivery.py | 109 ++++++++++++--- hermes_cli/cron.py | 6 +- tests/cron/test_cron_live_bot_delivery.py | 70 ++++++++++ tests/tools/test_bot_mode_dm.py | 53 +++++++- tools/bot_mode_dm.py | 127 ++++++++++++++++-- tools/cronjob_tools.py | 4 +- .../docs/developer-guide/cron-internals.md | 2 +- website/docs/user-guide/bot-mode.md | 2 + website/docs/user-guide/features/cron.md | 3 + 12 files changed, 358 insertions(+), 38 deletions(-) create mode 100644 tests/cron/test_cron_live_bot_delivery.py diff --git a/agent/monitoring/cron_health.py b/agent/monitoring/cron_health.py index 09d0c0a05d732..d0cb28fc0b900 100644 --- a/agent/monitoring/cron_health.py +++ b/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"} diff --git a/cron/executions.py b/cron/executions.py index 755e398415ee0..e10a36c733cf6 100644 --- a/cron/executions.py +++ b/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 diff --git a/cron/scheduler.py b/cron/scheduler.py index 5fc61d0cc2cda..70c1a169eea2a 100644 --- a/cron/scheduler.py +++ b/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 diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index a810000f50db6..d9678ff9a3550 100644 --- a/cron/scheduler_delivery.py +++ b/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 ] 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( diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index 58ea8a858aa86..a016272e00749 100644 --- a/hermes_cli/cron.py +++ b/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}") diff --git a/tests/cron/test_cron_live_bot_delivery.py b/tests/cron/test_cron_live_bot_delivery.py new file mode 100644 index 0000000000000..70fbd6ccc78d4 --- /dev/null +++ b/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 diff --git a/tests/tools/test_bot_mode_dm.py b/tests/tools/test_bot_mode_dm.py index 7a5de3b96dd1b..50629a585ad0a 100644 --- a/tests/tools/test_bot_mode_dm.py +++ b/tests/tools/test_bot_mode_dm.py @@ -214,7 +214,10 @@ def fake_terminal_tool(command, **kwargs): def _runner_parts(command): parts = shlex.split(command) marker = parts.index("--run-delivery") - return parts[marker + 1], parts[marker + 2], parts[marker + 3 :] + argv = parts[marker + 3 :] + if argv[:1] == ["--profile-home"]: + argv = argv[2:] + return parts[marker + 1], parts[marker + 2], argv def test_local_delivery_command_and_ack(tmp_path, monkeypatch): @@ -355,6 +358,54 @@ def boom(command, **kwargs): assert "could not be started" in result["error"] +def test_live_dm_admitted_before_waiter_failure(tmp_path, monkeypatch): + from tools import bot_live_delivery as live + + home = _managed_home(tmp_path) + target = home / "profiles" / "researcher" + owner = dict(profile_home=str(target), session_id="bot", lease_id="lease", live_session_id="live") + monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: owner if Path(h) == target else None) + monkeypatch.setattr(bot_mode_dm, "_dm_dir", lambda: tmp_path) + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "wrong-home")) + import tools.terminal_tool as terminal + monkeypatch.setattr(terminal, "terminal_tool", lambda *a, **k: json.dumps({"error": "spawn failed"})) + + result = json.loads(bot_mode_dm.message_agent_tool("researcher", "hello", agent=_FakeAgent(home))) + assert result["status"] == "queued" + record = live.read_delivery_result(target, result["delivery_id"]) + assert record is not None + assert record["owner"] == owner + assert record["message"] == "Message from 🤖 hermes (@hermes): hello" + assert "notification_error" in result + + +def test_live_dm_runner_retry_never_reexecutes_failed_claim(tmp_path, monkeypatch, capsys): + from tools import bot_live_delivery as live + + home = _managed_home(tmp_path) + target = home / "profiles" / "researcher" + monkeypatch.setenv("HERMES_HOME", str(home)) + owner = dict(profile_home=str(target), session_id="bot", lease_id="lease", live_session_id="live") + monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: owner) + monkeypatch.setattr(bot_mode_dm, "_LIVE_WAIT_SECONDS", 0) + monkeypatch.setattr(subprocess, "run", lambda *a, **k: pytest.fail("must not launch a model turn")) + dm_file = tmp_path / "message.txt" + dm_file.write_text("hello", encoding="utf-8") + argv = ["hermes", "-p", "researcher"] + assert bot_mode_dm._run_delivery(argv, str(dm_file), stdin_file=False) == 0 + queued = json.loads(capsys.readouterr().out) + assert queued["status"] == "queued" + claimed = live.claim_pending_delivery(target, owner) + assert claimed is not None + live.complete_delivery(target, claimed["delivery_id"], status="failed", error="HTTP 429 rate limit") + monkeypatch.setattr(live, "find_canonical_live_owner", lambda h: None) + assert bot_mode_dm._run_delivery(argv, str(dm_file), stdin_file=False) == 1 + failed = json.loads(capsys.readouterr().out) + assert failed["status"] == "failed" + assert failed["delivery_id"] == queued["delivery_id"] + assert dm_file.read_text(encoding="utf-8") == "hello" + + # ── plaintext tempfile lifecycle ───────────────────────────────────────────── diff --git a/tools/bot_mode_dm.py b/tools/bot_mode_dm.py index c3543c2fc8277..db3587acf5935 100644 --- a/tools/bot_mode_dm.py +++ b/tools/bot_mode_dm.py @@ -16,6 +16,7 @@ from __future__ import annotations import contextlib +import hashlib import json import logging import os @@ -43,6 +44,7 @@ # the machine dies between spawn ack and the runner's finally. _DM_DIR_NAME = "hermes-dm" _DM_STALE_SECONDS = 24 * 60 * 60 +_LIVE_WAIT_SECONDS = 300 # '/' — peer names are lowercase (``hermes peer`` normalizes them). _PEER_TARGET_RE = re.compile(r"^([a-z0-9][a-z0-9_-]{0,63})/([a-zA-Z0-9][a-zA-Z0-9_-]{0,63})$") @@ -192,7 +194,8 @@ def message_agent_tool(target: str = "", message: str = "", task_id: Optional[st return _err(f"Bot Mode gate check failed: {exc}") root, me = _hermes_root(Path(home)), _self_profile_name(Path(home)) - roster = [name for name, _dir in _roster(root)] + roster_homes = dict(_roster(root)) + roster = list(roster_homes) peers = _peers(root) teammates = [_handle(n) for n in roster if n != me] @@ -243,7 +246,7 @@ def _roster_err(msg: str) -> str: "machine, or on a registered peer. Pick a name from the roster " "(roles are listed in your system prompt).") return _start_delivery(["hermes", "-p", resolved, *BOT_CHAT_TURN_ARGS], content, f"@{_handle(resolved)}", - stdin_file=False, **delivery) + stdin_file=False, profile_home=roster_homes[resolved], **delivery) def _try_relay_delivery(root: Path, raw_target: str, content: str, me: str, *, @@ -396,9 +399,73 @@ def _turn(): return proc.returncode -def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: - """Run one DM transport and remove its plaintext file after consumption. The turn - window (not the enqueue) holds the target profile's cross-process lock, so two +def _admit_live_dm(profile_home: Path | None, dm_file: str) -> dict | None: + """Pin intent before admission; retries may inspect, never change transport.""" + from tools.bot_live_delivery import ( + _fsync_dir, deliver_to_live_owner, find_canonical_live_owner, read_delivery_result, + ) + + intent: dict[str, Any] + intent_path = Path(dm_file + ".live.json") + if intent_path.exists(): + intent = json.loads(intent_path.read_text(encoding="utf-8")) + else: + assert profile_home is not None + owner = find_canonical_live_owner(profile_home) + if owner is None: + return None + intent = dict(owner=owner, message=Path(dm_file).read_text(encoding="utf-8"), + delivery_id=hashlib.sha256(str(Path(dm_file).resolve()).encode()).hexdigest()) + try: + fd = os.open(intent_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + intent = json.loads(intent_path.read_text(encoding="utf-8")) + else: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(intent, stream) + stream.flush() + os.fsync(stream.fileno()) + _fsync_dir(intent_path.parent) + home = intent["owner"]["profile_home"] + record = read_delivery_result(home, intent["delivery_id"]) + if record is None: + record = deliver_to_live_owner(home, intent["owner"], intent["message"], + delivery_id=intent["delivery_id"]) + return record + + +def _wait_live_dm(home: str, delivery_id: str) -> int: + from tools.bot_live_delivery import read_delivery_result + + deadline = time.monotonic() + _LIVE_WAIT_SECONDS + while True: + record = read_delivery_result(home, delivery_id) + status = record["status"] if record else "ambiguous" + if status not in ("queued", "claimed") or time.monotonic() >= deadline: + break + time.sleep(min(0.5, max(0, deadline - time.monotonic()))) + payload = {key: record[key] for key in ("reply", "error", "reason") if record and record.get(key)} + payload.update(status=status, delivery_id=delivery_id) + if status in ("queued", "claimed", "ambiguous"): + payload["detail"] = "Delivery remains pending or its outcome is unknown. Do not resend; receipt is retained." + print(json.dumps(payload)) + return 0 if status in ("settled", "queued", "claimed") else 1 + + +def _local_delivery_home(argv: list[str]) -> Path | None: + cli = (argv[0] if argv else "").rsplit("\\", 1)[-1].rsplit("/", 1)[-1] + if len(argv) < 3 or cli not in ("hermes", "hermes.exe") or argv[1] != "-p": + return None + from tools.bot_mode_probe import _hermes_root, _roster + + return dict(_roster(_hermes_root(Path(_default_home())))).get(argv[2]) + + +def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool, + profile_home: Path | None = None) -> int: + """Route to the live owner before attempting a CLI transport. Live deliveries + retain their intent/payload and immutable receipt; only CLI/peer payloads are + removed after consumption. The CLI turn window holds the profile lock, so two deliveries into one profile queue; a bounded wait ends in a 'target_busy' refusal. Local (query-file) turns get one policy-gated retry (#93091 item 5): transient failures re-run the same @@ -407,6 +474,20 @@ def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: ever minted. Auth/quota/config failures never retry. Peer transports (stdin mode) retry on their own gateway's deliver path, not here. """ + # The live consumer owns turn admission; never compete for its CLI lease. + if not stdin_file: + home = profile_home or _local_delivery_home(argv) + if home is not None or Path(dm_file + ".live.json").exists(): + try: + record = _admit_live_dm(home, dm_file) + except Exception as exc: + print(json.dumps({"status": "ambiguous", "delivery_id": hashlib.sha256( + str(Path(dm_file).resolve()).encode()).hexdigest(), + "error": f"Live admission outcome unknown: {exc}. Do not resend.", + "evidence_file": dm_file})) + return 1 + if record is not None: + return _wait_live_dm(record["profile_home"], record["delivery_id"]) try: with _delivery_lock(argv, stdin_file=stdin_file): if not stdin_file: @@ -419,10 +500,14 @@ def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: _unlink_dm_file(dm_file) -def _delivery_command(argv: list[str], dm_file: str, *, stdin_file: bool) -> str: +def _delivery_command(argv: list[str], dm_file: str, *, stdin_file: bool, + profile_home: Path | None = None) -> str: """Build an argv-safe command for the cleanup-owning background runner.""" runner_argv = [sys.executable, str(Path(__file__).resolve()), "--run-delivery", - "stdin" if stdin_file else "query-file", dm_file, *argv] + "stdin" if stdin_file else "query-file", dm_file] + if profile_home is not None: + runner_argv.extend(["--profile-home", str(Path(profile_home).resolve())]) + runner_argv.extend(argv) if sys.platform == "win32": # The tracked local backend uses Git Bash on native Windows: forward slashes keep drive # paths executable there; backslash paths are parsed as command names (exit 127). @@ -431,11 +516,29 @@ def _delivery_command(argv: list[str], dm_file: str, *, stdin_file: bool) -> str def _start_delivery(argv: list[str], content: str, label: str, *, stdin_file: bool, - task_id: Optional[str], agent: Any) -> str: + task_id: Optional[str], agent: Any, profile_home: Path | None = None) -> str: """Create a DM file and transfer its cleanup ownership to the runner.""" dm_file = _write_dm_file(content) + if profile_home is not None: + try: + record = _admit_live_dm(profile_home, dm_file) + except Exception as exc: + return json.dumps({"status": "ambiguous", "delivery_id": hashlib.sha256( + str(Path(dm_file).resolve()).encode()).hexdigest(), + "error": f"Live delivery admission could not be confirmed: {exc}. Do not resend.", + "evidence_file": dm_file}) + if record is not None: + command = _delivery_command(argv, dm_file, stdin_file=False, profile_home=profile_home) + notification = json.loads(_spawn_delivery(command, label, task_id=task_id, agent=agent)) + result = dict(status=record["status"], delivery_id=record["delivery_id"], to=label, + detail="Durably queued for the live Bot Chat owner. Do NOT wait or resend; finish your turn.") + if notification.get("error"): + result["notification_error"] = notification["error"] + elif notification.get("process_id"): + result["process_id"] = notification["process_id"] + return json.dumps(result) try: - command = _delivery_command(argv, dm_file, stdin_file=stdin_file) + command = _delivery_command(argv, dm_file, stdin_file=stdin_file, profile_home=profile_home) except BaseException: _unlink_dm_file(dm_file) raise @@ -485,7 +588,10 @@ def _delivery_main(args: list[str]) -> int: if len(args) < 3 or args[0] != "--run-delivery" or args[1] not in ("stdin", "query-file"): return 2 try: - return _run_delivery(args[3:], args[2], stdin_file=args[1] == "stdin") + argv, profile_home = args[3:], None + if len(argv) >= 2 and argv[0] == "--profile-home": + profile_home, argv = Path(argv[1]), argv[2:] + return _run_delivery(argv, args[2], stdin_file=args[1] == "stdin", profile_home=profile_home) except Exception as exc: # 'target_busy': the queued delivery gave up after its bounded wait — surface the # structured payload on stdout so the completion notification carries it back. @@ -521,4 +627,5 @@ def _session_title(agent: Any) -> str: if __name__ == "__main__": # pragma: no cover - exercised as a background process + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) raise SystemExit(_delivery_main(sys.argv[1:])) diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 79ad09e570d71..fb0de38258d92 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -168,6 +168,8 @@ def _manual_run_delivery_note(deliver: str, refreshed: Dict[str, Any]) -> str: return " (output saved locally only)" err = str(refreshed.get("last_delivery_error") or "").strip() if not err: + if refreshed.get("last_delivery_queued"): + return " (output queued for Bot Chat; completion unverified, do not resend)" return " (output was delivered there by the job itself)" return f" (⚠ delivery FAILED: {err[:200]})" @@ -317,7 +319,7 @@ def _run_claimed_job(job: Dict[str, Any], extra_prompt: Optional[str] = None) -> # That is NOT a success for the caller — the calling agent relays this result — so report it as # failed and surface the delivery error, which lives in last_delivery_error (last_error is None for # these runs, and a bare success=False with error=None reads as an unexplained failure). See #83993. - ok = last_status == "ok" + ok = last_status in {"ok", "delivery_queued"} if execution is not None and execution.get("status") != "completed": ok = False run_error = execution.get("error") or f"execution ended in {execution.get('status') or 'unknown'} state" diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md index d68f676ed92c0..29b2b42e6c92c 100644 --- a/website/docs/developer-guide/cron-internals.md +++ b/website/docs/developer-guide/cron-internals.md @@ -286,7 +286,7 @@ Platforms in the first group have explicit, validated target syntax — named ch For **Telegram topics**, use `telegram::` (e.g., `telegram:-1001234567890:17585`). For **Slack threads**, the third segment is the parent message's `thread_ts` (e.g., `slack:C0123ABCD45:1700000000.000100`), so it only applies when replying under an existing message. -**Bot Chat** (`bot-chat`, `bot-chat:`) is a machine-local pseudo-platform, not a gateway adapter: the scheduler delivers by running `hermes [-p ] chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file ` — the same lane Bot Mode agent-to-agent messages use — so the output arrives as a real inbound turn in the profile's canonical Bot Chat and the bot runs a full agent turn on it (alternation-safe by construction; this is the chat command lane, not a transcript mirror). The bare token targets the job's own profile; the named form is validated against `~/.hermes/profiles/` at create time and again at fire time, and never resolves across machines. Bot-chat targets are excluded from the `all` routing token and from delivery preflight (no gateway credentials involved). The per-delivery subprocess timeout is `cron.bot_chat_delivery_timeout_seconds` (default 600). +**Bot Chat** (`bot-chat`, `bot-chat:`) is a machine-local pseudo-platform, not a gateway adapter. A mailbox-capable canonical live owner receives durable admission immediately (idle or busy); only that owner executes the incoming turn. `scheduler_delivery._deliver_to_bot_chat` resolves the target with `get_profile_dir` or the job's current `get_hermes_home`, derives the receipt ID from the source home, job ID, durable `execution_id`, and target home, and checks the receipt before discovering an owner. An existing receipt never permits CLI fallback. Without a mailbox owner it retains `hermes [-p ] chat --in ~ -c "Bot Chat" --create-if-missing -Q --query-file ` and normal ownership fencing. Both lanes deliver a real inbound turn, not a transcript mirror. Queued/claimed receipts populate `last_delivery_queued` with receipt IDs. The delivery aggregator excludes admission notices from genuine errors and records execution `delivery_outcome=queued`; successful jobs use `last_status=delivery_queued`. Genuine errors on mixed targets take precedence as failed while retaining queued receipt metadata. The target profile’s durable receipt is authoritative for terminal completion. Queued is the historical admission outcome, not proof of delivery. Historical cron status does not automatically track later receipt completion. Bot-chat targets are excluded from `all` and credential preflight. Bot-chat-only external workers bypass the gateway delivery queue; mixed external-worker targets retain gateway handoff. `cron.bot_chat_delivery_timeout_seconds` (default 600) bounds only the legacy subprocess lane. ### Response Wrapping diff --git a/website/docs/user-guide/bot-mode.md b/website/docs/user-guide/bot-mode.md index f3d9ffab78976..519818c9897ea 100644 --- a/website/docs/user-guide/bot-mode.md +++ b/website/docs/user-guide/bot-mode.md @@ -109,6 +109,8 @@ Bots message each other with attribution, and you can hand work off from any cha - **Renamed Bots keep their tags in sync** — give a Bot a friendly name (the pencil in its chat header, or `hermes profile rename`) and it becomes taggable by that name: a Bot titled *Research Buddy* answers to `@research-buddy` (and `@researchbuddy`), in regular chats and in group rooms alike. The composer's `@` autocomplete offers the renamed tag and also matches when you type the old profile name, which keeps resolving too. - **Direct messages** — every Bot Chat carries the `message_agent` tool: a Bot messages a teammate by calling `message_agent(target="researcher", message="…")`. The tool validates the target against the live roster, prefixes the sender's `Message from 🤖 (@):` attribution automatically, and delivers into the teammate's canonical Bot Chat. Delivery is **fire-and-forget**: the sender gets an acknowledgement, finishes its turn, and the reply arrives later as a background completion notification. The message travels as a real parameter (nothing shell-interpreted — quotes, `$(...)`, and backticks arrive verbatim), and the Bot composes its own message rather than forwarding your words. The teammate roster — names **and roles** from each profile's title/description — is part of every Bot Chat's system prompt, so Bots know who does what before choosing a recipient. The tool exists **only** in canonical Bot Chat sessions on Bot-Mode-managed installs; regular chats, group-room member sessions, and CLI sessions never see it. +Local messages also reach a Bot Chat that stays open in Desktop or the TUI. The receiving backend keeps ownership: it reads durable ingress on its existing notification poller, admits immediately when idle, or waits until the running turn and already queued human prompts finish. A `queued` acknowledgement confirms durable admission, **not** a completed reply. The target profile retains the delivery ID and receipt under `runtime/bot_live_delivery/`; `settled` confirms completion. A crashed or cancelled imported turn is not automatically replayed, and pending work pinned to a departed owner remains inspectable rather than being silently rerun. Do not resend a delivery whose outcome is unknown. Older backends without live-delivery capability retain the existing ownership refusal; restart that backend after upgrading. + The backend teaches each Bot's canonical Bot Chat session the messaging protocol automatically at prompt-build time — including when a teammate opens it headlessly from the CLI. Only the canonical Bot Chat gets the protocol section; your regular sessions and your SOUL.md stay untouched. This is controlled by `agent.bot_mode_protocol` in `config.yaml` (default: on): ```yaml diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index 0db4cf022b9ea..36a3aadb1a10d 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -496,6 +496,9 @@ error. A delivery failure does not count toward the job's `failure_streak` - `bot-chat:` targets another profile **on the same machine**. Names are validated against `hermes profile list` when the job is created; profiles on other gateways or machines can never be targeted, so same-named profiles across machines are unambiguous. - Each delivery costs the target bot one full agent turn — mind the schedule frequency. - Composes with other targets (`bot-chat,telegram`) but is never included in `all`. +- If the canonical chat is open in a mailbox-capable Desktop/TUI backend, delivery is **durably queued immediately**, whether the bot is idle or busy. Only that live owner runs the incoming turn; cron does not start a competing CLI writer. Without a live mailbox owner, the existing `hermes chat -c "Bot Chat" --create-if-missing` lane remains available (normal session ownership checks still apply). +- **Queued is not completed.** Cron records receipt IDs and `queued`/`claimed` statuses in `last_delivery_queued`, with delivery outcome `queued` (neither delivered nor failed). A successful job shows `delivery_queued`; genuine errors on other targets still take precedence as delivery failures. The bot may complete later. The durable receipt in the target profile's `runtime/bot_live_delivery/.json` is authoritative; cron's historical status is not automatically refreshed. +- Rechecking the same execution inspects its existing receipt, even if the owner has disappeared. It never falls back to another writer after acceptance. `failed`, `cancelled`, or `ambiguous` receipts are not automatically replayed; inspect the chat and receipt before intentionally starting new work. Each new cron execution has a distinct delivery ID. ### Routing intent (`all`)