fix: core modules import again on Python 3.11-3.13 (re-quote eager forward references)

#135348 ran ruff's UP037 autofix with ruff.strict.toml's target-version = "py314". On 3.14
annotations are lazy (PEP 649), so stripping quotes is safe there; on 3.11-3.13 a function or
variable annotation is evaluated when the def or statement runs. Every unquoted annotation that
names the enclosing class (`def clone_for_agent(self) -> ContextEngine`), a class bound later in
the module, or a TYPE_CHECKING-only import now raises NameError on import: 300 modules on main
(gateway.config, run_agent, hermes_state, cron.jobs, the Telegram adapter, ...) fail to import
under 3.11, the update path crashes mid-`hermes update` on any install whose interpreter is
older than 3.14, and the e2e-upgrade suite's retry leg fails with
`NameError: name 'ContextEngine' is not defined`.

- Re-quote exactly the 79 annotations (36 files) that name something unbound when they are
  evaluated (including tui_gateway/server.py's `_TurnScopes`, which only reaches that module's
  globals through register() after import); no other text changes.
- agent/turn_scripted_prelude.py: `Generator[...]` gets its third argument back. d76cce4f11
  applied UP043 (default type args, valid only from 3.13) to a module-level alias, so
  `Generator[Y, S]` raises TypeError on 3.11/3.12 and takes gateway.run, agent.conversation_loop
  and tui_gateway.{server,ws,entry} down with it.
- ruff.strict.toml target-version = "py311", the requires-python floor. With it UP037 leaves
  these quotes alone and only strips the two that are genuinely redundant, which this commit
  strips; any future version-gated fix stays valid on every supported Python.
- tests/ci/test_strict_ruff_target.py pins the target to the requires-python floor (red on main).

agent/context_engine.py

@@ -223,7 +223,7 @@ class ContextEngine(ABC):
             "compression_count": self.compression_count,
         }
 
-    def clone_for_agent(self) -> ContextEngine:
+    def clone_for_agent(self) -> "ContextEngine":
         """Per-agent instance of a plugin-registered engine (the plugin system holds ONE shared
         instance; every AIAgent gets its own so a child's update_model() cannot mutate the parent's).
         Override when the engine holds uncopyable state (locks, DB connections): return a fresh

agent/turn_scripted_prelude.py

@@ -21,7 +21,7 @@ from agent.message_sanitization import deterministic_call_id
 from agent.transports.types import NormalizedResponse, build_tool_call
 from agent.turn_tool_round import run_tool_round
 
-Prelude = Generator[tuple[str, str, dict], Optional[str]]
+Prelude = Generator[tuple[str, str, dict], Optional[str], None]
 
 
 def run_scripted_prelude(agent: Any, s: Any, prelude: Prelude) -> Any:

cron/jobs.py

@@ -121,7 +121,7 @@ class _CronStorePaths:
     output_dir: Path
 
     @classmethod
-    def for_dir(cls, cron_dir: Path) -> _CronStorePaths:
+    def for_dir(cls, cron_dir: Path) -> "_CronStorePaths":
         return cls(cron_dir, cron_dir / "jobs.json", cron_dir / "output")
 
 

gateway/config.py

@@ -267,7 +267,7 @@ class Platform(Enum):
         return cls._add_pseudo_member(value) if registered else None
 
     @classmethod
-    def _add_pseudo_member(cls, value: str) -> Platform:
+    def _add_pseudo_member(cls, value: str) -> "Platform":
         pseudo = object.__new__(cls)
         pseudo._value_ = value
         pseudo._name_ = value.upper().replace("-", "_").replace(" ", "_")
@@ -359,7 +359,7 @@ class HomeChannel:
         return {"platform": self.platform.value, "chat_id": self.chat_id, "name": self.name, **optional}
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> HomeChannel:
+    def from_dict(cls, data: dict[str, Any]) -> "HomeChannel":
         optional = {k: str(data[k]) if data.get(k) else None for k in ("thread_id", "user_id", "scope_id")}
         return cls(platform=Platform(data["platform"]), chat_id=str(data["chat_id"]), name=data.get("name", "Home"), **optional)
 
@@ -386,7 +386,7 @@ class ChannelOverride:
         return {k: v for k, v in asdict(self).items() if v is not None}
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> ChannelOverride:
+    def from_dict(cls, data: dict[str, Any]) -> "ChannelOverride":
         return cls(**{f.name: data.get(f.name) for f in fields(cls)}) if data else cls()
 
 
@@ -442,7 +442,7 @@ class PlatformConfig:
     })
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> PlatformConfig:
+    def from_dict(cls, data: dict[str, Any]) -> "PlatformConfig":
         data = _coerce_dict(data)
         home = data.get("home_channel")
         # Adapters read their settings from ``extra`` (``config.extra.get("port")``), but users
@@ -518,7 +518,7 @@ class StreamingConfig:
         return asdict(self)
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> StreamingConfig:
+    def from_dict(cls, data: dict[str, Any]) -> "StreamingConfig":
         if not isinstance(data, dict) or not data:
             return cls()
 
@@ -721,7 +721,7 @@ class GatewayConfig:
         }
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> GatewayConfig:
+    def from_dict(cls, data: dict[str, Any]) -> "GatewayConfig":
         data = _coerce_dict(data)
         nested_gateway = _coerce_dict(data.get("gateway"))
 

gateway/delivery.py

@@ -121,7 +121,7 @@ class DeliveryTarget:
     unknown_platform: Optional[str] = None
 
     @classmethod
-    def parse(cls, target: str, origin: Optional[SessionSource] = None) -> DeliveryTarget:
+    def parse(cls, target: str, origin: Optional[SessionSource] = None) -> "DeliveryTarget":
         """Parse "origin" | "local" | "<platform>" | "<platform>:<chat_id>[:<thread_id>]"."""
         target = target.strip()
         if target.lower() == "origin":

gateway/platforms/base.py

@@ -2135,7 +2135,7 @@ class BasePlatformAdapter(ABC):
         return t("gateway.progress.tool_preview", emoji=emoji, tool=tool, preview=self.format_tool_preview(prepared))
 
 
-    def format_tool_preview(self, preview: ToolPreview) -> str:
+    def format_tool_preview(self, preview: "ToolPreview") -> str:
         """Platform-native formatting of a compact tool preview; rich-text adapters may use
         the preview's metadata (e.g. a URL shortened for display)."""
         return preview.text
@@ -2155,7 +2155,7 @@ class BasePlatformAdapter(ABC):
         return chat_id in self._auto_tts_enabled_chats or (
             chat_id not in self._auto_tts_disabled_chats and bool(self._auto_tts_default))
 
-    def set_fatal_error_handler(self, handler: Callable[[BasePlatformAdapter], Awaitable[None] | None]) -> None:
+    def set_fatal_error_handler(self, handler: "Callable[[BasePlatformAdapter], Awaitable[None] | None]") -> None:
         self._fatal_error_handler = handler
 
     #: Published when an adapter is installed and running but its receive
@@ -3628,7 +3628,7 @@ class BasePlatformAdapter(ABC):
             logger.debug("[%s] Failed to resolve media delivery scope", self.name, exc_info=True)
             return contextlib.nullcontext()
 
-    def _final_delivery_adapter(self, source: Optional[SessionSource]) -> BasePlatformAdapter:
+    def _final_delivery_adapter(self, source: Optional[SessionSource]) -> "BasePlatformAdapter":
         """The runner's CURRENT adapter for a new final-response send: a reconnect can swap the
         registry adapter mid-task; an unsent final response belongs on the replacement transport,
         while message IDs, edits and deletes stay owned by the old one (nothing is migrated)."""
@@ -4244,7 +4244,7 @@ class BasePlatformAdapter(ABC):
 
     async def _record_delivery_obligation(
         self, event: MessageEvent, session_key: str, text_content: str,
-        delivery_adapter: BasePlatformAdapter, is_ephemeral_response: bool) -> Optional[str]:
+        delivery_adapter: "BasePlatformAdapter", is_ephemeral_response: bool) -> Optional[str]:
         """Ledger the final response BEFORE the send so a crash before platform ACK redelivers on
         next boot; best-effort, skips slash-command and ephemeral replies. Returns the obligation id
         or None."""
@@ -4278,7 +4278,7 @@ class BasePlatformAdapter(ABC):
 
     async def _finalize_delivery_obligation(
         self, obligation_id: str, result: Any, event: MessageEvent,
-        delivery_adapter: BasePlatformAdapter) -> None:
+        delivery_adapter: "BasePlatformAdapter") -> None:
         """Mark the ledger row delivered/failed (best-effort). On ``send_path_degraded`` with a
         replacement adapter live, trigger another redelivery sweep (the watcher's may have run
         before this failure landed; atomic claiming keeps it idempotent). On any other rejection arm
@@ -4378,7 +4378,7 @@ class BasePlatformAdapter(ABC):
     async def send_final_ledgered(
         self, event: MessageEvent, session_key: str, text_content: str, metadata: dict[str, Any], *,
         reply_to: Optional[str], is_ephemeral_response: bool = False,
-    ) -> tuple[SendResult, BasePlatformAdapter]:
+    ) -> "tuple[SendResult, BasePlatformAdapter]":
         """The delivery-ledger bracket every final text goes through, on the CURRENT transport
         (a reconnect may have replaced this adapter): record the obligation before the send,
         send with retry, finalize from the result — so a refused final (flood control, a dead

gateway/platforms/event.py

@@ -107,7 +107,7 @@ class MessageEvent:
     # Run-owned final presentation snapshot; never deserialized from ingress metadata.
     _notification_reply_muted: Optional[bool] = field(default=None, init=False, repr=False, compare=False)
 
-    def absorb_reply_expected(self, other: MessageEvent) -> None:
+    def absorb_reply_expected(self, other: "MessageEvent") -> None:
         """One turn now answers *other* too: an addressed message wins, then an unknown one."""
         if self.reply_expected is not True and other.reply_expected is not False:
             self.reply_expected = other.reply_expected

gateway/platforms/helpers.py

@@ -60,7 +60,7 @@ class MessageDeduplicator:
     def clear(self):
         self._seen.clear()
 
-    def absorb(self, other: MessageDeduplicator) -> None:
+    def absorb(self, other: "MessageDeduplicator") -> None:
         """Adopt *other*'s still-live IDs (at their original seen times) into this cache."""
         cutoff = time.time() - self._ttl
         self._seen.update({k: v for k, v in other._seen.items() if v > cutoff and k not in self._seen})

gateway/platforms/webhook.py

@@ -80,7 +80,7 @@ class _WebhookDeliveryIdentity:
     delivery_id: str
 
     @classmethod
-    def from_parts(cls, profile: Optional[str], route: str, delivery_id: str) -> _WebhookDeliveryIdentity:
+    def from_parts(cls, profile: Optional[str], route: str, delivery_id: str) -> "_WebhookDeliveryIdentity":
         return cls(profile=profile or "default", route=route, delivery_id=delivery_id)
 
     @property

gateway/run.py

@@ -1828,7 +1828,7 @@ async def _async_profile_runtime_scope(profile_home: Path):
         yield
 
 
-def load_gateway_config_for_runner() -> GatewayConfig:
+def load_gateway_config_for_runner() -> "GatewayConfig":
     """Load gateway config for the process-level GatewayRunner. An UNSET ``multiplex_profiles`` is
     settled first by ``resolve_multiplex_mode`` (the default is on; the boot guard keeps a fleet that
     still runs per-profile gateways standalone). Multiplexed: set multiplex-active, then reload
@@ -1898,7 +1898,7 @@ async def _discover_gateway_mcp_tools(config: object) -> None:
                 logger.warning("MCP tool discovery failed for profile '%s'", profile_name, exc_info=True)
 
 
-def _platform_has_bot_credential(platform: Platform, platform_config: PlatformConfig) -> bool:
+def _platform_has_bot_credential(platform: "Platform", platform_config: "PlatformConfig") -> bool:
     """Return True when a token-authenticated platform has a usable bot credential; platforms not using
     ``PlatformConfig.token`` (Signal session paths, port-binding HTTP adapters) always return True."""
     from gateway.config import PLATFORM_TOKEN_ENV_NAMES, Platform

gateway/session.py

@@ -147,7 +147,7 @@ class SessionSource:
         return d
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> SessionSource:
+    def from_dict(cls, data: dict[str, Any]) -> "SessionSource":
         plain = {
             name: data.get(name)
             for name in cls._ALWAYS_FIELDS[1:] + cls._OPTIONAL_PRE_SCOPE + cls._OPTIONAL_POST_SCOPE + cls._OPTIONAL_TAIL
@@ -580,7 +580,7 @@ class SessionEntry:
         return result
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> SessionEntry:
+    def from_dict(cls, data: dict[str, Any]) -> "SessionEntry":
         origin = data.get("origin")
         origin = SessionSource.from_dict(origin) if isinstance(origin, dict) else None
         platform = None
@@ -767,7 +767,7 @@ class _RouteDecision:
 class AsyncSessionStore:
     """Async boundary for the synchronous, thread-safe SessionStore."""
 
-    def __init__(self, store: SessionStore) -> None:
+    def __init__(self, store: "SessionStore") -> None:
         self._store = store
 
     def __getattr__(self, name: str):

gateway/turn_lease.py

@@ -25,7 +25,7 @@ DEFAULT_MAX_LEASES = 512
 DEFAULT_LEASE_WAIT = 5.0
 
 
-def _holder_desc(holder: Optional[TurnLeaseToken]) -> tuple:
+def _holder_desc(holder: "Optional[TurnLeaseToken]") -> tuple:
     return (holder.owner_key, holder.generation) if holder else ("?", "?")
 
 
@@ -45,7 +45,7 @@ class TurnLeaseToken:
 
     __slots__ = ("generation", "lease", "owner_key", "released", "session_id")
 
-    def __init__(self, session_id: str, owner_key: str, generation: int, lease: _SessionLease) -> None:
+    def __init__(self, session_id: str, owner_key: str, generation: int, lease: "_SessionLease") -> None:
         self.session_id, self.owner_key, self.generation = session_id, owner_key, generation
         self.released = False
         # The concrete lease, so release resolves by identity even after a rotation re-aliases

hermes_cli/banner.py

@@ -685,7 +685,7 @@ def _banner_skill_lines(skills_by_category: dict[str, list[str]], skills_enabled
 
 
 def build_welcome_banner(
-    console: Console, model: str, cwd: str, tools: list[dict] | None = None, enabled_toolsets: list[str] | None = None,
+    console: "Console", model: str, cwd: str, tools: list[dict] | None = None, enabled_toolsets: list[str] | None = None,
     session_id: str | None = None, get_toolset_for_tool=None, context_length: int | None = None, provider: str | None = None,
     availability: dict[str, Any] | None = None, skills_by_category: dict[str, list[str]] | None = None,
     context_pinned: bool = False,

hermes_cli/local_runtime/hardware.py

@@ -419,7 +419,7 @@ def _gpu_engine_runs_the_model() -> bool:
     return False
 
 
-def _windows_igpu_bytes(device: "dict | None") -> int | None:
+def _windows_igpu_bytes(device: dict | None) -> int | None:
     """What Windows lets the integrated GPU allocate (Task Manager's GPU memory), or None.
 
     The adapter is the one the engine named, else the only non-NVIDIA hardware adapter when a

hermes_cli/web_server_config.py

@@ -477,7 +477,7 @@ def _normalize_main_model_assignment(provider: str, model: str) -> tuple[str, st
 
 def _validated_main_model_selection(
     cfg: dict, provider: str, model: str, base_url: str = "", api_key: str = ""
-) -> ModelSwitchResult:
+) -> "ModelSwitchResult":
     """Route a dashboard main-slot pick through ``switch_model`` (catalog/alias/credential
     validation) seeded with the configured route, exactly like a ``/model <model> --provider
     <provider> --global``. A bare ``custom`` target carries the submitted endpoint as the current
@@ -510,7 +510,7 @@ def _validated_main_model_selection(
     return result
 
 
-def _apply_main_model_assignment(model_cfg: Any, result: ModelSwitchResult, api_key: str = "") -> dict:
+def _apply_main_model_assignment(model_cfg: Any, result: "ModelSwitchResult", api_key: str = "") -> dict:
     """Apply a main-slot selection to a ``model`` config dict via the canonical /model shape
     (``hermes_cli.model_switch.apply_model_selection``). An explicit key for a custom endpoint is
     the one inline credential the runtime reads (``model.api_key``); the legacy ``api`` alias is
@@ -739,7 +739,7 @@ def _provider_entry(cfg: dict, provider: str) -> Any:
     return providers_cfg.get(provider) if isinstance(providers_cfg, dict) else None
 
 
-def _prepare_main_assignment(cfg: dict, provider: str, model: str, base_url: str, api_key: str) -> tuple[str, ModelSwitchResult]:
+def _prepare_main_assignment(cfg: dict, provider: str, model: str, base_url: str, api_key: str) -> "tuple[str, ModelSwitchResult]":
     """Validation half of a main-slot assignment: ``(effective base_url, switch result)``.
     ``switch_model`` fetches catalogs / probes endpoints, so callers run this BEFORE taking
     ``_CONFIG_MUTATION_LOCK``; it writes nothing."""
@@ -753,7 +753,7 @@ def _prepare_main_assignment(cfg: dict, provider: str, model: str, base_url: str
 
 
 def _apply_main_assignment_sync(cfg: dict, provider: str, model: str, base_url: str, api_key: str,
-                                prepared: Optional[tuple[str, ModelSwitchResult]] = None) -> dict:
+                                prepared: "Optional[tuple[str, ModelSwitchResult]]" = None) -> dict:
     from hermes_cli.config import save_config
     from hermes_cli.free_tier_bootstrap import reconcile_record
     base_url, result = prepared or _prepare_main_assignment(cfg, provider, model, base_url, api_key)
@@ -888,7 +888,7 @@ def _apply_aux_assignment_sync(cfg: dict, provider: str, model: str, task: str,
 
 def _apply_model_assignment_sync(
     scope: str, provider: str, model: str, task: str, base_url: str, api_key: str = "",
-    reasoning_effort: Optional[str] = _UNSET, prepared: Optional[tuple[str, ModelSwitchResult]] = None,
+    reasoning_effort: Optional[str] = _UNSET, prepared: "Optional[tuple[str, ModelSwitchResult]]" = None,
 ):
     """Synchronous body of POST /api/model/set.
 

hermes_cli/web_server_lifecycle.py

@@ -194,7 +194,7 @@ def _eager_reconcile_own_session_db() -> None:
         )
 
 
-def _read_bound_port(server: uvicorn.Server, fallback: int) -> int:
+def _read_bound_port(server: "uvicorn.Server", fallback: int) -> int:
     """Read the OS-assigned port from the live uvicorn socket (ephemeral port-0 discovery)."""
     if server.servers and server.servers[0].sockets:
         return server.servers[0].sockets[0].getsockname()[1]

hermes_cli/web_server_mcp.py

@@ -105,7 +105,7 @@ def _mcp_server_summary(name: str, cfg: dict[str, Any], plugin: str | None = Non
     }
 
 
-_mcp_oauth_flows: dict[str, DashboardOAuthFlow] = {}
+_mcp_oauth_flows: "dict[str, DashboardOAuthFlow]" = {}
 _mcp_oauth_transactions: dict[tuple[str, str], threading.Lock] = {}
 _mcp_oauth_transactions_lock = threading.Lock()
 

hermes_platform/host/gpu_adapters.py

@@ -59,7 +59,7 @@ def _method(obj: ctypes.c_void_p, slot: int, *argtypes):
 
 
 @functools.cache
-def windows_gpu_adapters() -> "tuple[Adapter, ...]":
+def windows_gpu_adapters() -> tuple[Adapter, ...]:
     """Every adapter DXGI enumerates, in its order; empty off Windows or when DXGI can't answer."""
     with suppress(OSError, AttributeError):
         factory = ctypes.c_void_p()

hermes_state.py

@@ -1488,7 +1488,7 @@ class SessionDB(
         except Exception as exc:
             logger.warning("WAL checkpoint (PASSIVE) failed: %s", exc)
 
-    def __enter__(self) -> SessionDB:
+    def __enter__(self) -> "SessionDB":
         """``with SessionDB(path) as db:`` closes on exit; owners must release deterministically.
 
         Ownership of a SessionDB should be released explicitly. Historically an instance with a started

hermes_state_readpool.py

@@ -157,7 +157,7 @@ class _PathReadBudget:
         self._members: weakref.WeakSet[SessionDB] = weakref.WeakSet()
         self._duplicate_handles_warned = False
 
-    def register(self, db: SessionDB) -> None:
+    def register(self, db: "SessionDB") -> None:
         with self._lock:
             self._members.add(db)
             # Only writable handles carry the cost the warning names (writer connection, write
@@ -190,12 +190,12 @@ class _PathReadBudget:
                 handles, db.db_path, _READ_POOL_MAX, creation_sites,
             )
 
-    def unregister(self, db: SessionDB) -> None:
+    def unregister(self, db: "SessionDB") -> None:
         """Remove a closed writer from duplicate-handle diagnostics immediately."""
         with self._lock:
             self._members.discard(db)
 
-    def acquire(self, requester: SessionDB) -> bool:
+    def acquire(self, requester: "SessionDB") -> bool:
         """Take a permit for a new read connection, or refuse (caller degrades to the
         locked writer connection). Gates, broadest first: fd headroom, process
         ceiling, this file's ceiling."""
@@ -222,12 +222,12 @@ class _PathReadBudget:
             _reclaim_idle_read_conn_anywhere() and _process_read_permits.acquire(blocking=False)
         )
 
-    def _acquire_path_permit(self, requester: SessionDB) -> bool:
+    def _acquire_path_permit(self, requester: "SessionDB") -> bool:
         return self.permits.acquire(blocking=False) or (
             self.reclaim_idle(exclude=requester) and self.permits.acquire(blocking=False)
         )
 
-    def reclaim_idle(self, exclude: Optional[SessionDB] = None) -> bool:
+    def reclaim_idle(self, exclude: "Optional[SessionDB]" = None) -> bool:
         """Close one idle pooled connection held by a member; True if one went.
         Its release() returns both permits, so both ceilings reclaim through here."""
         with self._lock:

plugins/memory/holographic/store.py

@@ -302,7 +302,7 @@ class MemoryStore:
                         MemoryStore._shared.pop(self._key, None)
             self._entry = None
 
-    def __enter__(self) -> MemoryStore:
+    def __enter__(self) -> "MemoryStore":
         return self
 
     def __exit__(self, *_: object) -> None:

plugins/platforms/telegram/adapter.py

@@ -751,7 +751,7 @@ class TelegramAdapter(TelegramHeldInboundMixin, BasePlatformAdapter):
             return False
         return not bool(getattr(self, "_fatal_error_retryable", True))
 
-    def _replacement_telegram_adapter(self) -> Optional[TelegramAdapter]:
+    def _replacement_telegram_adapter(self) -> "Optional[TelegramAdapter]":
         """Live adapter if the reconnect watcher replaced us in ``runner.adapters`` (an in-flight
         ``send()`` still holds the old instance whose ``_bot`` stays None)."""
         runner = getattr(self, "gateway_runner", None)

plugins/platforms/telegram/telegram_context.py

@@ -11,7 +11,7 @@ if TYPE_CHECKING:
     from plugins.platforms.telegram.adapter import TelegramAdapter
 
 
-def mentions_other_participants(adapter: TelegramAdapter, message: Message) -> bool:
+def mentions_other_participants(adapter: "TelegramAdapter", message: "Message") -> bool:
     """True when a ``mention``/``text_mention`` entity names someone other than this bot."""
     own = adapter._current_bot_username()
     bot_id = getattr(adapter._bot, "id", None) if adapter._bot else None
@@ -29,7 +29,7 @@ def mentions_other_participants(adapter: TelegramAdapter, message: Message) -> b
     return False
 
 
-def group_trigger_text(adapter: TelegramAdapter, message: Message, text: Optional[str]) -> Optional[str]:
+def group_trigger_text(adapter: "TelegramAdapter", message: "Message", text: Optional[str]) -> Optional[str]:
     """Strip our own handle only when we are the sole addressee. With other participants named,
     ``@research_bot , @ops_bot are you both listening?`` must not reach us as ``, @ops_bot …``."""
     if adapter._is_group_chat(message) and mentions_other_participants(adapter, message):
@@ -38,7 +38,7 @@ def group_trigger_text(adapter: TelegramAdapter, message: Message, text: Optiona
 
 
 def group_identity_prompt(
-    adapter: TelegramAdapter, message: Message, channel_prompt: Optional[str],
+    adapter: "TelegramAdapter", message: "Message", channel_prompt: Optional[str],
 ) -> Optional[str]:
     """Session-stable identity line so the model can read retained @mentions as itself or not."""
     if not adapter._is_group_chat(message) or not getattr(adapter, "_bot", None):

plugins/platforms/telegram/telegram_held_inbound.py

@@ -72,7 +72,7 @@ class TelegramHeldInboundMixin:
         if schedule and not self._should_drop_delayed_delivery():
             self._schedule_held_inbound_redispatch()
 
-    def adopt_held_inbound(self, predecessor: TelegramHeldInboundMixin) -> None:
+    def adopt_held_inbound(self, predecessor: "TelegramHeldInboundMixin") -> None:
         """Take over the hold queue of the instance the runner just replaced with us (#132829): it only
         drains on its own ``_mark_connected``, which never comes; later holds there forward here."""
         self._held_inbound_successor = None  # we own the queue again: a reverse link would forward in a cycle

ruff.strict.toml

@@ -2,8 +2,9 @@
 line-length = 88
 indent-width = 4
 
-# Assume Python 3.14
-target-version = "py314"
+# The oldest Python we support (requires-python): quote-stripping (UP037) and every other
+# version-gated fix must stay valid where annotations are evaluated eagerly (< 3.14).
+target-version = "py311"
 
 [format]
 # Like Black, use double quotes for strings.

run_agent.py

@@ -54,7 +54,7 @@ def _launch_cwd_for_session(source: str) -> Optional[str]:
         return None
 
 
-def _gateway_origin_json(agent: AIAgent) -> Optional[str]:
+def _gateway_origin_json(agent: "AIAgent") -> Optional[str]:
     """Gateway routing ``origin_json`` for a session row; None when the agent carries no gateway identity.
 
     Mirrors ``SessionSource.to_dict()`` so state.db consumers see the same fields ``record_gateway_session_peer`` writes.

tests/agent/test_codex_app_server_thread_resume.py

@@ -22,7 +22,7 @@ class _WireClient:
     id (or refuses ids in ``dead``), and records every JSON-RPC method it saw."""
 
     dead: set[str] = set()
-    instances: list[_WireClient] = []
+    instances: "list[_WireClient]" = []
     counter = 0
 
     def __init__(self, **kwargs):

tests/ci/test_strict_ruff_target.py

@@ -0,0 +1,23 @@
+"""The strict lint config may only rewrite code to syntax the oldest supported Python runs.
+
+ruff's version-gated autofixes (UP037 strips annotation quotes, UP017/UP006/UP007 swap in newer
+spellings) are safe only for ``target-version`` and newer. The Python floor is ``requires-python``
+in pyproject.toml; before 3.14 annotations are evaluated eagerly, so a quote stripped for a py314
+target (a method annotated with its own class, a later-defined class) raises NameError at import
+on 3.11-3.13. Pinning the target to the floor keeps every such fix valid on every supported Python.
+"""
+
+import re
+import tomllib
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def test_strict_ruff_target_is_the_supported_python_floor():
+    floor = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]["requires-python"]
+    major, minor = re.search(r">=\s*(\d+)\.(\d+)", floor).groups()
+    strict = tomllib.loads((ROOT / "ruff.strict.toml").read_text(encoding="utf-8"))
+    assert strict["target-version"] == f"py{major}{minor}", (
+        f"ruff.strict.toml target-version {strict['target-version']!r} is not the requires-python "
+        f"floor {floor!r}: its autofixes would emit syntax the oldest supported Python cannot import")

tests/gateway/test_feishu.py

@@ -1497,7 +1497,7 @@ class TestPendingInboundQueue(unittest.TestCase):
 class TestWebhookSecurity(unittest.TestCase):
     """Tests for webhook signature verification, rate limiting, and body size limits."""
 
-    def _make_adapter(self, encrypt_key: str = "") -> FeishuAdapter:
+    def _make_adapter(self, encrypt_key: str = "") -> "FeishuAdapter":
         from gateway.config import PlatformConfig
         from plugins.platforms.feishu.adapter import FeishuAdapter
 

tools/browser_lightpanda.py

@@ -28,7 +28,7 @@ _READY_TIMEOUT_S = 10.0
 _POLL_INTERVAL_S = 0.1
 _STDERR_TAIL_LIMIT = 2000
 
-_servers: dict[str, LightpandaServer] = {}
+_servers: "dict[str, LightpandaServer]" = {}
 _servers_lock = threading.Lock()
 
 

tools/environments/base.py

@@ -56,7 +56,7 @@ _activity_callback_local = threading.local()
 # Foreground commands in flight in THIS process, across every environment. Each runs in its
 # own session/process group, so a host that exits mid-command (TUI client gone, SIGTERM) would
 # orphan the whole tree; the process-exit funnel ``cleanup_all_environments`` kills them.
-_live_foreground: dict[int, tuple[BaseEnvironment, ProcessHandle]] = {}
+_live_foreground: "dict[int, tuple[BaseEnvironment, ProcessHandle]]" = {}
 # Reentrant, and the hard-exit path only ever takes it with a timeout: a signal handler can run
 # on a thread that already holds it.
 _live_foreground_cond = threading.Condition(threading.RLock())
@@ -74,7 +74,7 @@ def _enter_foreground_spawn() -> bool:
         return True
 
 
-def _leave_foreground_spawn(env: BaseEnvironment, spawned) -> bool:
+def _leave_foreground_spawn(env: "BaseEnvironment", spawned) -> bool:
     """Publish ``spawned`` (None: the spawn failed); True when the exit fence went up meanwhile."""
     global _spawns_in_flight
     with _live_foreground_cond:

tools/mcp_oauth.py

@@ -275,7 +275,7 @@ def _reserve_callback_port() -> int:
     return _bind_reserved(0)  # type: ignore[return-value]  # port 0 never returns None
 
 
-def _cached_client_info(storage: HermesTokenStorage | None) -> dict | None:
+def _cached_client_info(storage: "HermesTokenStorage | None") -> dict | None:
     """The on-disk client registration for *storage*, or None."""
     try:
         info = _read_json(storage._client_info_path()) if storage is not None else None
@@ -286,7 +286,7 @@ def _cached_client_info(storage: HermesTokenStorage | None) -> dict | None:
     return info if isinstance(info, dict) else None
 
 
-def _cached_redirect(storage: HermesTokenStorage | None) -> tuple[str | None, int | None]:
+def _cached_redirect(storage: "HermesTokenStorage | None") -> tuple[str | None, int | None]:
     """``(https proxy URI, loopback callback port)`` from the cached client registration (None when
     absent): a DCR ``client_id`` is bound to its registered redirect URI, so a new random port under
     it gets ``redirect_uri does not match any registered URIs``."""
@@ -483,11 +483,11 @@ class HermesTokenStorage:
         self.loaded_issuer = data.pop("hermes_issuer", None)
         self._rebase_expires_in(data)
 
-    async def get_tokens(self) -> OAuthToken | None:
+    async def get_tokens(self) -> "OAuthToken | None":
         self.loaded_issuer = None
         return self._load_model(self._tokens_path(), "OAuthToken", "tokens", self._fixup_loaded_tokens)
 
-    async def set_tokens(self, tokens: OAuthToken) -> None:
+    async def set_tokens(self, tokens: "OAuthToken") -> None:
         payload = _model_json(tokens)
         # Absolute ``expires_at``: see _rebase_expires_in.
         if payload.get("expires_in") is not None:
@@ -545,7 +545,7 @@ class HermesTokenStorage:
             return True
         return False
 
-    async def get_client_info(self) -> OAuthClientInformationFull | None:
+    async def get_client_info(self) -> "OAuthClientInformationFull | None":
         coerced: list[bool] = []
         info = self._load_model(
             self._client_info_path(), "OAuthClientInformationFull", "client info",
@@ -554,19 +554,19 @@ class HermesTokenStorage:
             _write_json(self._client_info_path(), _model_json(info))  # persist so later flows skip the coercion
         return info
 
-    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+    async def set_client_info(self, client_info: "OAuthClientInformationFull") -> None:
         data = _model_json(client_info)
         self._coerce_secret_auth_method(data)
         _write_json(self._client_info_path(), data)
         logger.debug("OAuth client info saved for %s", self._server_name)
 
-    def save_oauth_metadata(self, metadata: OAuthMetadata) -> None:
+    def save_oauth_metadata(self, metadata: "OAuthMetadata") -> None:
         """Persist server metadata so a restarted process can refresh without re-discovery;
         otherwise the SDK guesses ``{server_url}/token`` (404) and forces re-auth."""
         _write_json(self._meta_path(), _model_json(metadata))
         logger.debug("OAuth metadata saved for %s", self._server_name)
 
-    def load_oauth_metadata(self) -> OAuthMetadata | None:
+    def load_oauth_metadata(self) -> "OAuthMetadata | None":
         return self._load_model(self._meta_path(), "OAuthMetadata", "OAuth metadata")
 
     def mark_cimd_rejected(self) -> None:
@@ -1119,7 +1119,7 @@ def apply_oauth_provider_defaults(cfg: dict, *, server_name: str = "", server_ur
     return cfg
 
 
-def _build_client_metadata(cfg: dict) -> OAuthClientMetadata:
+def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
     """Build OAuthClientMetadata; requires ``_configure_callback_port`` first."""
     port = cfg.get("_resolved_port")
     if port is None:
@@ -1175,7 +1175,7 @@ def _invalidate_tokens_on_client_change(
             storage._server_name, old_client_id, new_client_id, storage._server_name)
 
 
-def _maybe_preregister_client(storage: HermesTokenStorage, cfg: dict, client_metadata: OAuthClientMetadata) -> None:
+def _maybe_preregister_client(storage: HermesTokenStorage, cfg: dict, client_metadata: "OAuthClientMetadata") -> None:
     """If cfg has a pre-registered client_id, persist it to storage."""
     client_id = cfg.get("client_id")
     if not client_id:
@@ -1231,7 +1231,7 @@ def humanize_oauth_registration_error(
         "API-key / local server instead.")
 
 
-def build_oauth_auth(server_name: str, server_url: str, oauth_config: dict | None = None) -> OAuthClientProvider | None:
+def build_oauth_auth(server_name: str, server_url: str, oauth_config: dict | None = None) -> "OAuthClientProvider | None":
     """``httpx.Auth`` OAuth handler for an MCP server; None if the SDK lacks OAuth. Legacy API — new code
     uses :func:`tools.mcp_oauth_manager.get_manager` so state is shared across config-time, runtime and reconnect paths."""
     global HermesOAuthClientProvider

tools/mcp_tool_registration.py

@@ -153,7 +153,7 @@ def _remove_server_scope(key, scope: str) -> None:
     _restore_server_toolset_alias(key)
 
 
-def _select_utility_schemas(server_name: str, server: MCPServerTask, config: dict) -> list[dict]:
+def _select_utility_schemas(server_name: str, server: "MCPServerTask", config: dict) -> list[dict]:
     """Utility schemas allowed by config (``tools.resources``/``tools.prompts``) and advertised
     capabilities. ``initialize_result.capabilities`` is the truth (sub-object non-None iff the
     family is served); without it fall back to the legacy session-method check, which never
@@ -367,7 +367,7 @@ def _register_candidates(name: str, candidates: list[_Candidate], *, check_fn: C
     return registered
 
 
-def _write_schema_cache(name: str, server: MCPServerTask, config: dict, should_register) -> None:
+def _write_schema_cache(name: str, server: "MCPServerTask", config: dict, should_register) -> None:
     """Write-through: persist the manifest so the next startup registers this server lazily (no spawn). Never raises."""
     try:
         # Write-through (#56832): refresh the on-disk schema cache after a live connect so the next startup
@@ -397,7 +397,7 @@ def _write_schema_cache(name: str, server: MCPServerTask, config: dict, should_r
         logger.debug("MCP schema cache write failed for '%s': %s", name, exc)
 
 
-def _register_server_tools(name: str, server: MCPServerTask, config: dict) -> list[str]:
+def _register_server_tools(name: str, server: "MCPServerTask", config: dict) -> list[str]:
     """Register a connected server's tools plus utilities (initial discovery and list_changed
     refresh); returns the names. Toolset aliases derive from the live registry, not
     ``toolsets.TOOLSETS``; lossy normalization collisions (``read-file``/``read_file``) fail closed."""

tools/process_registry.py

@@ -516,14 +516,14 @@ def _not_found(session_id: str) -> dict:
     return {"status": "not_found", "error": f"No process with ID {session_id}"}
 
 
-def _output_tail(session: ProcessSession, n: int) -> str:
+def _output_tail(session: "ProcessSession", n: int) -> str:
     """Last *n* chars of the session output with ANSI sequences stripped."""
     from tools.ansi_strip import strip_ansi
 
     return strip_ansi(session.output_buffer[-n:])
 
 
-def _completion_output(session: ProcessSession) -> dict:
+def _completion_output(session: "ProcessSession") -> dict:
     """``output`` sized by the session's ``completion_output_chars`` plus ``output_cut`` when trimmed.
 
     Shared by the completion notification AND the wait/poll/kill snapshots: a bot in an api_server

tools/process_registry_notifications.py

@@ -375,7 +375,7 @@ class TimelineNotification(str):
         return instance
 
     @classmethod
-    def for_delegation(cls, text: str, event: dict) -> TimelineNotification:
+    def for_delegation(cls, text: str, event: dict) -> "TimelineNotification":
         from agent.notification_presentation import diagnostic_process_event
         return cls(text, async_delegation_display_text(event), "async_delegation_complete",
                    "diagnostic" if diagnostic_process_event(event) else "result")

tools/process_registry_termination.py

@@ -146,7 +146,7 @@ class ProcessTerminationMixin:
     # escalated kill as incomplete.
     _KILL_SETTLE_SECONDS = 1.0
 
-    def _post_kill_survivors(self, session: ProcessSession) -> list[int]:
+    def _post_kill_survivors(self, session: "ProcessSession") -> list[int]:
         """Host PIDs still alive once the kill signals have had time to land (#115490).
 
         Fail-closed: anything unverifiable counts as a survivor, so a kill
@@ -160,7 +160,7 @@ class ProcessTerminationMixin:
                 return survivors
             time.sleep(0.05)
 
-    def _probe_survivors(self, session: ProcessSession) -> list[int]:
+    def _probe_survivors(self, session: "ProcessSession") -> list[int]:
         survivors: list[int] = []
         proc = getattr(session, "process", None)
         if proc is not None:

tools/skills_sync.py

@@ -222,7 +222,7 @@ def _copy_dir(src: Path, dest: Path) -> None:
     shutil.copytree(src, dest, ignore=_ignore_runtime_cache)
 
 
-def _recover_renamed_skill(st: _SyncState, skill_name: str, dest: Path) -> Optional[str]:
+def _recover_renamed_skill(st: "_SyncState", skill_name: str, dest: Path) -> Optional[str]:
     """Move a bundled skill's stale copy to its new canonical path after an upstream RENAME /
     RECATEGORIZATION (else it is misread as user-deleted and stranded forever). Only a copy
     byte-identical to the origin hash — proof *we* placed it — moves. Returns rel source path."""

trajectory_compressor.py

@@ -134,7 +134,7 @@ class CompressionConfig:
     metrics_output_file: str = "compression_metrics.json"
 
     @classmethod
-    def from_yaml(cls, yaml_path: str) -> CompressionConfig:
+    def from_yaml(cls, yaml_path: str) -> "CompressionConfig":
         """Load configuration from YAML file (missing keys keep the defaults)."""
         with open(yaml_path, 'r', encoding="utf-8") as f:
             data = yaml.safe_load(f) or {}

tui_gateway/server.py

@@ -1002,7 +1002,7 @@ def _wait_agent_for_prompt(session: dict, rid: str, sid: str) -> dict | None:
     return _err(rid, 5032, err) if (err := session.get("agent_error")) else None
 
 
-def _bind_build_profile_scopes(profile_home: str | None) -> _TurnScopes | None:
+def _bind_build_profile_scopes(profile_home: str | None) -> "_TurnScopes | None":
     """Bind a session profile's HERMES_HOME / secret / terminal scopes for an agent build. ``None`` is the
     launch profile: its own launch-env secret scope (live env while single-profile, frozen once
     multiplexing is active — a hosted-room turn for a default member otherwise died at build with
@@ -1020,7 +1020,7 @@ def _bind_build_profile_scopes(profile_home: str | None) -> _TurnScopes | None:
     return scopes
 
 
-def _release_build_profile_scopes(scopes: _TurnScopes | None) -> None:
+def _release_build_profile_scopes(scopes: "_TurnScopes | None") -> None:
     with contextlib.suppress(Exception):
         _release_profile_runtime_scope_tokens(scopes)
 

tui_gateway/slash_worker.py

@@ -112,7 +112,7 @@ def _refuse_skill_slash(command: str) -> None:
         raise SkillSlashRefused(base)
 
 
-def _run(cli: HermesCLI, command: str) -> str:
+def _run(cli: "HermesCLI", command: str) -> str:
     """Run one command; return its captured, ANSI-stripped output.
 
     A command like /prompt or /blueprint parks the composed text on the one-shot