"""Writable runner initialization, separate from process ownership.""" from __future__ import annotations import logging logger = logging.getLogger("gateway.run") # log-record parity with gateway/run.py class GatewayRuntimeInitMixin: def _init_runtime_settings(self) -> None: """Load ephemeral per-call config (prefill, reasoning, busy modes, timeouts, routing).""" from gateway.run import (Dict, Optional, _load_gateway_config) self._prefill_messages = self._load_prefill_messages() self._reasoning_config = self._load_reasoning_config() self._service_tier = self._load_service_tier() self._show_reasoning = self._load_show_reasoning() self._busy_input_mode = self._load_busy_input_mode() self._busy_text_mode = self._load_busy_text_mode() # Secondary-profile busy modes snapshotted at multiplex startup; handlers never reread config. self._busy_input_modes_by_profile: Dict[str, str] = {} self._busy_text_modes_by_profile: Dict[str, str] = {} self._busy_text_timing = self._busy_text_timing_from_config(_load_gateway_config()) self._busy_text_timing_by_profile: Dict[str, tuple[float, float]] = {} self._human_delay = self._human_delay_from_config(_load_gateway_config()) self._human_delay_by_profile: Dict[str, Optional[tuple[int, int]]] = {} self._restart_drain_timeout = self._load_restart_drain_timeout() # Live launchd ``ExitTimeOut`` for this job (None when not launchd-owned). Read once at # boot — launchd fixes it at load — and applied only to signal-driven stops, which are the # only stops launchd times. See _load_launchd_exit_timeout(). self._stop_requested_by_signal = False self._launchd_exit_timeout_s = self._load_launchd_exit_timeout(self._restart_drain_timeout) self._restart_after_turn_timeout = self._load_restart_after_turn_timeout() self._cron_drain_timeout = self._load_cron_drain_timeout() self._signal_interrupt_grace_timeout = self._load_signal_interrupt_grace_timeout() self._provider_routing = self._load_provider_routing() self._fallback_model = self._load_fallback_model() def _init_session_store(self) -> None: """Build the SessionStore (with process-registry reset guard), its async facade and the router.""" from gateway.run import (AsyncSessionStore, DeliveryRouter, SessionStore) from tools.process_registry import process_registry self.session_store = SessionStore( self.config.sessions_dir, self.config, has_active_processes_fn=lambda key: process_registry.has_active_for_session( key)) # Loop-side boundary: sync helpers use ``session_store`` directly; async handlers await this facade. self._async_session_store = AsyncSessionStore(self.session_store) self.delivery_router = DeliveryRouter(self.config) def _init_lifecycle_state(self) -> None: """Initialise run/exit/restart flags, per-session state, and completion-delivery bookkeeping.""" from gateway.run import (Dict, List, MessageEvent, Optional, OrderedDict, Platform, SessionSource, SessionState, SessionTurnLeaseRegistry, asyncio, concurrent, threading, time) self._running = self._exit_cleanly = self._exit_with_failure = self._draining = False self._gateway_loop: Optional[asyncio.AbstractEventLoop] = None self._shutdown_event = asyncio.Event() self._exit_reason: Optional[str] = None self._exit_code: Optional[int] = None self._profile_failed_platforms: Dict[str, Dict[Platform, asyncio.Task]] = {} self._systemd_watchdog = None # External (NAS-driven) drain, distinct from one-way ``_draining``: set while ``.drain_request.json`` # exists — NEW turns refused, process stays up, removing the marker reverts to ``running``. self._external_drain_active = False # ``_signal_initiated_shutdown``: SIGTERM/SIGINT with no planned-stop/takeover marker (container, # OOM, bare kill); _stop_impl must NOT persist gateway_state=stopped or container_boot won't restart. self._restart_requested = self._signal_initiated_shutdown = self._restart_task_started = False self._restart_detached = self._restart_via_service = self._detached_restart_helper_started = False self._restart_command_source: Optional[SessionSource] = None # Construction clock: bounds the /restart redelivery guard's window (missing dedup marker = stale). self._startup_time: float = time.time() # True when booted from a chat /restart (.restart_notify.json existed). One-shot signal so the # marker-missing fallback suppresses a /restart only when we KNOW we just restarted. self._booted_from_restart: bool = False self._stop_task: Optional[asyncio.Task] = None self._restart_task: Optional[asyncio.Task] = None self._executor_lock = threading.Lock() self._executor: Optional[concurrent.futures.ThreadPoolExecutor] = None # Best-effort session housekeeping runs on its OWN pool; see _run_housekeeping_in_executor. self._housekeeping_executor: Optional[concurrent.futures.ThreadPoolExecutor] = None # Set on gateway stop so the recreate-on-shutdown path can't resurrect the pool. self._executor_closing = False # ALL per-session state lives here (gateway/session_state.py); use _session_state / _peek_session_state. self._sessions: Dict[str, SessionState] = {} # Per-SESSION_ID turn lease: serializes [load history → run → flush] when two ROUTING KEYS resolve # to one session_id (switch_session's many-to-one mapping), which routing-key guards cannot see. self._turn_leases = SessionTurnLeaseRegistry() # Stall-notified keys clear when pending clears / activity resumes / conversation boundary. # Held turn-lease tokens live on SessionState.turn.lease_tokens keyed by run generation, so a # stale unwind can never free a newer turn's lease (#28686). Runner-level queued interrupt text lives on # SessionState.persistent.pending_command_text (NOTE: distinct from the adapter-level # _pending_messages Dict[str, MessageEvent] in gateway/platforms/base.py, which shares the legacy # name). Last successfully-resolved (non-empty) model, keyed by session. Used as a fallback when a # fresh config read transiently returns an empty model (e.g. an mtime-keyed config-cache miss during # a post-interrupt recovery turn). Without this, the agent is built with model="" and every API call # fails HTTP 400 "No models provided" — the session goes silent until the user manually re-sends. # See #35314. The ``"*"`` session entry holds a process-wide last-known-good for sessions seen for # the first time. Lives on SessionState.conversation.last_resolved_model. Overflow buffer for # explicit /queue commands. The adapter-level _pending_messages dict is a single slot per session # (designed for "next-turn" follow-ups where repeated sends collapse into one event). /queue has # different semantics: each invocation must produce its own full agent turn, in FIFO order, with no # merging. When the slot is occupied, additional /queue items land here and are promoted # one-at-a-time after each run's drain. Cleared on /new and /reset. /model and other mid-session # operations preserve the queue. Lives on SessionState.conversation.queued_events; native image # paths, busy-ack debounce timestamps and the monotonic run-generation counter (#28686, NEVER reset) # live on SessionState too. See gateway.session_stall. self._session_stall_notified: Dict[str, bool] = {} # Consecutive "persisted transcript lagged live cached history" turns per session key; see # run_turn_runner._load_turn_history (#114266). Cleared on /new. self._transcript_lag_streaks: Dict[str, int] = {} # Startup restore gate: while restart-interrupted sessions auto-resume, real inbound messages # queue instead of competing with the synthetic resume turns; drained after all resume tasks end. self._startup_restore_in_progress = False self._startup_restore_queue: List[MessageEvent] = [] self._startup_restore_tasks: List[asyncio.Task] = [] # Set by start_gateway() only for an explicit ``--replace`` launch; scoped to each adapter's # cold-start connect and removed before any reconnect can run. self._platform_lock_takeover_on_start = False # Capped LRU of live SessionSources for fallback routing (shutdown notices, synthetic events) when # the persisted origin is missing and _parse_session_key can't recover thread_id. self._session_sources: OrderedDict[str, SessionSource] = OrderedDict() self._session_sources_max = 512 # Lifecycle-scoped completion dedup: closes queue/watcher races inside one gateway without claiming # exactly-once across a crash; durable replay state stays owned by tools.async_delegation. self._completion_delivery_lock = threading.Lock() self._completion_deliveries_inflight: set[tuple[str, str, object]] = set() self._completion_deliveries_delivered: OrderedDict[tuple[str, str, object], None] = OrderedDict() self._completion_delivery_retention = 2048 # Agent-triggered terminal completions from one conversation often land in the same scheduler # tick; hold them briefly so the agent gets one synthetic turn instead of one per process. # See #70300. self._completion_notification_batches: dict[tuple[str, ...], list[tuple[str, dict, asyncio.Future]]] = {} self._completion_notification_batch_tasks: dict[tuple[str, ...], asyncio.Task] = {} self._completion_notification_batch_flush_tasks: set[asyncio.Task] = set() self._completion_notification_batch_window = 0.1 self._completion_notification_batches_stopping = False def _init_runtime_caches(self) -> None: """Agent cache, profile identity, Teams runtime, failed-platform tracking, slash-confirm counter.""" from gateway.run import (Any, Dict, Optional, OrderedDict, Platform, threading) # AIAgent per session preserves prompt caching (fresh agent per message ~10x cost on Anthropic). # Value: (AIAgent, config_signature); LRU cap in _enforce_agent_cache_cap, TTL in expiry watcher. self._agent_cache: OrderedDict[str, tuple] = OrderedDict() self._agent_cache_lock = threading.Lock() # Launch-time identity of the profile that owns ``self.adapters``; ``_authorization_adapter`` # compares against this rather than the per-turn ``_active_profile_name()``. A multiplex # host's primary map is always the default profile, even when a named profile launched # the process (that launcher is a secondary adapter owner). launch = self._active_profile_name() self._kanban_notifier_profile = launch self._primary_profile_name = ( "default" if getattr(self.config, "multiplex_profiles", False) else launch ) # Teams meeting pipeline runtime (bound later when msgraph_webhook adapter exists). self._teams_pipeline_runtime = None self._teams_pipeline_runtime_error: Optional[str] = None # Failed-to-connect platforms for background reconnection: Platform -> {config, attempts, next_retry} self._failed_platforms: Dict[Platform, Dict[str, Any]] = {} # Strong refs to detached fatal-error handler tasks so the loop can't GC them mid-run. self._fatal_handler_tasks: set = set() # Slash-confirm state lives in tools.slash_confirm (module-level) so adapters resolve callbacks # without a runner backref; local counter keeps confirm_ids compact (64-byte callback_data caps). import itertools self._slash_confirm_counter = itertools.count(1) def _init_session_db(self) -> None: """Open the session DB for the active scope and run opportunistic state.db / checkpoint maintenance.""" from gateway.run import (Any, Dict, Path, _SESSION_DB_UNPINNED, _housekeeping_chore, _housekeeping_state_db_maintenance, _launch_sessions_dir, threading) # Session DB is a property caching one AsyncSessionDB per path (a handle bound here would pin the # root home under multiplex); priming here keeps startup diagnostics at init. # Initialize session database for session_search tool support. Same frozen-handle class of bug as # SessionStore._db (#88532): a handle bound here is pinned to the process's root home, but /resume, # /title, /history and session search all run inside _profile_runtime_scope on a multiplexed gateway # and must see that profile's own state.db. self._session_db_pinned: Any = _SESSION_DB_UNPINNED self._session_db_handles: Dict[Path, Any] = {} self._session_db_handles_lock = threading.Lock() from gateway.session_db_recovery import RecoverableHandleCache self._session_db_handle_cache = RecoverableHandleCache( handles=self._session_db_handles, lock=self._session_db_handles_lock) try: self._open_session_db_for_active_scope(raise_on_error=True) except Exception as e: # WARNING (not DEBUG) so it lands in errors.log; else an NFS HERMES_HOME silently loses /resume etc. logger.warning("SQLite session store not available: %s", e, exc_info=True) self._session_db_init_error = str(e) # surfaced on the home channel(s) once connected # Opportunistic state.db maintenance (prune + optional VACUUM), at most once per min_interval_hours. # A few blocking seconds per day is fine for a long-lived gateway; failures log, never raise. # Surface the failure to the user via their home channel(s) once the gateway connects. Without this, # state.db corruption or NFS/SMB lock failures silently degrade the entire gateway — messages may # flow but nothing is persisted, and the user has no indication until they try /resume and find # nothing (#88235). # Once per SERVED profile, each under its own scope: both the store and the ``sessions:`` # config that governs it must be the profile's own. Bound to ``self._session_db`` this ran # against the construction-time launch home only, so a multiplexed secondary profile's # state.db was never pruned or vacuumed by anybody, and the launch profile's # retention_days/auto_prune decided whether it happened at all. from gateway.run_profile_reconcile import _for_each_served_profile _launch_sessions = _launch_sessions_dir(self.config) # resolved OUTSIDE any profile scope _housekeeping_chore( "state.db startup maintenance", lambda: _for_each_served_profile( self, lambda _label: _housekeeping_state_db_maintenance(_launch_sessions))) # Checkpoint store pruning is a housekeeping chore (``_housekeeping_checkpoint_prune``), not a # constructor step: its ``git gc`` repacks the whole store (tens of seconds on a GB store) and # here it ran before the control socket, adapters and the code_sha stamp — so the first # restart of the day (the ``hermes update`` one) looked hung and failed fleet verification. def _init_registries_and_clocks(self) -> None: """Pairing stores, hook registry, voice modes, background-task set, liveness and idle clocks.""" from gateway.run import (Dict, List, Optional, asyncio, time) # ``pairing_store``: global/default store (CLI, callers without profile context); ``pairing_stores``: # per-profile map ``authz_mixin._is_user_authorized`` routes through (one whitelist per profile). from gateway.pairing import PairingStore from gateway.hooks import ProfileHookRegistries self.pairing_store = PairingStore() self.pairing_stores: Dict[str, PairingStore] = {} # One HookRegistry per served profile home, resolved from the active scope at emit time. self.hooks = ProfileHookRegistries() # Per-chat voice reply mode: "off" | "voice_only" | "all" self._voice_mode: Dict[str, str] = self._load_voice_modes() # Per-(bot, guild, text channel, user) transcript dedup: voice/STT can emit one utterance twice. self._recent_voice_transcripts: Dict[tuple, List[tuple[float, str]]] = {} # Background tasks kept referenced so they are not garbage-collected mid-execution. self._background_tasks: set = set() # Event-loop liveness heartbeat: rewritten every 30s while the loop dispatches; supervisors use # the file mtime / updated_at to tell "process alive" from "loop frozen". # See #66892. self._gateway_started_at: float = time.time() self._loop_heartbeat_task: Optional[asyncio.Task] = None self._loop_floor_timer_handle = self._loop_liveness_watchdog = None # scale-to-zero: gateway-scoped "last inbound seen" clock, stamped in _handle_message (the single # inbound chokepoint) and seeded to "now" so a fresh gateway isn't idle from epoch. self._last_inbound_at: float = time.time() # Re-arm cooldown after a wake so we don't go dormant again before the drained backlog updates # the clock; and a one-shot latch so the "platform owns the suspend" notice logs once. self._scale_to_zero_cooldown_until: float = 0.0 self._scale_to_zero_no_suspend_logged: bool = False self._scale_to_zero_direct_platform_logged: bool = False