diff --git a/tests/tui_gateway/test_bot_live_owner_delivery.py b/tests/tui_gateway/test_bot_live_owner_delivery.py index f16d7c212b..cf4b0a7c58 100644 --- a/tests/tui_gateway/test_bot_live_owner_delivery.py +++ b/tests/tui_gateway/test_bot_live_owner_delivery.py @@ -80,6 +80,7 @@ def test_local_work_blocks_mailbox_claim_without_consuming_envelope(monkeypatch, "_run_prompt_submit": submit, "_notif_release_turn": lambda session: session.update(running=False), }) + mailbox._root(tmp_path).mkdir(parents=True) # a delivery was admitted for this profile 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"): @@ -93,3 +94,51 @@ def test_local_work_blocks_mailbox_claim_without_consuming_envelope(monkeypatch, assert submitted == [("imported", author)] and not pending assert receipts[0][0][1] == "receipt" assert receipts[0][1]["reply"] == "reply" + + +def test_mailbox_poll_skips_owner_lookup_without_a_mailbox(monkeypatch, tmp_path): + """No mailbox directory → no state.db open / registry lock per pass; the lookup runs once one exists.""" + import tools.bot_live_delivery as mailbox + lookups = [] + monkeypatch.setattr(mailbox, "find_canonical_live_owner", lambda home: lookups.append(home) or None) + poll = rebind(session_notifications._poll_bot_live_delivery_once, {"_session_home": lambda session: tmp_path}) + session = {"history_lock": threading.RLock(), "agent": object(), "session_key": "chat", + "active_session_lease": SimpleNamespace(lease_id="lease", released=False)} + assert poll("live", session) is False + assert lookups == [] + mailbox._root(tmp_path).mkdir(parents=True) + assert poll("live", session) is False + assert lookups == [tmp_path] + + +def test_failing_mailbox_poll_backs_off_and_warns_once_per_window(): + """A failing poll is retried only after the backoff and logged at WARNING once per window.""" + import logging + records = [] + + class _Handler(logging.Handler): + def emit(self, record): + records.append(record) + + log = logging.getLogger("test.bot_poll_throttle") + log.addHandler(_Handler()) + log.setLevel(logging.DEBUG) + attempts = [] + + def failing(sid, session): + attempts.append(sid) + raise RuntimeError("active session file lock unavailable") + + guarded = rebind(session_notifications._poll_bot_live_delivery_guarded, { + "_poll_bot_live_delivery_once": failing, "logger": log, + "_BOT_POLL_FAILURE_BACKOFF_S": session_notifications._BOT_POLL_FAILURE_BACKOFF_S, + "_BOT_POLL_WARN_INTERVAL_S": session_notifications._BOT_POLL_WARN_INTERVAL_S}) + session = {} + for now in (0.0, 0.5, 1.0, 6.0, 12.0, 61.0): + guarded("live", session, now) + assert len(attempts) == 4 # 0.0, 6.0, 12.0, 61.0 — the 0.5/1.0 passes sat out the backoff + warnings = [r for r in records if r.levelno == logging.WARNING] + assert [r.getMessage() for r in warnings] == [ + "Bot live-owner delivery poll failed (0 repeat(s) suppressed since the last report)", + "Bot live-owner delivery poll failed (2 repeat(s) suppressed since the last report)"] + assert all(r.exc_info for r in warnings) diff --git a/tools/bot_live_delivery.py b/tools/bot_live_delivery.py index ebec4efeec..1ede618844 100644 --- a/tools/bot_live_delivery.py +++ b/tools/bot_live_delivery.py @@ -76,6 +76,13 @@ def _root(home: Path | str) -> Path: return Path(home).resolve() / "runtime" / DELIVERY_DIR_NAME +def has_mailbox(profile_home: Path | str) -> bool: + """Whether any delivery was ever admitted for this profile (the mailbox directory is created on + first admission only). A cheap pre-check for pollers: no mailbox means nothing to claim, so the + owner lookup — a state.db open plus the exclusive active-session registry lock — can be skipped.""" + return _root(profile_home).is_dir() + + @contextmanager def _locked(home: Path | str): root = _root(home) diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 8b93074a96..3ba919d9fd 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -545,9 +545,13 @@ def _notif_handle_ready(sid, session, events, emitted, registry, fmt, 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 + from tools.bot_live_delivery import claim_pending_delivery, complete_delivery, find_canonical_live_owner, has_mailbox home = _session_home(session) + # Most profiles never receive a delivery: without a mailbox there is nothing to claim, and the owner + # lookup below costs a state.db open plus the exclusive active-session registry lock every pass (#111719). + if not has_mailbox(home): + return False with session["history_lock"]: if any(session.get(key) for key in ( "running", "_closing", "_finalized", "queued_prompt", "queued_prompts", @@ -595,6 +599,35 @@ def _poll_bot_live_delivery_once(sid: str, session: dict) -> bool: return started +# A failing mailbox poll (typically the active-session registry lock unavailable under contention) retries +# every ``queue.get`` slice; back off between attempts and log the failure once per window, not per attempt. +_BOT_POLL_FAILURE_BACKOFF_S = 5.0 +_BOT_POLL_WARN_INTERVAL_S = 60.0 + + +def _poll_bot_live_delivery_guarded(sid: str, session: dict, now: float) -> None: + """One poller-loop pass of the mailbox poll: skipped while backing off after a failure; a failure is + logged at WARNING once per ``_BOT_POLL_WARN_INTERVAL_S`` (with the count of suppressed repeats) and at + DEBUG otherwise. An unthrottled poll logged ``Bot live-owner delivery poll failed`` ~2×/minute per session + for days, 91% of an install's WARNING output (#111719).""" + if now < session.get("_bot_poll_retry_at", 0.0): + return + try: + _poll_bot_live_delivery_once(sid, session) + except Exception: + session["_bot_poll_retry_at"] = now + _BOT_POLL_FAILURE_BACKOFF_S + suppressed = int(session.get("_bot_poll_warn_suppressed", 0)) + if now - session.get("_bot_poll_warned_at", -_BOT_POLL_WARN_INTERVAL_S) < _BOT_POLL_WARN_INTERVAL_S: + session["_bot_poll_warn_suppressed"] = suppressed + 1 + logger.debug("Bot live-owner delivery poll failed (repeat)", exc_info=True) + return + session["_bot_poll_warned_at"], session["_bot_poll_warn_suppressed"] = now, 0 + logger.warning("Bot live-owner delivery poll failed (%d repeat(s) suppressed since the last report)", + suppressed, exc_info=True) + return + session["_bot_poll_warn_suppressed"] = 0 + + 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 @@ -613,10 +646,7 @@ 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) + _poll_bot_live_delivery_guarded(sid, session, now) # /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: