diff --git a/gateway/run_adapters.py b/gateway/run_adapters.py index 608f260da8..ec8db588c7 100644 --- a/gateway/run_adapters.py +++ b/gateway/run_adapters.py @@ -188,13 +188,23 @@ class GatewayAdapterLifecycleMixin: tasks = getattr(self, "_fatal_handler_tasks", None) if tasks is None: tasks = self._fatal_handler_tasks = set() - task = asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter)) - tasks.add(task) - task.add_done_callback(tasks.discard) + task = self._track_task_in(tasks, asyncio.create_task(self._handle_adapter_fatal_error_detached(adapter))) # shield(): a plain `await task` would tunnel the caller's cancellation into the detached # task; with shield the caller sees CancelledError and the handler runs to completion. await asyncio.shield(task) + def _reconnect_queue_entry( + self, platform, adapter, platform_config, *, attempts: int, delay: float, queued: bool = True + ) -> dict: + """Build a ``_failed_platforms`` entry (startup failures and runtime fatals share the shape).""" + now = time.monotonic() + return { + "config": platform_config, "attempts": attempts, "next_retry": now + delay, + **({"queued_at": now} if queued else {}), + "credential_claim": self._adapter_credential_claim(platform, adapter), + "listener_claim": self._adapter_listener_claim(platform, adapter), + } + def _queue_retryable_fatal_platform(self, adapter: BasePlatformAdapter) -> bool: """Queue a retryable fatal adapter for background reconnection (True when newly queued). @@ -211,14 +221,9 @@ class GatewayAdapterLifecycleMixin: # permanent outage — nothing retries and the stranded check treats "queued" as safe. self._ensure_reconnect_watcher_running() return False - self._failed_platforms[adapter.platform] = { - "config": platform_config, - "attempts": 0, - "next_retry": time.monotonic(), - "queued_at": time.monotonic(), - "credential_claim": self._adapter_credential_claim(adapter.platform, adapter), - "listener_claim": self._adapter_listener_claim(adapter.platform, adapter), - } + self._failed_platforms[adapter.platform] = self._reconnect_queue_entry( + adapter.platform, adapter, platform_config, attempts=0, delay=0.0, + ) logger.info("%s queued for background reconnection", adapter.platform.value) self._ensure_reconnect_watcher_running() return True @@ -341,6 +346,13 @@ class GatewayAdapterLifecycleMixin: task.add_done_callback(tasks.discard) return task + @staticmethod + def _track_task_in(tasks: set, task: "asyncio.Task") -> "asyncio.Task": + """Register ``task`` in an arbitrary lifecycle set with self-removal on completion.""" + tasks.add(task) + task.add_done_callback(tasks.discard) + return task + def _request_clean_exit(self, reason: str) -> None: self._exit_cleanly = True self._exit_reason = reason @@ -640,6 +652,13 @@ class GatewayAdapterLifecycleMixin: retrying_since=retrying_since_iso, ) + def _mark_platform_fatal(self, status_key: str, adapter) -> None: + """Record an adapter's fatal error code/message as ``fatal`` runtime status.""" + self._update_platform_runtime_status( + status_key, platform_state="fatal", error_code=adapter.fatal_error_code, + error_message=adapter.fatal_error_message, + ) + def _bump_reconnect_backoff( self, platform, info: dict, attempt: int, error_code, error_message: str ) -> int: @@ -687,10 +706,7 @@ class GatewayAdapterLifecycleMixin: if success: await self._install_reconnected_adapter(platform, adapter) elif adapter.has_fatal_error and not adapter.fatal_error_retryable: - self._update_platform_runtime_status( - platform.value, platform_state="fatal", error_code=adapter.fatal_error_code, - error_message=adapter.fatal_error_message, - ) + self._mark_platform_fatal(platform.value, adapter) logger.warning( "Reconnect %s: non-retryable error (%s), removing from retry queue", platform.value, adapter.fatal_error_message, @@ -779,8 +795,7 @@ class GatewayAdapterLifecycleMixin: _done, unfinished = await asyncio.wait(tasks, timeout=timeout) if unfinished: logger.warning( - "Timed out waiting for %d secondary profile reconnect task(s) during shutdown", - len(unfinished), + "Timed out waiting for %d secondary profile reconnect task(s) during shutdown", len(unfinished), ) pending.clear() @@ -880,18 +895,9 @@ class GatewayAdapterLifecycleMixin: # Register this profile's shell hooks / outbound webhooks: start() registers before any # profile scope exists, so a secondary profile's `hooks:` block would be silently inert. - try: - from hermes_cli.config import load_config as _load_profile_config - from agent.shell_hooks import register_from_config as _register_shell_hooks - from agent.outbound_webhooks import (register_from_config as _register_outbound_webhooks) - - _profile_hooks_cfg = _load_profile_config() - _register_shell_hooks(_profile_hooks_cfg, accept_hooks=False) - _register_outbound_webhooks(_profile_hooks_cfg) - except Exception: - logger.warning( - "shell-hook/webhook registration failed for profile '%s'", profile_name, exc_info=True, - ) + self._register_config_hooks( + "shell-hook/webhook registration failed for profile '%s'", profile_name, level=logging.WARNING, + ) profile_cfg = load_gateway_config() violation = _own_policy_open_startup_violation(profile_cfg) @@ -929,31 +935,27 @@ class GatewayAdapterLifecycleMixin: owner = claimed.get(claim) if claim is not None else None if owner is None: return False + pv = platform.value if kind == "credential": message = ( - f"Profile '{owner}' and '{profile_name}' both configure " - f"{platform.value} with the same credential. Give each " - f"profile its own {platform.value} credential." + f"Profile '{owner}' and '{profile_name}' both configure {pv} with the same credential. " + f"Give each profile its own {pv} credential." ) logger.error( - "Profile '%s' and '%s' both configure %s with the same " - "credential — refusing to start the duplicate (one " - "credential cannot be consumed twice). Give each profile " - "its own %s credential.", owner, profile_name, platform.value, platform.value, + "Profile '%s' and '%s' both configure %s with the same credential — refusing to start the " + "duplicate (one credential cannot be consumed twice). Give each profile its own %s credential.", + owner, profile_name, pv, pv, ) else: bind, port = claim[-2:] message = ( - f"Profile '{owner}' and '{profile_name}' both configure " - f"{platform.value} sidecars on the same listener. Configure " - f"a distinct listener for profile '{profile_name}'." + f"Profile '{owner}' and '{profile_name}' both configure {pv} sidecars on the same listener. " + f"Configure a distinct listener for profile '{profile_name}'." ) logger.error( - "Profile '%s' and '%s' both configure %s sidecars on " - "%s:%s — refusing to start the duplicate listener. " - "Set platforms.%s.extra.sidecar_port to a distinct port " - "for profile '%s'.", - owner, profile_name, platform.value, bind, port, platform.value, profile_name, + "Profile '%s' and '%s' both configure %s sidecars on %s:%s — refusing to start the duplicate " + "listener. Set platforms.%s.extra.sidecar_port to a distinct port for profile '%s'.", + owner, profile_name, pv, bind, port, pv, profile_name, ) self._update_platform_runtime_status( f"{profile_name}:{platform.value}", platform_state="fatal", @@ -1204,10 +1206,7 @@ class GatewayAdapterLifecycleMixin: "gateway (%s) — parked, not retried. %s", profile_name, platform.value, adapter.fatal_error_code, adapter.fatal_error_message or "", ) - self._update_platform_runtime_status( - f"{profile_name}:{platform.value}", platform_state="fatal", - error_code=adapter.fatal_error_code, error_message=adapter.fatal_error_message, - ) + self._mark_platform_fatal(f"{profile_name}:{platform.value}", adapter) return def _handoff() -> None: diff --git a/gateway/run_notifications.py b/gateway/run_notifications.py index ff81650461..3ed6b08ed9 100644 --- a/gateway/run_notifications.py +++ b/gateway/run_notifications.py @@ -14,14 +14,12 @@ import logging import time from contextlib import suppress from pathlib import Path -from typing import TYPE_CHECKING, Any, Dict, Optional, cast +from typing import Any, Dict, Optional, cast from gateway.config import Platform, _BUILTIN_PLATFORM_VALUES from gateway.platforms.base import MessageEvent, MessageType from gateway.session import SessionEntry, SessionSource - -if TYPE_CHECKING: # string annotations only; never imported at runtime (cycle) - from gateway.run import GatewayRunner, TurnRunner # noqa: F401 +from gateway.run_shutdown import _notice_target_key, _send_error, _send_failed # Log-record parity with the origin module. logger = logging.getLogger("gateway.run") @@ -78,6 +76,9 @@ class GatewayNotificationsMixin: from gateway.run import _non_conversational_metadata return _non_conversational_metadata(self.metadata, platform=self.platform) + async def send(self, text: str): + return await self.adapter.send(self.chat_id, text, metadata=self.send_metadata()) + @dataclasses.dataclass class _CompletionClaim: """Pre-flight outcome for one completion delivery.""" @@ -99,11 +100,10 @@ class GatewayNotificationsMixin: if config and getattr(source, "platform", None) == Platform.SLACK and _is_slack_ignored_channel(config, chat_id): logger.info("Skipping Slack platform notice for configured ignored channel %s", chat_id) return - - notice_delivery = "public" - if config and hasattr(config, "get_notice_delivery"): - notice_delivery = config.get_notice_delivery(source.platform) - + notice_delivery = ( + config.get_notice_delivery(source.platform) if config and hasattr(config, "get_notice_delivery") + else "public" + ) metadata = self._thread_metadata_for_source(source) if notice_delivery == "private" and getattr(source, "user_id", None): try: @@ -400,12 +400,9 @@ class GatewayNotificationsMixin: return None platform = Platform(platform_str) adapter = self.adapters.get(platform) - metadata = self._thread_metadata_for_target( - platform, chat_id, pending.get("thread_id"), chat_type=pending.get("chat_type"), - reply_to_message_id=pending.get("message_id"), adapter=adapter, - ) if not adapter: return None + metadata = self._pending_marker_metadata(platform, chat_id, pending, adapter) # Fallback session key if not stored (old pending files) return self._UpdateTarget( adapter, chat_id, session_key or f"{platform_str}:{chat_id}", metadata, platform, @@ -414,6 +411,13 @@ class GatewayNotificationsMixin: pass return None + def _pending_marker_metadata(self, platform, chat_id, data: dict, adapter): + """Thread metadata for a persisted update/restart marker (thread_id/chat_type/message_id keys).""" + return self._thread_metadata_for_target( + platform, chat_id, data.get("thread_id"), chat_type=data.get("chat_type"), + reply_to_message_id=data.get("message_id"), adapter=adapter, + ) + async def _watch_update_completion_only(self, paths: "_UpdatePaths", deadline: float, poll_interval: float) -> None: """Fallback when no adapter/chat can be resolved: wait for the exit code, then notify.""" logger.warning("Update watcher: cannot resolve adapter/chat_id, falling back to completion-only") @@ -448,9 +452,7 @@ class GatewayNotificationsMixin: max_chunk = 3500 for i in range(0, len(clean), max_chunk): try: - await target.adapter.send( - target.chat_id, f"```\n{clean[i:i + max_chunk]}\n```", metadata=target.send_metadata(), - ) + await target.send(f"```\n{clean[i:i + max_chunk]}\n```") except Exception as e: logger.debug("Update stream send failed: %s", e) @@ -470,11 +472,9 @@ class GatewayNotificationsMixin: if not sent_buttons: default_hint = f" (default: {default})" if default else "" _p = getattr(adapter, "typed_command_prefix", "/") - await adapter.send( - target.chat_id, + await target.send( f"⚕ **Update needs your input:**\n\n{prompt_text}{default_hint}\n\n" - f"Reply `{_p}approve` (yes) or `{_p}deny` (no), or type your answer directly.", - metadata=target.send_metadata(), + f"Reply `{_p}approve` (yes) or `{_p}deny` (no), or type your answer directly." ) # Keep the prompt marker on disk until the user answers so a watcher after a mid-prompt # gateway restart can recover by re-forwarding it. @@ -531,11 +531,9 @@ class GatewayNotificationsMixin: await _flush_buffer() try: exit_code = int(paths.exit_code.read_text(encoding="utf-8").strip() or "1") - await target.adapter.send( - target.chat_id, + await target.send( "✅ Hermes update finished." if exit_code == 0 - else "❌ Hermes update failed (exit code {}).".format(exit_code), - metadata=target.send_metadata(), + else "❌ Hermes update failed (exit code {}).".format(exit_code) ) logger.info("Update finished (exit=%s), notified %s", exit_code, session_key) except Exception as e: @@ -570,9 +568,7 @@ class GatewayNotificationsMixin: paths.exit_code.write_text("124", encoding="utf-8") await _flush_buffer() with suppress(Exception): - await target.adapter.send( - target.chat_id, "❌ Hermes update timed out after 30 minutes.", metadata=target.send_metadata(), - ) + await target.send("❌ Hermes update timed out after 30 minutes.") self._clear_update_markers(paths, session_key) async def _send_update_notification(self) -> bool: @@ -626,10 +622,7 @@ class GatewayNotificationsMixin: return _defer("Update notification deferred: %s adapter not connected yet", platform_str) if adapter and chat_id: - metadata = self._thread_metadata_for_target( - platform, chat_id, pending.get("thread_id"), chat_type=pending.get("chat_type"), - reply_to_message_id=pending.get("message_id"), adapter=adapter, - ) + metadata = self._pending_marker_metadata(platform, chat_id, pending, adapter) from tools.ansi_strip import strip_ansi output = strip_ansi(output).strip() if output: @@ -682,10 +675,7 @@ class GatewayNotificationsMixin: ) return None - metadata = self._thread_metadata_for_target( - platform, chat_id, thread_id, chat_type=data.get("chat_type"), - reply_to_message_id=data.get("message_id"), adapter=transport.adapter, - ) + metadata = self._pending_marker_metadata(platform, chat_id, data, transport.adapter) if data.get("delivered_via_upstream_relay") is True: metadata = dict(metadata or {}) for field in ("user_id", "scope_id"): @@ -697,10 +687,9 @@ class GatewayNotificationsMixin: ) # adapter.send() catches provider errors (e.g. "Chat not found") and returns # SendResult(success=False) rather than raising, so inspect the result before claiming success. - if result is not None and getattr(result, "success", True) is False: + if _send_failed(result): logger.warning( - "Restart notification to %s:%s was not delivered: %s", - platform_str, chat_id, getattr(result, "error", "send returned success=False"), + "Restart notification to %s:%s was not delivered: %s", platform_str, chat_id, _send_error(result), ) return None @@ -740,10 +729,8 @@ class GatewayNotificationsMixin: result = await transport.send(platform, str(home.chat_id), message, metadata=send_metadata) else: result = await transport.adapter.send(str(home.chat_id), message) - if result is not None and getattr(result, "success", True) is False: - logger.warning( - failure_fmt, platform.value, home.chat_id, getattr(result, "error", "send returned success=False"), - ) + if _send_failed(result): + logger.warning(failure_fmt, platform.value, home.chat_id, _send_error(result)) return False return True except Exception as exc: @@ -770,7 +757,7 @@ class GatewayNotificationsMixin: ) continue - target = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None) + target = _notice_target_key(platform.value, home.chat_id, home.thread_id) if target in skipped or target in delivered: continue @@ -921,22 +908,19 @@ class GatewayNotificationsMixin: """ from gateway.wake import deliver_wake, persist_delegation_delivery if evt.get("type") == "async_delegation": - try: - logger.info( - "Async delegation completion — persisting delivery row for api_server session %s (no wake turn)", - raw_sid, - ) - await persist_delegation_delivery(adapter, text=synth_text, session_id=raw_sid, evt=evt) - return True - except Exception as e: - logger.warning("Async delegation delivery persist failed for session %s: %s", raw_sid, e) - return False + info = "Async delegation completion — persisting delivery row for api_server session %s (no wake turn)" + fail = "Async delegation delivery persist failed for session %s: %s" + deliver = lambda: persist_delegation_delivery(adapter, text=synth_text, session_id=raw_sid, evt=evt) # noqa: E731 + else: + info = "Watch pattern notification — waking api_server session %s via self-post" + fail = "Watch notification self-post wake failed for session %s: %s" + deliver = lambda: deliver_wake(adapter, text=synth_text, session_id=raw_sid) # noqa: E731 try: - logger.info("Watch pattern notification — waking api_server session %s via self-post", raw_sid) - await deliver_wake(adapter, text=synth_text, session_id=raw_sid) + logger.info(info, raw_sid) + await deliver() return True except Exception as e: - logger.warning("Watch notification self-post wake failed for session %s: %s", raw_sid, e) + logger.warning(fail, raw_sid, e) return False def _resolve_injection_adapter(self, platform_name: str): @@ -972,10 +956,9 @@ class GatewayNotificationsMixin: # API-server sessions bind the RAW X-Hermes-Session-Id key (_bind_api_server_session), not a # structured ``agent:main:...`` key, so _build_process_event_source returned None above. raw_sid = str(evt.get("origin_session_id") or "").strip() - if not raw_sid: - _sk = str(evt.get("session_key") or "").strip() - if _sk and _parse_session_key(_sk) is None: - raw_sid = _sk + _sk = str(evt.get("session_key") or "").strip() + if not raw_sid and _sk and _parse_session_key(_sk) is None: + raw_sid = _sk if raw_sid: adapter = self.adapters.get(Platform.API_SERVER) if adapter is not None and not adapter_supports_push(adapter): @@ -1297,13 +1280,17 @@ class GatewayNotificationsMixin: finally: # Never strand watcher futures when formatting, delivery, or cancellation interrupts a batch: # False follows the existing watcher retry path; None remains the ordinary dedupe result. - for _text, _evt, future in entries: - if not future.done(): - future.set_result(delivered) + self._settle_batch_waiters(entries, delivered) # Do not remove a newer flush task that reused the same route key. if self._completion_notification_batch_tasks.get(key) is current_task: self._completion_notification_batch_tasks.pop(key, None) + @staticmethod + def _settle_batch_waiters(entries, result) -> None: + for _text, _evt, future in entries: + if not future.done(): + future.set_result(result) + async def _cancel_process_completion_batch_tasks(self) -> None: """Settle pending completion batches before adapter teardown.""" self._completion_notification_batches_stopping = True @@ -1320,9 +1307,7 @@ class GatewayNotificationsMixin: # Defensive cleanup for an orphaned queue with no live flush task. batches = getattr(self, "_completion_notification_batches", {}) for entries in batches.values(): - for _text, _evt, future in entries: - if not future.done(): - future.set_result(False) + self._settle_batch_waiters(entries, False) batches.clear() getattr(self, "_completion_notification_batch_tasks", {}).clear() getattr(self, "_completion_notification_batch_flush_tasks", set()).clear() @@ -1352,10 +1337,8 @@ class GatewayNotificationsMixin: task = asyncio.create_task(self._flush_process_completion_batch(key)) self._completion_notification_batch_tasks[key] = task # Keep the flush alive under the gateway's normal lifecycle accounting. - self._background_tasks.add(task) - self._completion_notification_batch_flush_tasks.add(task) - task.add_done_callback(self._background_tasks.discard) - task.add_done_callback(self._completion_notification_batch_flush_tasks.discard) + self._retain_background_task(task) + self._track_task_in(self._completion_notification_batch_flush_tasks, task) return await future def _enrich_async_delegation_routing(self, evt: dict) -> None: @@ -1642,11 +1625,9 @@ class GatewayNotificationsMixin: "via wait/log — skipping raw notification (#65379)", session_id, ) break - should_notify = ( - notify_mode in {"concise", "all", "result"} - or (notify_mode == "error" and session.exit_code not in {0, None}) - ) - if should_notify: + if notify_mode in {"concise", "all", "result"} or ( + notify_mode == "error" and session.exit_code not in {0, None} + ): message_text = self._format_process_final_message(session_id, session, notify_mode) await self._send_watcher_message(platform_name, chat_id, thread_id, message_text) break @@ -1655,9 +1636,9 @@ class GatewayNotificationsMixin: # New output — deliver a status update (only in "all" mode; agent_notify watchers # only care about completion). new_output = self._redacted_output_tail(session, 500) - message_text = ( - f"[Background process {session_id} is still running~ New output:\n{new_output}]" + await self._send_watcher_message( + platform_name, chat_id, thread_id, + f"[Background process {session_id} is still running~ New output:\n{new_output}]", ) - await self._send_watcher_message(platform_name, chat_id, thread_id, message_text) logger.debug("Process watcher ended: %s", session_id) diff --git a/gateway/run_shutdown.py b/gateway/run_shutdown.py index 7a012b9f6d..baff2d22fa 100644 --- a/gateway/run_shutdown.py +++ b/gateway/run_shutdown.py @@ -18,7 +18,7 @@ import threading import time from contextlib import suppress from pathlib import Path -from typing import TYPE_CHECKING, Any, Callable, Dict, Optional +from typing import Any, Callable, Dict, Optional from gateway.config import Platform from gateway.restart import ( @@ -27,9 +27,6 @@ from gateway.restart import ( from gateway.run_common import _UNSET from gateway.shutdown_watchdog import arm_shutdown_watchdog, resolve_shutdown_watchdog_delay -if TYPE_CHECKING: # string annotations only; never imported at runtime (cycle) - from gateway.run import GatewayRunner, TurnRunner # noqa: F401 - # Log-record parity with the origin module. logger = logging.getLogger("gateway.run") @@ -88,6 +85,16 @@ def _send_failed(result: Any) -> bool: return result is not None and getattr(result, "success", True) is False +def _send_error(result: Any) -> str: + """Error text of a failed ``send()`` result (adapters may omit it).""" + return getattr(result, "error", "send returned success=False") + + +def _notice_target_key(platform_value: str, chat_id, thread_id) -> tuple: + """Dedup key for one notice destination: thread/topic platforms share a chat but route apart.""" + return (platform_value, str(chat_id), str(thread_id) if thread_id else None) + + class GatewayShutdownMixin: """Stop/drain/restart, scale-to-zero and active-work accounting methods for GatewayRunner.""" @@ -321,17 +328,14 @@ class GatewayShutdownMixin: from cron.scheduler import get_running_job_ids return len(get_running_job_ids()) - def _dashboard_seen(): - from gateway.scale_to_zero import dashboard_client_last_seen - return dashboard_client_last_seen() - cron_count = _read_or_awake("cron work count", _cron_count, 1) api_count = _read_or_awake("api work count", lambda: self._api_server_hook("active_agent_work_count"), 1) # An attached dashboard/desktop/TUI client (heartbeat file mtime from the dashboard process) is # inbound activity too — folded into the inbound clock, not a conjunct, so disconnect gets the # same idle_timeout grace as a message and a lingering marker cannot pin the box. last_inbound = self._last_inbound_at - seen = _read_or_awake("dashboard heartbeat", _dashboard_seen, time.time()) + from gateway.scale_to_zero import dashboard_client_last_seen + seen = _read_or_awake("dashboard heartbeat", dashboard_client_last_seen, time.time()) if seen is not None and seen > last_inbound: last_inbound = seen return is_idle( @@ -524,10 +528,9 @@ class GatewayShutdownMixin: info["pause_reason"] = reason or "auto-paused after repeated failures" # Push next_retry far enough out that a stale code path missing "paused" still never fires. info["next_retry"] = float("inf") - with suppress(Exception): - self._update_platform_runtime_status( - platform.value, platform_state="paused", error_code=None, error_message=info["pause_reason"], - ) + self._update_platform_runtime_status( + platform.value, platform_state="paused", error_code=None, error_message=info["pause_reason"], + ) logger.warning( "%s paused after %d consecutive failures (%s) — " "fix the underlying issue then run `/platform resume %s` " @@ -544,10 +547,7 @@ class GatewayShutdownMixin: info.pop("pause_reason", None) info["attempts"] = 0 info["next_retry"] = time.monotonic() # retry on next watcher tick - with suppress(Exception): - self._update_platform_runtime_status( - platform.value, platform_state="retrying", error_code=None, error_message=None, - ) + self._update_platform_runtime_status(platform.value, platform_state="retrying") logger.info("%s resumed — retrying on next watcher tick", platform.value) return True @@ -687,7 +687,7 @@ class GatewayShutdownMixin: continue chat_id = str(target.get("chat_id")) thread_id = target.get("thread_id") - dedup_key = (job_id, platform.value, chat_id, str(thread_id) if thread_id else None) + dedup_key = (job_id, *_notice_target_key(platform.value, chat_id, thread_id)) if dedup_key in notified: continue try: @@ -695,8 +695,7 @@ class GatewayShutdownMixin: result = await adapter.send(chat_id, msg, metadata=metadata) if _send_failed(result): logger.debug( - "Cron interrupt notice to %s:%s failed: %s", platform.value, chat_id, - getattr(result, "error", "send returned success=False"), + "Cron interrupt notice to %s:%s failed: %s", platform.value, chat_id, _send_error(result), ) continue notified.add(dedup_key) @@ -735,8 +734,8 @@ class GatewayShutdownMixin: result = await adapter.send(chat_id, msg, **send_kwargs) if _send_failed(result): logger.debug( - "Failed to send shutdown notification to %s%s:%s: %s", - where, platform_str, chat_id, getattr(result, "error", "send returned success=False"), + "Failed to send shutdown notification to %s%s:%s: %s", where, platform_str, chat_id, + _send_error(result), ) return False logger.info("Sent shutdown notification to %s %s:%s", kind, platform_str, chat_id) @@ -762,9 +761,8 @@ class GatewayShutdownMixin: restart_key = None if restart_source is not None: with suppress(Exception): - restart_key = ( - restart_source.platform.value, str(restart_source.chat_id), - str(restart_source.thread_id) if restart_source.thread_id else None, + restart_key = _notice_target_key( + restart_source.platform.value, restart_source.chat_id, restart_source.thread_id ) notified: set[tuple[str, str, Optional[str]]] = set() for session_key in self._snapshot_running_agents(): @@ -772,9 +770,7 @@ class GatewayShutdownMixin: if target is None: continue source, platform_str, chat_id, thread_id = target - # Dedupe only identical targets: thread/topic platforms share a parent chat yet route to - # distinct destinations via metadata. - dedup_key = (platform_str, chat_id, str(thread_id) if thread_id else None) + dedup_key = _notice_target_key(platform_str, chat_id, thread_id) if dedup_key in notified: continue try: @@ -830,7 +826,7 @@ class GatewayShutdownMixin: platform.value, ) continue - dedup_key = (platform.value, str(home.chat_id), str(home.thread_id) if home.thread_id else None) + dedup_key = _notice_target_key(platform.value, home.chat_id, home.thread_id) if dedup_key in notified: continue try: @@ -921,12 +917,10 @@ class GatewayShutdownMixin: await self._cleanup_agent_resources_off_loop(agent, context=context) self._track_deferred_agent_worker(future, agent) - task = asyncio.create_task(_cleanup_when_done()) tasks = getattr(self, "_deferred_agent_cleanup_tasks", None) if tasks is None: tasks = self._deferred_agent_cleanup_tasks = set() - tasks.add(task) - task.add_done_callback(tasks.discard) + self._track_task_in(tasks, asyncio.create_task(_cleanup_when_done())) async def _finalize_session_off_loop(self, *, session_id: Any, platform: str, reason: str, **extra: Any) -> None: """Run hermes_cli.lifecycle.finalize_session off-loop, bounded; on timeout the worker is left alone.""" @@ -1213,8 +1207,7 @@ class GatewayShutdownMixin: "deferring stop() until they finish (cap=%.0fs) so in-flight " "turns are not amputated (#77184)", active, timeout, ) - with suppress(Exception): - self._update_runtime_status("draining") + self._scale_to_zero_status("draining", "restart wait: status mark failed") loop = asyncio.get_running_loop() deadline = loop.time() + timeout last_status_at = 0.0 @@ -1233,8 +1226,7 @@ class GatewayShutdownMixin: "(%d wedged and excluded; %.0fs remaining before force drain)", self._awaitable_work_count(), self._wedged_agent_count(), deadline - now, ) - with suppress(Exception): - self._update_runtime_status("draining") + self._scale_to_zero_status("draining", "restart wait: status mark failed") last_status_at = now await asyncio.sleep(0.1) if self._active_work_count() > 0: @@ -1298,6 +1290,15 @@ class GatewayShutdownMixin: # stop() phases. Invoked as ``GatewayRunner._stop_(self, ctx)`` so shutdown-path tests # can drive them from bare doubles that are not GatewayRunner instances. + @staticmethod + def _quiet_step(label: str, fn: Callable[[], Any]) -> Any: + """Run one best-effort teardown step; a failure is debug-logged as ``"