ci: bring the branch's own code under main's strict ruff config

Main made the strict ruff config the default (`ruff check .`, blocking), and
the branch's code carried 278 findings under it (main itself is clean). Fixed:

- PLW1510 (86): `check=False` on each subprocess.run, preserving behaviour,
  as main's own sweep (0006a5e229) did.
- ASYNC220 (9, test fixtures only): `noqa` with the reason, main's convention
  for a Popen that returns at once and whose handle the test drives.
- ruff's safe autofixes plus the narrow unsafe ones (UP006/UP037/RUF100/RUF013/
  PLR0402/RUF059 underscore renames/UP031/UP041/UP017/FURB167/PIE807/RUF007/
  RUF015/RUF022/FURB162), and by hand: four %-formats to f-strings, three
  unused unpacked names, one fixture loop over `.items()`.

No behaviour change. `ruff check .` passes; scripts/check passes; the 412 test
files mirroring every edited file pass (3 tests/pm files that fail only under
that batch's uv lookup pass solo, before and after this commit).

agent/credential_pool_timestamps.py

@@ -25,7 +25,7 @@ def _parse_absolute_timestamp(value: Any) -> Optional[float]:
         except ValueError:
             pass
         try:
-            return datetime.fromisoformat(raw.replace("Z", "+00:00")).timestamp()
+            return datetime.fromisoformat(raw).timestamp()
         except ValueError:
             return None
     return None

evals/desktop_gateway_ticket_native/run.py

@@ -16,7 +16,7 @@ import psutil
 
 REPO = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(REPO))
-from hermes_cli.gateway_runtime_discovery import query_identify  # noqa: E402
+from hermes_cli.gateway_runtime_discovery import query_identify
 
 logger = logging.getLogger(__name__)
 

evals/desktop_gateway_ticket_native/run_multiplex.py

@@ -16,8 +16,8 @@ import time
 
 REPO = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(REPO))
-from evals.desktop_gateway_ticket_native.run import stop_owned  # noqa: E402
-from hermes_cli.gateway_runtime_discovery import query_identify  # noqa: E402
+from evals.desktop_gateway_ticket_native.run import stop_owned
+from hermes_cli.gateway_runtime_discovery import query_identify
 
 logger = logging.getLogger(__name__)
 

evals/hosted_attachment_transfer/bench.py

@@ -32,7 +32,7 @@ TMP = Path(tempfile.mkdtemp(prefix='hosted-attachment-bench-'))
 os.environ['HOME'] = str(TMP)
 os.environ['HERMES_HOME'] = str(TMP / '.hermes')
 
-from pytest import MonkeyPatch  # noqa: E402
+from pytest import MonkeyPatch
 
 
 class SourceMeter:

evals/shared_gateway/search_readonly_ab.py

@@ -72,7 +72,7 @@ def main():
                        'PYTHONDONTWRITEBYTECODE': '1'}
                 proc = subprocess.run([sys.executable, str(Path(__file__).resolve()), '--child', str(repo),
                                        '--case', case, '--account', tmp], cwd=repo, env=env,
-                                      stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=45)
+                                      stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=45, check=False)
                 if proc.returncode:
                     raise RuntimeError(f'{label}/{case} failed: {proc.stderr}')
                 row = json.loads(proc.stdout.strip().splitlines()[-1])

evals/shared_gateway/session_readers_cli_ab.py

@@ -40,7 +40,7 @@ for (label, tree), (name, command), populated in itertools.product(TREES.items()
         env.update(HOME=tmp, USERPROFILE=tmp, HERMES_HOME=str(home), PYTHONPATH=str(tree))
         def run(args):
             return subprocess.run([PYTHON, *args], cwd=tree, env=env, stdin=subprocess.DEVNULL,
-                                  capture_output=True, text=True, timeout=45)
+                                  capture_output=True, text=True, timeout=45, check=False)
         if populated:
             seed = run(['-c', "from hermes_state import SessionDB; db=SessionDB(); db.create_session('reader-fixture',source='tui'); db.append_message('reader-fixture',role='user',content='fixture'); db.set_session_pinned('reader-fixture',True); db.close()"])
             assert seed.returncode == 0, seed.stderr

gateway/hosted_room_attachments.py

@@ -1266,21 +1266,21 @@ class HostedRoomAttachmentStore:
 
 
 __all__ = [
+    "CLASSIC_ATTACHMENT_TTL_SECONDS",
+    "DISBANDED_GRACE_SECONDS",
+    "MAX_ATTACHMENTS_PER_MESSAGE",
+    "MAX_ATTACHMENT_BYTES",
+    "MAX_MESSAGE_ATTACHMENT_BYTES",
+    "MAX_TASK_ATTACHMENTS",
+    "MAX_TASK_ATTACHMENT_BYTES",
+    "UNCOMMITTED_TTL_SECONDS",
     "AttachmentConflictError",
     "AttachmentData",
     "AttachmentError",
     "AttachmentIntegrityError",
     "AttachmentNotFoundError",
     "AttachmentQuotaError",
-    "CLASSIC_ATTACHMENT_TTL_SECONDS",
-    "DISBANDED_GRACE_SECONDS",
     "HostedRoomAttachmentStore",
-    "MAX_ATTACHMENT_BYTES",
-    "MAX_ATTACHMENTS_PER_MESSAGE",
-    "MAX_MESSAGE_ATTACHMENT_BYTES",
-    "MAX_TASK_ATTACHMENT_BYTES",
-    "MAX_TASK_ATTACHMENTS",
-    "UNCOMMITTED_TTL_SECONDS",
     "decode_content_base64",
     "default_attachment_root",
     "encode_content_base64",

gateway/platforms/msgraph_webhook.py

@@ -25,6 +25,7 @@ from gateway.platforms.base import (
     BasePlatformAdapter, SendResult, is_network_accessible,
 )
 from gateway.platforms.event import MessageEvent, MessageType
+from datetime import UTC
 
 logger = logging.getLogger(__name__)
 
@@ -341,7 +342,7 @@ class MSGraphWebhookAdapter(BasePlatformAdapter):
             source = event.source.to_dict()
             source['is_bot'] = event.source.is_bot
             envelope = {'source': source, 'route': route, 'provenance': provenance,
-                'timestamp': datetime.fromtimestamp(0, timezone.utc).isoformat(),
+                'timestamp': datetime.fromtimestamp(0, UTC).isoformat(),
                 'event': {'message_id': event.message_id},
                 'automation': {'identity': event.message_id, 'owner': ref.session_id}}
             row = admit_session_input(authority.db, epoch=authority.epoch,

gateway/relay/adapter.py

@@ -35,6 +35,7 @@ from gateway.relay.egress import (
 )
 from gateway.relay.media import RelayMediaClient
 from gateway.relay.transport import RelayTransport
+from datetime import UTC
 
 logger = logging.getLogger(__name__)
 
@@ -895,7 +896,7 @@ class RelayAdapter(RelayDiscordInteractionMixin, BasePlatformAdapter):
             from gateway.platforms.webhook_ingress import admit_producer
             # The connector does not transmit a provider timestamp. Local arrival
             # time must not change the fingerprint of a retried provider identity.
-            event.timestamp = datetime.fromtimestamp(0, timezone.utc)
+            event.timestamp = datetime.fromtimestamp(0, UTC)
             # Wait only for the SQLite receipt, never inference or outbound ACKs
             # (those need this same WS reader). Exceptions deliberately suppress ACK.
             if await admit_producer(self, event) is None and getattr(event, '_consumer_reply', None):

gateway/relay/adapter_discord_interaction.py

@@ -27,7 +27,7 @@ class RelayDiscordInteractionMixin:
         session key matches the one the follow-up capability was bound under."""
         try:
             payload = json.loads(bytes(getattr(forward, "body", b"")).decode("utf-8"))
-        except Exception:  # noqa: BLE001
+        except Exception:
             return None
         if not isinstance(payload, dict):
             return None

gateway/run_inbound_hooks.py

@@ -24,8 +24,8 @@ class GatewayInboundHooksMixin:
     """Hook, quick-command and plugin-command dispatch for ``GatewayInboundMixin``."""
 
     async def _hm_post_admission_consume(
-        self, event: "MessageEvent", source: SessionSource, session_key: str
-    ) -> Tuple[bool, Optional[str]]:
+        self, event: MessageEvent, source: SessionSource, session_key: str
+    ) -> tuple[bool, Optional[str]]:
         """``post_gateway_admission`` exactly once per inbound message: ``(consumed, reply)``.
 
         Never while an admitted row EXECUTES (first run or restart replay): the authority path
@@ -38,8 +38,8 @@ class GatewayInboundHooksMixin:
         return await run_post_admission_hook(self, event, source, session_key)
 
     async def _hm_pre_gateway_dispatch_hook(
-        self, event: "MessageEvent", source: SessionSource
-    ) -> Optional["MessageEvent"]:
+        self, event: MessageEvent, source: SessionSource
+    ) -> Optional[MessageEvent]:
         """Run the ``pre_gateway_dispatch`` plugin hook; None = drop, else the (maybe rewritten) event.
         Results: ``{"action": "skip"}`` → drop; ``{"action": "rewrite", "text"}`` → replace ``event.text``;
         ``allow``/None → normal dispatch. Runs BEFORE auth so plugins can handle unauthorized senders."""
@@ -81,7 +81,7 @@ class GatewayInboundHooksMixin:
         return qc if isinstance(qc, dict) else {}
 
     @staticmethod
-    def _hm_expand_alias_quick_command(event: "MessageEvent", qcmd: dict) -> Optional[str]:
+    def _hm_expand_alias_quick_command(event: MessageEvent, qcmd: dict) -> Optional[str]:
         """Rewrite ``event.text`` to an alias quick command's target; returns the new command name."""
         target = (qcmd.get("target") or "").strip()
         if not target:
@@ -92,8 +92,8 @@ class GatewayInboundHooksMixin:
         return target_command.split()[0] if target_command else target_command
 
     async def _hm_command_hooks(
-        self, event: "MessageEvent", source: SessionSource, _quick_key: str, command: str, canonical: str
-    ) -> Tuple[bool, Optional[str], Optional[str]]:
+        self, event: MessageEvent, source: SessionSource, _quick_key: str, command: str, canonical: str
+    ) -> tuple[bool, Optional[str], Optional[str]]:
         """Fire ``pre_command`` (observer) and ``command:<canonical>`` (interceptor) hooks →
         ``(handled, result, new_command)`` (``new_command`` set when a handler rewrote the command).
         The running-agent path deliberately does NOT fire these — a slow or hostile plugin must not
@@ -159,8 +159,8 @@ class GatewayInboundHooksMixin:
             return t("gateway.quick_command.error", error=e)
 
     async def _hm_dispatch_quick_and_plugin_commands(
-        self, event: "MessageEvent", source: SessionSource, command: Optional[str]
-    ) -> Tuple[bool, Optional[str], Optional[str]]:
+        self, event: MessageEvent, source: SessionSource, command: Optional[str]
+    ) -> tuple[bool, Optional[str], Optional[str]]:
         """Drain gate, user-defined quick commands (exec/alias) and plugin slash commands →
         ``(handled, result, command)``; an alias quick command rewrites ``command``."""
         if self._draining:

gateway/run_notifications_update.py

@@ -89,7 +89,7 @@ class GatewayUpdateNotificationsMixin:
             logger.debug("Skipping update notification watcher: no running event loop")
 
     @classmethod
-    def _update_paths(cls) -> "GatewayUpdateNotificationsMixin._UpdatePaths":
+    def _update_paths(cls) -> GatewayUpdateNotificationsMixin._UpdatePaths:
         from gateway.run import _hermes_home
         return cls._UpdatePaths(
             pending=_hermes_home / ".update_pending.json",
@@ -130,7 +130,7 @@ class GatewayUpdateNotificationsMixin:
         now = datetime.now(stamped.tzinfo) if stamped.tzinfo else datetime.now()
         return (now - stamped).total_seconds()
 
-    def _resolve_update_target(self, paths: "_UpdatePaths") -> Optional["_UpdateTarget"]:
+    def _resolve_update_target(self, paths: _UpdatePaths) -> Optional[_UpdateTarget]:
         """Resolve adapter/chat/session for update watcher messages from the pending marker."""
         for path in (paths.claimed, paths.pending):
             if not path.exists():
@@ -162,7 +162,7 @@ class GatewayUpdateNotificationsMixin:
             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:
+    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")
         # Poll until _send_update_notification delivers (it returns False while the platform reconnects).
@@ -176,7 +176,7 @@ class GatewayUpdateNotificationsMixin:
             await self._send_update_notification()
 
     @staticmethod
-    def _update_exit_code(paths: "_UpdatePaths") -> int:
+    def _update_exit_code(paths: _UpdatePaths) -> int:
         return int(paths.exit_code.read_text(encoding="utf-8-sig").strip() or "1")
 
     @staticmethod
@@ -190,7 +190,7 @@ class GatewayUpdateNotificationsMixin:
             return "", len(data)
         return data[offset:].decode("utf-8", errors="replace"), len(data)
 
-    async def _send_update_output(self, target: "_UpdateTarget", text: str) -> None:
+    async def _send_update_output(self, target: _UpdateTarget, text: str) -> None:
         """Send buffered update output as fenced chunks that fit message limits (Telegram: 4096)."""
         from tools.ansi_strip import strip_ansi
         clean = strip_ansi(text).strip()
@@ -201,7 +201,7 @@ class GatewayUpdateNotificationsMixin:
             with _log_suppressed(logging.DEBUG, "Update stream send failed: %s"):
                 await target.send(f"```\n{clean[i:i + max_chunk]}\n```")
 
-    async def _forward_update_prompt(self, target: "_UpdateTarget", prompt_text: str, default: str) -> None:
+    async def _forward_update_prompt(self, target: _UpdateTarget, prompt_text: str, default: str) -> None:
         """Forward an update prompt: platform-native buttons first (Discord, Telegram), else text."""
         sent_buttons = False
         adapter = target.adapter
@@ -220,7 +220,7 @@ class GatewayUpdateNotificationsMixin:
         self._session_state(target.session_key).persistent.update_prompt_pending = True
         logger.info("Forwarded update prompt to %s: %s", target.session_key, prompt_text[:80])
 
-    def _clear_update_markers(self, paths: "_UpdatePaths", session_key: Optional[str]) -> None:
+    def _clear_update_markers(self, paths: _UpdatePaths, session_key: Optional[str]) -> None:
         paths.unlink_all()
         state = self._peek_session_state(session_key)
         if state is not None:

gateway/run_runtime_init.py

@@ -115,13 +115,13 @@ class GatewayRuntimeInitMixin:
         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: 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_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.
@@ -137,7 +137,7 @@ class GatewayRuntimeInitMixin:
         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: 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
@@ -212,7 +212,7 @@ class GatewayRuntimeInitMixin:
         from gateway.pairing import PairingStore
         from gateway.hooks import ProfileHookRegistries
         self.pairing_store = PairingStore()
-        self.pairing_stores: Dict[str, "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"

gateway/run_turn_hygiene.py

@@ -32,8 +32,8 @@ from typing import Any, Callable, Dict, List, Optional, Tuple
 from utils import base_url_hostname
 
 if TYPE_CHECKING:  # string annotations only; never imported at runtime (cycle)
-    from gateway.run import GatewayRunner  # noqa: F401
-    from gateway.run_turn_runner import TurnRunner  # noqa: F401
+    from gateway.run import GatewayRunner
+    from gateway.run_turn_runner import TurnRunner
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("gateway.run")
@@ -237,7 +237,7 @@ class GatewayTurnHygieneMixin:
         fence = attempt.commit_fence
         while True:
             if fence.is_cancelled:
-                raise asyncio.TimeoutError
+                raise TimeoutError
             # Charge the idle budget from the LAST PROGRESS event, else silence can approach 2x timeout.
             _hyg_waited = time.monotonic() - attempt.wait_started
             _slice = min(
@@ -254,7 +254,7 @@ class GatewayTurnHygieneMixin:
             try:
                 _compressed, _ = await asyncio.wait_for(asyncio.shield(attempt.future), timeout=_slice)
                 return _compressed
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 if fence.is_cancelled:
                     raise
                 _hyg_waited = time.monotonic() - attempt.wait_started
@@ -746,7 +746,7 @@ class GatewayTurnHygieneMixin:
                 _compressed = await self._hmwa_hygiene_wait_for_summary(attempt, hs, session_entry)
             except HygieneTurnHoldExceeded:
                 _compressed = await self._hmwa_hygiene_on_turn_hold(attempt, hs, session_entry, session_key, source)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 _compressed = await self._hmwa_hygiene_on_timeout(attempt, hs, session_entry, session_key, source)
             except BaseException:
                 self._hmwa_hygiene_on_unwind(attempt, hs, session_entry, session_key)

gateway/run_turn_persistence.py

@@ -34,8 +34,8 @@ from typing import Any, Callable, Dict, List, Optional, Tuple
 from utils import base_url_hostname
 
 if TYPE_CHECKING:  # string annotations only; never imported at runtime (cycle)
-    from gateway.run import GatewayRunner  # noqa: F401
-    from gateway.run_turn_runner import TurnRunner  # noqa: F401
+    from gateway.run import GatewayRunner
+    from gateway.run_turn_runner import TurnRunner
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("gateway.run")

gateway/run_turn_progress.py

@@ -24,8 +24,8 @@ from hermes_cli.config import cfg_get
 from utils import is_truthy_value
 
 if TYPE_CHECKING:  # string annotations only; never imported at runtime (cycle)
-    from gateway.run import GatewayRunner  # noqa: F401
-    from gateway.run_turn_runner import TurnRunner  # noqa: F401
+    from gateway.run import GatewayRunner
+    from gateway.run_turn_runner import TurnRunner
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("gateway.run")
@@ -125,7 +125,7 @@ class GatewayTurnProgressMixin:
 
     # ── progress_callback (agent thread → progress queue) ───────────────────────────────────
 
-    def progress_callback(self, event_type: str, tool_name: str = None, preview: str = None, args: dict = None, **kwargs):
+    def progress_callback(self, event_type: str, tool_name: str | None = None, preview: str | None = None, args: dict | None = None, **kwargs):
         """Callback invoked by agent on tool lifecycle events."""
         ctx = self._ctx
         # Failed subagent → one clean user-facing notice, handled FIRST, before every progress-queue
@@ -329,8 +329,8 @@ class GatewayTurnProgressMixin:
     class _TaskCardState:
         """Task-card rail state for ``_send_native_task_card_progress``."""
         adapter: Any
-        tasks: Dict[str, Dict[str, str]] = dataclasses.field(default_factory=dict)
-        task_order: List[str] = dataclasses.field(default_factory=list)
+        tasks: dict[str, dict[str, str]] = dataclasses.field(default_factory=dict)
+        task_order: list[str] = dataclasses.field(default_factory=list)
         fallback_msg_id: Optional[str] = None
         native_failed: bool = False
         # TERMINAL for the turn, distinct from native_failed: no later publication
@@ -347,7 +347,7 @@ class GatewayTurnProgressMixin:
             text = re.sub(r"\s+", " ", str(value or "")).strip()
             return text if len(text) <= limit else text[: limit - 3].rstrip() + "..."
 
-        def visible_tasks(self) -> List[Dict[str, str]]:
+        def visible_tasks(self) -> list[dict[str, str]]:
             return [self.tasks[task_id] for task_id in self.task_order[-8:]]
 
         def fallback_text(self) -> str:
@@ -358,7 +358,7 @@ class GatewayTurnProgressMixin:
                      for task in self.visible_tasks()]
             return t("gateway.progress.task_card_title") + "\n" + "\n".join(lines)
 
-        def _upsert(self, call_id: str, title: str) -> Dict[str, str]:
+        def _upsert(self, call_id: str, title: str) -> dict[str, str]:
             if call_id not in self.tasks:
                 self.task_order.append(call_id)
             self.tasks[call_id] = {"id": call_id, "title": self._compact(title), "status": "in_progress"}
@@ -548,7 +548,7 @@ class GatewayTurnProgressMixin:
         _PROGRESS_TEXT_LIMIT: int
         _edit_accepts_metadata: bool
 
-    def _progress_edit_state(self, adapter) -> "TurnRunner._ProgressEditState":
+    def _progress_edit_state(self, adapter) -> TurnRunner._ProgressEditState:
         ctx = self._ctx
         len_fn = adapter.message_len_fn if isinstance(adapter, BasePlatformAdapter) else len
         try:

gateway/runtime_downgrade_exit.py

@@ -40,9 +40,9 @@ def downgrade_reason(skew, *, update_in_progress: bool, predates: Callable[[], b
     """The log line when this gateway must stop for a downgraded tree, else None."""
     if not skew or update_in_progress or not predates():
         return None
-    return ("This gateway loaded %s, but its checkout is now at %s, a release that predates the gateway "
+    return ("This gateway loaded {}, but its checkout is now at {}, a release that predates the gateway "
             "runtime. Downgrading is unsupported; stopping (no restart) so the older release can run its "
-            "own gateway." % tuple(skew))
+            "own gateway.".format(*tuple(skew)))
 
 
 async def _stop_or_exit(runner, *, hard_exit: Callable[[int], None] = os._exit) -> None:

gateway/session.py

@@ -21,7 +21,7 @@ from gateway.session_recovery import SessionRecoveryMixin
 from gateway.session_lifecycle import SessionLifecycleMixin, _iso, _new_session_id, _now, _parse_iso
 from gateway.session_transcript import SessionTranscriptMixin
 # Inverse of _session_key_namespace, defined storage-side (hermes_state must not import gateway).
-from hermes_state_keys import profile_from_session_key_namespace  # noqa: F401
+from hermes_state_keys import profile_from_session_key_namespace
 
 logger = logging.getLogger(__name__)
 

gateway/session_execution.py

@@ -23,7 +23,7 @@ from hermes_cli.config import cfg_get
 from utils import is_truthy_value
 
 if TYPE_CHECKING:  # string annotations only; never imported at runtime (cycle)
-    from gateway.run import GatewayRunner  # noqa: F401
+    from gateway.run import GatewayRunner
 
 # Log-record parity with the origin module.
 logger = logging.getLogger("gateway.run")

gateway/session_shell.py

@@ -64,7 +64,7 @@ def _run(command, cwd):
     try:
         result = subprocess.run(command, cwd=cwd, shell=True, env=env, capture_output=True,
                                 text=True, encoding='utf-8', errors='replace', timeout=TIMEOUT_SECONDS,
-                                stdin=subprocess.DEVNULL, creationflags=windows_hide_flags())
+                                stdin=subprocess.DEVNULL, creationflags=windows_hide_flags(), check=False)
     except subprocess.TimeoutExpired:
         raise RuntimeStoreError('command_timeout') from None
     except OSError:

hermes_cli/config_show.py

@@ -45,7 +45,7 @@ _SHOW_CONFIG_API_KEYS = (
     ("FAL_KEY", "FAL"))
 
 
-def _show_model_section(config: Dict[str, Any]) -> None:
+def _show_model_section(config: dict[str, Any]) -> None:
     _section("Model")
     print(f"  Model:        {_cfg.redact_config_value(config.get('model', 'not set'))}")
     cfg_max_turns = config.get('agent', {}).get('max_turns', _cfg.DEFAULT_CONFIG['agent']['max_turns'])
@@ -61,7 +61,7 @@ def _show_model_section(config: Dict[str, Any]) -> None:
                     f"(run 'hermes doctor --fix' to remove)", Colors.YELLOW))
 
 
-def _show_display_section(config: Dict[str, Any]) -> None:
+def _show_display_section(config: dict[str, Any]) -> None:
     _section("Display")
     display = config.get('display', {})
     try:
@@ -69,7 +69,7 @@ def _show_display_section(config: Dict[str, Any]) -> None:
         active_personality = active_personality_name(config) or 'none'
     except Exception:
         active_personality = display.get('personality') or 'none'
-    on_off = lambda flag: 'on' if flag else 'off'  # noqa: E731
+    on_off = lambda flag: 'on' if flag else 'off'
     print(f"  Personality:  {active_personality}")
     print(f"  Reasoning:    {on_off(display.get('show_reasoning', True))}")
     print(
@@ -80,14 +80,14 @@ def _show_display_section(config: Dict[str, Any]) -> None:
     print(f"  User preview: first {ump.get('first_lines', 2)} line(s), last {ump.get('last_lines', 2)} line(s)")
 
 
-def _show_terminal_section(config: Dict[str, Any]) -> None:
+def _show_terminal_section(config: dict[str, Any]) -> None:
     _section("Terminal")
     terminal = config.get('terminal', {})
     print(f"  Backend:      {terminal.get('backend', 'local')}")
     print(f"  Working dir:  {terminal.get('cwd', '.')}")
     print(f"  Timeout:      {terminal.get('timeout', 60)}s")
 
-    configured = lambda *names: 'configured' if all(_cfg.get_env_value(n) for n in names) else '(not set)'  # noqa: E731
+    configured = lambda *names: 'configured' if all(_cfg.get_env_value(n) for n in names) else '(not set)'
     from hermes_cli.config_defaults import DEFAULT_SANDBOX_IMAGE as default_img, DEFAULT_VERCEL_IMAGE as _DEFAULT_VERCEL_IMAGE
     backend_lines = {
         'docker': lambda: [f"  Docker image: {terminal.get('docker_image', default_img)}"],
@@ -109,7 +109,7 @@ def _show_terminal_section(config: Dict[str, Any]) -> None:
         print(line)
 
 
-def _show_compression_section(config: Dict[str, Any]) -> None:
+def _show_compression_section(config: dict[str, Any]) -> None:
     _section("Context Compression")
     compression = config.get('compression', {})
     enabled = compression.get('enabled', True)
@@ -133,7 +133,7 @@ def _show_compression_section(config: Dict[str, Any]) -> None:
         print(f"  Provider:     {comp_provider}")
 
 
-def _show_aux_overrides(config: Dict[str, Any]) -> None:
+def _show_aux_overrides(config: dict[str, Any]) -> None:
     aux_tasks = {"Vision": config.get('auxiliary', {}).get('vision', {})}
     overrides = {
         label: (t.get('provider', 'auto'), t.get('model', ''))

hermes_cli/gateway_runtime_service.py

@@ -68,7 +68,7 @@ class ExistingService:
 def _run(argv: list[str] | tuple[str, ...], deadline: float, *, encoding: str | None = "utf-8"):
     return subprocess.run(argv, stdin=subprocess.DEVNULL, capture_output=True,
                           text=encoding is not None, encoding=encoding, errors="strict" if encoding else None,
-                          creationflags=windows_hide_flags(), timeout=remaining(deadline))
+                          creationflags=windows_hide_flags(), timeout=remaining(deadline), check=False)
 
 
 def _verify_binding(verify, *args, **kwargs):

hermes_cli/gateway_runtime_service_identity.py

@@ -172,13 +172,13 @@ def verify_launchd_loaded(output: str, home: Path, *, uid: int, username: str) -
     # Only the job's own blocks, never inherited/default environments or a
     # different job embedded in launchctl diagnostic text.
     def block(name):
-        found = re.findall(r"^\t" + name + r" = \{\n(.*?)^\t\}", output, re.M | re.S)
+        found = re.findall(r"^\t" + name + r" = \{\n(.*?)^\t\}", output, re.MULTILINE | re.DOTALL)
         if len(found) != 1:
             _unverified()
         return found[0].splitlines()
 
     for key, expected in (("uid", str(uid)), ("username", username)):
-        values = re.findall(r"^\t" + key + r" = (.+)$", output, re.M)
+        values = re.findall(r"^\t" + key + r" = (.+)$", output, re.MULTILINE)
         if values and values != [expected]:
             raise ValueError("service_account_mismatch")
     env = {}
@@ -193,7 +193,7 @@ def verify_launchd_loaded(output: str, home: Path, *, uid: int, username: str) -
     if home.parent.name != "profiles" and not env.get("HERMES_SUPERVISED_CHILD"):
         _unverified()
     argv = [line.strip() for line in block("arguments")]
-    programs = re.findall(r"^\tprogram = (.+)$", output, re.M)
+    programs = re.findall(r"^\tprogram = (.+)$", output, re.MULTILINE)
     if len(programs) != 1 or not argv or programs[0] != argv[0]:
         _unverified()
     verify_gateway_argv(argv, home)

hermes_cli/kanban_worker_client.py

@@ -2,7 +2,7 @@
 # Spawned as a bare ``sys.executable -m``: in a PM install that interpreter carries no dependencies
 # until hermes_bootstrap selects the committed environment, so it must be the first import.
 try:
-    import hermes_bootstrap  # noqa: F401
+    import hermes_bootstrap
 except ModuleNotFoundError as exc:
     if exc.name != "hermes_bootstrap":
         raise

hermes_cli/runtime_provider.py

@@ -39,11 +39,11 @@ from utils import base_url_host_matches, base_url_hostname, base_url_path, env_i
 
 
 # A frozen session route's config (``frozen_runtime_config``); None = the live profile config.
-_FROZEN_RUNTIME_CONFIG: ContextVar[Optional[Dict[str, Any]]] = ContextVar("_FROZEN_RUNTIME_CONFIG", default=None)
+_FROZEN_RUNTIME_CONFIG: ContextVar[Optional[dict[str, Any]]] = ContextVar("_FROZEN_RUNTIME_CONFIG", default=None)
 
 
 @contextmanager
-def frozen_runtime_config(config: Dict[str, Any]):
+def frozen_runtime_config(config: dict[str, Any]):
     """Resolve against a session's frozen config instead of the live config.yaml.
 
     Every rung reads its endpoint through ``load_config``/``_get_model_config``, so a frozen route's

hermes_cli/web_server_app.py

@@ -74,7 +74,7 @@ async def standalone_lifespan(app: "FastAPI"):
     # contended state.db migration, so keep it off the pre-yield path: Group
     # Chat must degrade on its own rather than block every Desktop feature.
     from tui_gateway import methods_groups as _hosted_groups
-    import tui_gateway.server  # noqa: F401
+    import tui_gateway.server
 
     try:
         tui_gateway.server.install_tui_message_injector()
@@ -103,8 +103,8 @@ async def standalone_lifespan(app: "FastAPI"):
     # gateway running the scheduler. Server `hermes dashboard` is unaffected —
     # it relies on its own gateway.
     from hermes_cli.process_identity import is_desktop_owned_backend
-    cron_stop: "threading.Event | None" = None
-    cron_thread: "threading.Thread | None" = None
+    cron_stop: threading.Event | None = None
+    cron_thread: threading.Thread | None = None
     desktop_owned = is_desktop_owned_backend()
     if desktop_owned:
         # Reap an orphaned gateway from an abnormal previous exit (reparented to
@@ -147,7 +147,7 @@ async def standalone_lifespan(app: "FastAPI"):
             from hermes_cli.local_runtime.bootstrap import ensure_local_runtime
 
             ensure_local_runtime(load_config())
-        except Exception as exc:  # noqa: BLE001
+        except Exception as exc:
             logging.getLogger(__name__).warning("local runtime boot failed: %s", exc)
 
     threading.Thread(target=_boot_local_runtime, daemon=True, name="local-runtime-boot").start()

hermes_state_compression.py

@@ -278,9 +278,9 @@ class SessionCompressionMixin:
 
     def _publish_compression_child_on_conn(
             self, conn, *, parent_session_id: str, child_session_id: str, source: str,
-            messages: List[Dict[str, Any]], model: str = None, model_config: Dict[str, Any] = None,
-            system_prompt: str = None, cwd: str = None, profile_name: str = None,
-            compression_lock_holder: str = None, require_compression_lease: bool = True,
+            messages: list[dict[str, Any]], model: str | None = None, model_config: dict[str, Any] | None = None,
+            system_prompt: str | None = None, cwd: str | None = None, profile_name: str | None = None,
+            compression_lock_holder: str | None = None, require_compression_lease: bool = True,
             require_lease_refresh: bool = False, lease_ttl_seconds: float = 300.0,
             watermark: Optional[int] = None, watermark_ceiling: Optional[int] = None) -> None:
         """:meth:`publish_compression_child` on the caller's write transaction (never commits). The

hermes_state_local_lineage.py

@@ -8,6 +8,7 @@ import time
 
 from hermes_state_local import POLICY_PREFIX
 from hermes_state_runtime import RuntimeStoreError, _epoch, _json
+import itertools
 
 
 def validate_local_lineage(conn, receipt):
@@ -22,7 +23,7 @@ def validate_local_lineage(conn, receipt):
     if (not isinstance(lineage, list) or not lineage or len(set(lineage)) != len(lineage)
             or lineage[0] != root or lineage[-1] != receipt['entry']['session_id']):
         raise RuntimeStoreError('storage_unavailable')
-    for parent_id, child_id in zip(lineage, lineage[1:]):
+    for parent_id, child_id in itertools.pairwise(lineage):
         parent = conn.execute('SELECT end_reason FROM sessions WHERE id=?', (parent_id,)).fetchone()
         child = conn.execute('SELECT parent_session_id,model_config FROM sessions WHERE id=?', (child_id,)).fetchone()
         if parent is None or child is None or child['parent_session_id'] != parent_id:

hermes_state_messages.py

@@ -1023,12 +1023,12 @@ class SessionMessagesMixin:
             carried_messages=carried_messages, covered_ids=covered_ids, unresolved_held=unresolved_held), compacted_messages)
 
     def _archive_and_compact_on_conn(
-            self, conn, session_id: str, compacted_messages: List[Dict[str, Any]], *,
-            model_config_patch: Optional[Dict[str, Any]] = None, watermark: Optional[int] = None,
+            self, conn, session_id: str, compacted_messages: list[dict[str, Any]], *,
+            model_config_patch: Optional[dict[str, Any]] = None, watermark: Optional[int] = None,
             lock_holder: Optional[str] = None, tail_count: int = 0,
-            carried_messages: Optional[List[Dict[str, Any]]] = None,
-            covered_ids: Optional[List[int]] = None,
-            unresolved_held: Optional[List[Dict[str, Any]]] = None) -> int:
+            carried_messages: Optional[list[dict[str, Any]]] = None,
+            covered_ids: Optional[list[int]] = None,
+            unresolved_held: Optional[list[dict[str, Any]]] = None) -> int:
         """:meth:`archive_and_compact` on the caller's write transaction (never commits). The owner
         and the worker/canonical-mutation verbs share this one body so they cannot drift."""
         from hermes_state import SessionCompressionInProgressError

hermes_state_usage.py

@@ -297,7 +297,7 @@ class SessionUsageMixin:
         self._execute_write(lambda conn: self._update_token_counts_in_transaction(conn, session_id, **values))
 
     def _update_token_counts_in_transaction(
-        self, conn, session_id: str, input_tokens: int=0, output_tokens: int=0, model: str=None, cache_read_tokens: int=0,
+        self, conn, session_id: str, input_tokens: int=0, output_tokens: int=0, model: str | None=None, cache_read_tokens: int=0,
         cache_write_tokens: int=0, reasoning_tokens: int=0, estimated_cost_usd: Optional[float]=None,
         actual_cost_usd: Optional[float]=None, cost_status: Optional[str]=None, cost_source: Optional[str]=None,
         pricing_version: Optional[str]=None, billing_provider: Optional[str]=None, billing_base_url: Optional[str]=None,

hermes_state_worker_compression.py

@@ -209,7 +209,7 @@ def archive_on_connection(db, conn, session_id: str, compacted_messages: list[di
         tail_count=tail_count, carried_messages=carried_messages, covered_ids=covered_ids,
         unresolved_held=unresolved_held)
 
-def publish_on_connection(db, conn, *, parent_session_id: str, child_session_id: str, source: str, messages: list[dict[str, Any]], model: str=None, model_config: dict[str, Any]=None, system_prompt: str=None, cwd: str=None, profile_name: str=None, compression_lock_holder: str=None, require_compression_lease: bool=True, require_lease_refresh: bool=False, lease_ttl_seconds: float=300.0, watermark: Optional[int]=None, watermark_ceiling: Optional[int]=None):
+def publish_on_connection(db, conn, *, parent_session_id: str, child_session_id: str, source: str, messages: list[dict[str, Any]], model: str | None=None, model_config: dict[str, Any] | None=None, system_prompt: str | None=None, cwd: str | None=None, profile_name: str | None=None, compression_lock_holder: str | None=None, require_compression_lease: bool=True, require_lease_refresh: bool=False, lease_ttl_seconds: float=300.0, watermark: Optional[int]=None, watermark_ceiling: Optional[int]=None):
     # The owner's rotation body (lease, closure, child row, timestamps, tail clone, local target,
     # auto-archive lift) on this receipt's connection, so the two paths cannot drift.
     db._publish_compression_child_on_conn(conn, parent_session_id=parent_session_id,

plugins/platforms/a2a/adapter.py

@@ -275,7 +275,7 @@ class A2AAdapter(BasePlatformAdapter):
         # Request ownership outlives reply Futures and also covers synchronous profile forwards.
         self._active_tasks: set[str] = set()
         # Admission ids this process already forwarded; a resend of one is a retry, not a new turn.
-        self._forwarded_inputs: "OrderedDict[str, None]" = OrderedDict()
+        self._forwarded_inputs: OrderedDict[str, None] = OrderedDict()
         self._pending_lock = threading.Lock()
 
     @property

plugins/platforms/discord/adapter.py

@@ -1200,7 +1200,7 @@ class VoiceReceiver:
         )
 
 
-from plugins.platforms.discord.adapter_config_readers import (  # noqa: F401 - re-exported for callers/tests
+from plugins.platforms.discord.adapter_config_readers import (
     _DISCORD_PROMPT_TIMEOUT_DEFAULT, _DISCORD_PROMPT_TIMEOUT_MIN, _read_discord_prompt_timeout, _read_dm_role_auth_guild)
 
 

scripts/probe_active_session_exclusivity.py

@@ -160,7 +160,7 @@ def main() -> int:
         refused_b = subprocess.run(
             [str(PYTHON), "-u", "-m", "tui_gateway.entry"], cwd=str(REPO),
             env={**os.environ, "HERMES_HOME": str(home)}, stdin=subprocess.DEVNULL,
-            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=180)
+            capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=180, check=False)
         check("B cannot start a second writer on A's home", refused_b.returncode == 1
               and "gateway.ready" not in refused_b.stdout, (refused_b.stderr or "")[-200:])
 

tests/acp_adapter/test_gateway_acp_session_end.py

@@ -51,7 +51,7 @@ async def test_last_viewer_or_shutdown_ends_only_idle_acp_sessions(tmp_path, mon
         acp, acp_row = create('acp', 'editor')
         gui, gui_row = create('gui', 'desktop')
         busy, busy_row = create('acp', 'busy-editor')
-        unseen, unseen_row = create('acp', 'unseen-editor')
+        _unseen, unseen_row = create('acp', 'unseen-editor')
         first, second = viewer(), viewer()
         for connection in (first, second):
             await resume(connection, acp)

tests/acp_adapter/test_gateway_client_boundary.py

@@ -5,7 +5,7 @@ from acp_adapter.server import HermesACPAgent
 
 
 def test_default_acp_does_not_construct_an_embedded_session_manager(monkeypatch):
-    import acp_adapter.server as server
+    from acp_adapter import server
 
     def forbidden(*args, **kwargs):
         raise AssertionError("ACP constructed an independent session owner")

tests/acp_adapter/test_gateway_editor_mcp.py

@@ -149,7 +149,7 @@ async def test_native_cold_editor_rebind_refuses_changed_secret(daemon, specs, t
     os.kill(owned.pid, signal.SIGINT)
     await asyncio.to_thread(owned.wait, 20)
     with (tmp_path / 'cold.log').open('w') as log:
-        process = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,
+        process = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
             stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT)
         try:
             deadline = time.monotonic() + 40

tests/acp_adapter/test_gateway_new_cold.py

@@ -58,7 +58,7 @@ async def test_cold_acp_load_retains_policy_and_executes_after_daemon_restart(da
     config['platform_toolsets'] = {'acp': ['terminal']}
     (home / 'config.yaml').write_text(json.dumps(config))
     with (tmp_path / 'restart.log').open('w') as log:
-        process = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,
+        process = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
             stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT)
         try:
             deadline = time.monotonic() + 40

tests/acp_adapter/test_gateway_sessions.py

@@ -61,7 +61,7 @@ async def editor(daemon, tmp_path):
             process.stdin.close()
             try:
                 await asyncio.wait_for(process.wait(), 5)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 process.kill()
                 await process.wait()
             await peer.reader

tests/agent/test_runtime_session_store_proxy.py

@@ -28,8 +28,8 @@ def _refusing_proxy():
 def test_worker_rpc_reaches_loopback_owner_with_proxy_env(monkeypatch, tmp_path):
     from websockets.sync.server import serve
     from agent import runtime_session_store as store
-    import hermes_cli.gateway_client as gateway_client
-    import hermes_cli.gateway_runtime as gateway_runtime
+    from hermes_cli import gateway_client
+    from hermes_cli import gateway_runtime
     import hermes_cli.gateway_runtime_discovery as discovery
 
     def owner(ws):
@@ -39,7 +39,7 @@ def test_worker_rpc_reaches_loopback_owner_with_proxy_env(monkeypatch, tmp_path)
     proxy, seen = _refusing_proxy()
     with serve(owner, '127.0.0.1', 0, subprotocols=['hermes-gateway-v1']) as server:
         threading.Thread(target=server.serve_forever, daemon=True).start()
-        origin = 'http://127.0.0.1:%d' % server.socket.getsockname()[1]
+        origin = f'http://127.0.0.1:{server.socket.getsockname()[1]}'
         endpoint = SimpleNamespace(api_origin=origin, control_home=None)
         monkeypatch.setattr(gateway_runtime, 'discover_gateway_endpoint',
                             lambda home, timeout: SimpleNamespace(state='ready', endpoint=endpoint))
@@ -48,7 +48,7 @@ def test_worker_rpc_reaches_loopback_owner_with_proxy_env(monkeypatch, tmp_path)
         for name in ('NO_PROXY', 'no_proxy', 'HTTP_PROXY', 'http_proxy', 'HTTPS_PROXY', 'https_proxy',
                      'ALL_PROXY', 'all_proxy'):
             monkeypatch.delenv(name, raising=False)
-        monkeypatch.setenv('HTTPS_PROXY', 'http://127.0.0.1:%d' % proxy.getsockname()[1])
+        monkeypatch.setenv('HTTPS_PROXY', f'http://127.0.0.1:{proxy.getsockname()[1]}')
         try:
             assert store.WorkerRPC(tmp_path)('worker.persist') == {'method': 'worker.persist'}
         finally:

tests/agent/test_safe_worker_policy.py

@@ -22,7 +22,7 @@ def run_worker(tmp_path, body, mode="safe"):
     result = subprocess.run(
         [sys.executable, str(script)], env=env,
         cwd=root, stdin=subprocess.DEVNULL,
-        capture_output=True, text=True, timeout=45,
+        capture_output=True, text=True, timeout=45, check=False
     )
     assert result.returncode == 0, result.stdout + result.stderr
     return json.loads(result.stdout.strip().splitlines()[-1])

tests/e2e/core/dashboard/_helpers.py

@@ -175,7 +175,7 @@ class Sandbox:
                 with contextlib.suppress(Exception):
                     subprocess.run([sys.executable, "-m", "hermes_cli.main", "gateway", "stop"],
                                    env=self.env({"HERMES_HOME": str(home)}), cwd=str(self.home),
-                                   capture_output=True, timeout=60)
+                                   capture_output=True, timeout=60, check=False)
 
 
 class SandboxLeak(AssertionError):

tests/e2e/core/delivery/_cron_clock.py

@@ -613,7 +613,7 @@ def drain_deliveries() -> int:
     """The live gateway's housekeeping drain of the durable delivery queue. No live adapters
     exist in the soak, so every send takes the standalone wire (``FakeSink``). Returns how many
     queued sends were performed."""
-    import cron.scheduler as scheduler
+    from cron import scheduler
 
     total = 0
     while True:
@@ -623,7 +623,7 @@ def drain_deliveries() -> int:
         total += n
 
 
-def delivery_status(home: Path) -> Dict[str, str]:
+def delivery_status(home: Path) -> dict[str, str]:
     """execution id -> durable queue status (``pending``/``delivering``/``delivered``/``failed``/
     ``unknown``/``suppressed``), including tombstoned rows."""
     path = Path(home) / "cron" / "deliveries.db"
@@ -639,7 +639,7 @@ def delivery_status(home: Path) -> Dict[str, str]:
         conn.close()
 
 
-def effective_outcome(row: dict, deliveries: Dict[str, str]) -> Optional[str]:
+def effective_outcome(row: dict, deliveries: dict[str, str]) -> Optional[str]:
     """The ledger's delivery outcome, resolved through the durable queue when the run only
     booked the handoff (``queued``): the queue's verdict is the delivery's truth."""
     outcome = row["delivery_outcome"]

tests/e2e/core/history/_helpers.py

@@ -401,7 +401,7 @@ class TuiGateway:
         # the owner first so each "gw" hop is still a fresh process on the same durable store.
         subprocess.run([sys.executable, "-m", "hermes_cli.main", "gateway", "stop"], cwd=self.cwd,
                        env={**self.env, "PWD": self.cwd}, stdin=subprocess.DEVNULL,
-                       capture_output=True, text=True, encoding="utf-8", timeout=120)
+                       capture_output=True, text=True, encoding="utf-8", timeout=120, check=False)
         self.proc = self.spawned.add(subprocess.Popen(
             [sys.executable, "-m", "tui_gateway.entry"], cwd=self.cwd, env={**self.env, "PWD": self.cwd},
             stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

tests/e2e/core/parity/_helpers.py

@@ -472,7 +472,7 @@ def stop_profile_gateway(ph: ParityHome, timeout: float = STOP_TIMEOUT) -> bool:
     """
     proc = subprocess.run(
         hermes_argv("gateway", "stop"), cwd=ph.project, env=ph.env(), capture_output=True, text=True,
-        timeout=timeout, stdin=subprocess.DEVNULL,
+        timeout=timeout, stdin=subprocess.DEVNULL, check=False
     )
     assert proc.returncode == 0, f"hermes gateway stop exited {proc.returncode}: {proc.stderr[-2000:]}"
     return "Stopped gateway" in proc.stdout

tests/gateway/fixtures/automation_peer.py

@@ -124,7 +124,7 @@ async def probe(root, base, mode):
                 GATEWAY_RELAY_PLATFORMS='telegram', GATEWAY_RELAY_BOT_IDS='{"telegram":"fixture-bot"}',
                 GATEWAY_RELAY_ID='fixture-gateway', GATEWAY_RELAY_SECRET='fixture-secret')
             with (home / 'daemon.log').open('w+') as log:
-                proc = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,
+                proc = subprocess.Popen([sys.executable, '-m', 'gateway.run'], cwd=root, env=env,  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
                     stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT)
                 try:
                     ws = await asyncio.wait_for(ready.get(), 35)

tests/gateway/fixtures/cli_launch_policy_probe.py

@@ -106,7 +106,7 @@ def probe(tmp_path):
                         base_url=origin + '/' + side + '/v1', reasoning='high' if side == 'left' else 'low',
                         max_turns=1 if side == 'left' else 3, ignore_rules=side == 'left', toolsets='terminal')
         result = subprocess.run([sys.executable, '-c', 'import cli; cli.main(**' + repr(args) + ')'],
-            cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=70)
+            cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=70, check=False)
         # A budget run stops at --max-turns with the work unfinished: `chat -Q` exits 1 (documented
         # contract, same as the in-process CLI), never 0.
         assert result.returncode == (1 if budget else 0), result.stdout + result.stderr
@@ -156,11 +156,11 @@ def probe(tmp_path):
             config.write_text(json.dumps(cfg))
             for side in keys:
                 assert 'LAUNCH_POLICY_OK' in cli(side, sessions[side])
-            for side in keys:
+            for side, key in keys.items():
                 captured = [x for x in peer.requests if x['body']['model'] == 'gpt-5-' + side]
                 assert len(captured) == 2, captured
                 assert all(x['path'] == '/' + side + '/v1/chat/completions' for x in captured)
-                assert all(x['auth'] == 'Bearer ' + keys[side] for x in captured)
+                assert all(x['auth'] == 'Bearer ' + key for x in captured)
                 bodies = [x['body'] for x in captured]
                 assert all(b.get('reasoning_effort') == ('high' if side == 'left' else 'low') for b in bodies), bodies
                 systems = [[m for m in b['messages'] if m['role'] in {'system', 'developer'}] for b in bodies]

tests/gateway/fixtures/kanban_owner_probe.py

@@ -117,7 +117,7 @@ def main():
             time.sleep(.05)
         raise AssertionError(('deadline', receipt))
     try:
-        with daemon(root, home, env, barrier=False) as (proc, desc):
+        with daemon(root, home, env, barrier=False) as (_proc, desc):
             subprocess.Popen = spawn
             try:
                 with closing(connect(board='owned')) as conn:

tests/gateway/fixtures/legacy_migration_probe.py

@@ -30,7 +30,7 @@ def probe(base, legacy_root, owner_root=None):
     def cli(checkout, *args):
         result = subprocess.run([sys.executable, '-m', 'hermes_cli.main', 'chat', *args],
             cwd=checkout, env={**env, 'PYTHONPATH': str(checkout)}, stdin=subprocess.DEVNULL,
-            capture_output=True, text=True, timeout=120)
+            capture_output=True, text=True, timeout=120, check=False)
         (home / ('cli-' + str(len(list(home.glob('cli-*')))) + '.log')).write_text(result.stdout + result.stderr)
         assert result.returncode == 0, result.stdout + result.stderr
         return result.stdout

tests/gateway/fixtures/rotation_persistence_worker.py

@@ -45,7 +45,7 @@ def main():
         store.update_system_prompt(scope['session_id'], 'BYTE_STABLE_PREFIX')
         agent._session_db_created = True
         before = agent.session_id
-        compressed, prompt = agent._compress_context(messages, 'BYTE_STABLE_PREFIX', force=True, approx_tokens=30000)
+        compressed, _prompt = agent._compress_context(messages, 'BYTE_STABLE_PREFIX', force=True, approx_tokens=30000)
         assert len(compressed) < len(messages), (len(compressed), len(messages), [
             dict(operation=e['operation'], keys=[list(m) for m in e['payload'].get('messages', [])])
             for e in store.journal['pending']])

tests/gateway/fixtures/stack_dump_daemon.py

@@ -17,7 +17,7 @@ import time
 path = Path(os.environ['HERMES_HOME']) / 'logs' / 'stacks.txt'
 path.parent.mkdir(parents=True, exist_ok=True)
 delay = float(os.environ.get('UGW_STACK_DUMP_AFTER', '60'))
-_sink = open(path, "w", encoding="utf-8")  # noqa: SIM115 - must outlive the dump
+_sink = open(path, "w", encoding="utf-8")
 faulthandler.dump_traceback_later(delay, repeat=True, file=_sink)
 _real_run = asyncio.run
 
@@ -39,7 +39,7 @@ def _describe(sink):
                        f'draining={attrs.get("_draining")} descriptor={json.dumps(descriptor, default=str)}\n')
     sink.write(f'pipe={windows_pipe_name(Path(os.environ["HERMES_HOME"]))}\n')
     for obj in gc.get_objects():
-        if type(obj) is GatewayControlServer:  # noqa: E721 - never isinstance over arbitrary gc objects
+        if type(obj) is GatewayControlServer:
             reply = obj.handle_request_line(b'{"protocol":1,"id":1,"verb":"identify"}', 'local')
             sink.write(f'control_server home={obj._home} handlers={sorted(obj._handlers)} '
                        f'pipe_server={obj._pipe_server!r} error={getattr(obj._pipe_server, "_error", None)!r}\n'

tests/gateway/test_adversarial_mutation_lifecycle.py

@@ -71,7 +71,7 @@ async def test_cancelling_model_request_cannot_strand_next_admission(tmp_path, m
 @pytest.mark.asyncio
 async def test_profile_retirement_waits_for_mutation_thread(tmp_path, monkeypatch):
     from gateway.run_runtime import _retire_profile_authority
-    owner, connection, ref, task, release, completed = await pending_model(tmp_path, monkeypatch)
+    owner, connection, _ref, task, release, completed = await pending_model(tmp_path, monkeypatch)
     retirement = asyncio.create_task(_retire_profile_authority(owner))
     try:
         # Retirement must not finish while the writer can still mutate this profile.

tests/gateway/test_api_authority_execution.py

@@ -18,7 +18,7 @@ def _probe(tmp_path, *, advanced=False, runs=False):
     if runs:
         env['API_RUN'] = str(runs)
     result = subprocess.run([sys.executable, str(Path(__file__).parent / 'fixtures' / 'api_authority_peer.py')],
-                            cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100)
+                            cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100, check=False)
     assert result.returncode == 0, result.stdout + '\n' + result.stderr
     receipt = json.loads((state / 'receipt.json').read_text())
     assert receipt['same_agent'], receipt

tests/gateway/test_api_canonical_projection.py

@@ -86,7 +86,7 @@ async def test_initialization_failure_is_a_failed_receipt_not_completed(api, own
 async def test_run_sse_carries_controls_tool_payloads_and_authoritative_cancellation(api, owner):
     from tools import clarify_gateway
     admitted = admit_api_turn(api, session_id='sse', user_message='hello', conversation_history=[])
-    _, ref, row = admitted
+    _, ref, _row = admitted
     launch, queue = _launch(api, admitted)
     tool_payloads = []
 
@@ -228,8 +228,8 @@ async def test_room_grant_answers_the_clarify_prompt_its_run_raised(api, owner,
     headers = {'Authorization': f'HermesRoom {grant}'}
     admitted = admit_api_turn(api, session_id='room-clarify', request_id='run_room', user_message='hello',
                               conversation_history=[])
-    _, ref, row = admitted
-    launch, queue = _launch(api, admitted, run_id='run_room')
+    _, ref, _row = admitted
+    launch, _queue = _launch(api, admitted, run_id='run_room')
     scope_request = SimpleNamespace(headers=headers, path='/v1/runs/run_room/clarify', method='POST')
     api._run_owners['run_room'] = api._run_idempotency_scope(scope_request)
 

tests/gateway/test_api_retired_media.py

@@ -27,7 +27,7 @@ def test_rejected_api_images_are_collected_and_history_owns_accepted_images(api,
     settle_session_input(owner.db, epoch=owner.epoch, admission_id=row['admission_id'],
                          generation=started['generation'], outcome='completed')
     owner.db.create_session('branch', source='api_server', model_config={'_branched_from': 's'})
-    owner.db.append_message('branch', 'user', '[Image attached at: %s]' % path)
+    owner.db.append_message('branch', 'user', f'[Image attached at: {path}]')
     owner.db.delete_session('s')
     assert path.read_bytes() == PNG, 'branch history still owns its image'
     owner.db.delete_session('branch')

tests/gateway/test_api_runs_native_session_refusal.py

@@ -8,7 +8,7 @@ from pathlib import Path
 
 def test_runs_refuse_native_session_without_phantom_run(tmp_path):
     result = subprocess.run([sys.executable, __file__], cwd=Path(__file__).resolve().parents[2],
-        env=dict(os.environ, HERMES_HOME=str(tmp_path)), capture_output=True, text=True, timeout=90)
+        env=dict(os.environ, HERMES_HOME=str(tmp_path)), capture_output=True, text=True, timeout=90, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print((tmp_path / 'runs-receipt.json').read_text(encoding='utf-8'))
 

tests/gateway/test_api_unknown_resolution.py

@@ -251,7 +251,7 @@ async def test_http_resolve_unknown_refuses_stale_or_foreign_admission(
 async def test_fresh_adapter_recovers_non_idempotent_owner_but_refuses_other_credential(
     tmp_path, monkeypatch
 ):
-    client, adapters, store, owner, unknown, follower, executed, tasks = (
+    client, adapters, store, _owner, unknown, _follower, _executed, tasks = (
         await _restarted_api_runs(tmp_path, monkeypatch, idempotent=False)
     )
     try:

tests/gateway/test_authority_automation.py

@@ -11,7 +11,7 @@ def _probe(tmp_path, mode):
     env['PYTHONPATH'] = str(root)
     result = subprocess.run([sys.executable, str(root / 'tests/gateway/fixtures/automation_peer.py'),
                              str(tmp_path), mode], env=env, cwd=root, stdin=subprocess.DEVNULL,
-                            capture_output=True, text=True, timeout=200)
+                            capture_output=True, text=True, timeout=200, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print(result.stdout)
 

tests/gateway/test_authority_connection.py

@@ -44,7 +44,7 @@ async def test_repeated_resume_and_close_preserve_only_live_memberships(tmp_path
 async def test_close_after_a_subscribed_session_was_deleted_releases_every_membership(tmp_path, deleted_first):
     """Deletion evicts the live session before the viewer disconnects; teardown must still
     release the other subscription and the transport instead of raising not_found."""
-    import tui_gateway.ws  # noqa: F401 — the fanout drain thread imports it lazily during teardown
+    import tui_gateway.ws
     db = SessionDB(db_path=tmp_path / 'state.db')
     try:
         for sid in ('doomed', 'kept'):

tests/gateway/test_authority_events.py

@@ -18,7 +18,7 @@ def test_full_observer_cannot_block_real_authority_execution(tmp_path):
                PYTHONPATH=str(repo), PYTHONUNBUFFERED='1')
     result = subprocess.run(
         [sys.executable, str(Path(__file__).parent / 'fixtures' / 'authority_events_peer.py')],
-        cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100)
+        cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100, check=False)
     assert result.returncode == 0, result.stdout + '\n' + result.stderr
     print(result.stdout)
 

tests/gateway/test_authority_resident_stop.py

@@ -13,7 +13,7 @@ async def test_stop_reaches_resident_agent_after_real_cache_eviction(tmp_path, m
     db, authority = _authority(tmp_path, monkeypatch)
     calls = []
     agent = SimpleNamespace(interrupt=lambda: calls.append('stop'))
-    runner, state = _build_gateway(agent, [])
+    runner, _state = _build_gateway(agent, [])
     authority.runner = runner
     authority.sessions['s'].route = KEY
     with db:

tests/gateway/test_authority_stream.py

@@ -17,7 +17,7 @@ def test_authority_stream_has_one_order_and_rejects_retired_callbacks(tmp_path,
     env = {k: os.environ[k] for k in ('PATH', 'SYSTEMROOT', 'LANG', 'TZ') if k in os.environ}
     env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(state), PYTHONPATH=str(repo), STREAM_PROBE_MODE=mode)
     result = subprocess.run([sys.executable, str(Path(__file__).parent / 'fixtures' / 'authority_stream_peer.py')],
-                            cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100)
+                            cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100, check=False)
     assert result.returncode == 0, result.stdout + '\n' + result.stderr
     receipt = json.loads((state / 'receipt.json').read_text())
     print(json.dumps(receipt))

tests/gateway/test_authority_unknown_resolution.py

@@ -58,7 +58,7 @@ async def _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed):
 @pytest.mark.asyncio
 async def test_resolve_unknown_releases_the_paused_follower_exactly_once(tmp_path, monkeypatch):
     executed = []
-    store, authority, ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
+    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
     peer = Peer()
     viewer = AuthorityConnection(authority, peer, {'user_id': 'human'})
     try:
@@ -97,7 +97,7 @@ async def test_resolve_unknown_releases_the_paused_follower_exactly_once(tmp_pat
 @pytest.mark.asyncio
 async def test_resolve_unknown_is_fenced_by_generation_and_capability(tmp_path, monkeypatch):
     executed = []
-    store, authority, ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
+    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
     owner = AuthorityConnection(authority, Peer(), {'user_id': 'human'})
     reader = AuthorityConnection(authority, Peer(), {'user_id': 'human', 'capabilities': ['session:read', 'session:submit']})
     try:
@@ -127,7 +127,7 @@ async def test_discard_and_its_transcript_boundary_commit_atomically_on_the_phys
     """R1: a failure writing the boundary must not leave the lost admission terminal (its exact
     Discard retry refused, the follower claimable onto an open user tail). The boundary lands on
     the rotated physical transcript, not the logical owner."""
-    store, authority, ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
+    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
     db = authority.db
     db.create_session('s-tip', source='telegram', parent_session_id='s')
     db.end_session('s', 'compression')
@@ -165,7 +165,7 @@ async def test_discarded_lost_turn_is_closed_and_never_merged_into_the_follower_
     from tests.acp_adapter.test_failed_turn_closure import _LoopbackProvider
     monkeypatch.setenv('HERMES_DISABLE_PLUGINS', '1')
     monkeypatch.setenv('NO_PROXY', '127.0.0.1,localhost')
-    store, authority, ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
+    store, authority, _ref, unknown, _follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
     db = authority.db
     db.append_message('s', 'user', 'HEAD lost in restart')  # the lost turn's turn-start flush
     viewer = AuthorityConnection(authority, Peer(), {'user_id': 'human'})

tests/gateway/test_automation_compression_media.py

@@ -47,7 +47,7 @@ async def test_completion_after_compression_admits_to_the_logical_root(tmp_path,
     event = dict(pending(key, 'after-compression'), parent_session_id=child)
     assert await runner._deliver_async_delegation_group([event]) is True
     rows = list_session_admissions(db, session_id=root, pending_only=False)
-    assert [r['request_id'] for r in rows][0] == 'human' and len(rows) == 2, rows
+    assert next(r['request_id'] for r in rows) == 'human' and len(rows) == 2, rows
     assert 'after-compression' in rows[-1]['payload']['text']
     assert list_session_admissions(db, session_id=child, pending_only=False) == []
     # A lost-ACK redelivery is the same admission, found under the root's ledger.

tests/gateway/test_automation_notification_category.py

@@ -42,7 +42,7 @@ async def test_bot_category_is_committed_restored_and_part_of_retry_identity(tmp
     from gateway.session_local import create_local_session
     from gateway.session_local_title import title_new_session
     from gateway.session_automation import restore_local_automation
-    runner, authority = await _authority(tmp_path, monkeypatch)
+    _runner, authority = await _authority(tmp_path, monkeypatch)
     actor = Principal('test-owner', 'default', frozenset({'session:create', 'session:submit', 'session:read'}), 'fixture')
     try:
         ref = create_local_session(authority, actor, dict(request_id='bot', source='gui', model='fixture', toolsets=[]))
@@ -84,7 +84,7 @@ async def test_bot_category_is_committed_restored_and_part_of_retry_identity(tmp
 @pytest.mark.parametrize('category', [None, [], {}, 'other'])
 async def test_bot_invalid_category_refuses_before_admission(tmp_path, monkeypatch, category):
     from gateway.session_bot import deliver
-    runner, authority = await _authority(tmp_path, monkeypatch)
+    _runner, authority = await _authority(tmp_path, monkeypatch)
     actor = Principal('test-owner', 'default', frozenset({'session:submit'}), 'fixture')
     try:
         with pytest.raises(RuntimeStoreError, match='invalid_params'):

tests/gateway/test_bot_authority.py

@@ -11,6 +11,6 @@ def test_bot_delivery_busy_retry_keeps_exact_target(tmp_path):
     env['PYTHONPATH'] = str(root)
     proc = subprocess.run([sys.executable, str(root / 'tests/gateway/fixtures/bot_authority_peer.py'),
                            str(tmp_path)], cwd=root, env=env, stdin=subprocess.DEVNULL,
-                          capture_output=True, text=True, timeout=150)
+                          capture_output=True, text=True, timeout=150, check=False)
     assert proc.returncode == 0, proc.stdout + proc.stderr
     print(proc.stdout)

tests/gateway/test_boundary_evict_commits_memory.py

@@ -6,7 +6,7 @@ import pytest
 
 # session.create's resume path imports this lazily mid-request; importing it whole here keeps the
 # conftest server-state fixture from meeting a half-initialized module at teardown.
-import tui_gateway.server  # noqa: F401
+import tui_gateway.server
 from gateway.run import GatewayRunner
 
 
@@ -84,7 +84,7 @@ async def test_reset_and_compress_mutations_use_the_boundary_evict(tmp_path, mon
     from gateway.session import SessionStore
     from gateway.session_authority import initialize_session_authority
     from gateway.session_controls import AuthorityConnection
-    import gateway.run as run
+    from gateway import run
 
     monkeypatch.setenv('HERMES_HOME', str(tmp_path))
     monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {'default': 'fixture'}, 'platform_toolsets': {'cli': []}})

tests/gateway/test_cold_platform_activation.py

@@ -43,7 +43,7 @@ def _load(tmp_path, *, config_yaml="", env=None, dotenv=""):
                  if k in {"PATH", "LANG", "TZ", "HERMES_RUNTIME_DIR", "TMPDIR", "SYSTEMROOT"}}
     child_env.update(HOME=str(tmp_path), HERMES_HOME=str(home), PYTHONPATH=str(REPO), **(env or {}))
     out = subprocess.run([sys.executable, "-c", _PROBE], cwd=str(REPO), env=child_env,
-                         capture_output=True, text=True, timeout=120)
+                         capture_output=True, text=True, timeout=120, check=False)
     assert out.returncode == 0, out.stderr[-2000:]
     return json.loads(out.stdout.strip().splitlines()[-1])
 

tests/gateway/test_command_discovery.py

@@ -37,7 +37,7 @@ async def _legacy_ws_snapshot(root, home, env, requests):
     legacy_env = {**env, 'HERMES_HOME': str(legacy_home),
                   'HERMES_DASHBOARD_SESSION_TOKEN': 'disposable-discovery-token'}
     with (home.parent / 'legacy-serve.log').open('w+') as log:
-        process = subprocess.Popen([sys.executable, '-m', 'hermes_cli.main', 'serve',
+        process = subprocess.Popen([sys.executable, '-m', 'hermes_cli.main', 'serve',  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
                                     '--host', '127.0.0.1', '--port', '0', '--skip-build', '--isolated'],
                                    cwd=root, env=legacy_env, stdin=subprocess.DEVNULL,
                                    stdout=log, stderr=subprocess.STDOUT)
@@ -159,7 +159,7 @@ sys.__stdout__.write(json.dumps(results) + '\\n')
             # Ordinary startup syncs bundled skills; compare both transports on that same snapshot.
             legacy = subprocess.run([sys.executable, '-c', legacy_code, json.dumps(requests)],
                                     cwd=root, env=env, stdin=subprocess.DEVNULL,
-                                    capture_output=True, text=True, timeout=45)
+                                    capture_output=True, text=True, timeout=45, check=False)
             assert legacy.returncode == 0, legacy.stderr
             expected = json.loads(legacy.stdout.splitlines()[-1])
             pairs = dict(expected[0]['pairs'])

tests/gateway/test_hosted_room_attachment_publication.py

@@ -143,7 +143,7 @@ _INVALIDATE_APPEND = {
 
 @pytest.mark.parametrize("invalid", ["room", "event", "metadata", "private", "expired"])
 def test_append_requires_exact_live_room_visible_commitment(tmp_path, invalid):
-    db, store, manifest, args = _staged(tmp_path, viewer=invalid != "private")
+    db, _store, _manifest, args = _staged(tmp_path, viewer=invalid != "private")
     _INVALIDATE_APPEND[invalid](db, args)
     with pytest.raises(hosted_rooms.EventCursorConflictError):
         hosted_rooms.append_event(db, **args)

tests/gateway/test_kanban_recovery.py

@@ -21,6 +21,6 @@ def test_owner_board_recovery(tmp_path, mode):
                PYTHONPATH=str(repo), KANBAN_RECOVERY_MODE=mode)
     result = subprocess.run([sys.executable, str(repo / 'tests/gateway/fixtures/kanban_recovery_probe.py')],
                             cwd=repo, env=env, stdin=subprocess.DEVNULL,
-                            capture_output=True, text=True, timeout=180)
+                            capture_output=True, text=True, timeout=180, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print(json.dumps(json.loads((tmp_path / 'state/receipt.json').read_text())))

tests/gateway/test_local_automation.py

@@ -13,7 +13,7 @@ def _probe(tmp_path, source):
     env['PYTHONPATH'] = str(root)
     result = subprocess.run([sys.executable, str(root / 'tests/gateway/fixtures/local_automation_peer.py'),
         str(tmp_path), source], cwd=root, env=env, stdin=subprocess.DEVNULL,
-        capture_output=True, text=True, timeout=150)
+        capture_output=True, text=True, timeout=150, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print(result.stdout)
 

tests/gateway/test_local_session.py

@@ -18,7 +18,7 @@ def test_fresh_local_session_owns_execution_across_viewers(tmp_path, kind):
     env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(state), PYTHONPATH=str(repo))
     result = subprocess.run(
         [sys.executable, str(Path(__file__).parent / 'fixtures' / 'local_session_peer.py'), kind],
-        cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100)
+        cwd=repo, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=100, check=False)
     assert result.returncode == 0, result.stdout + '\n' + result.stderr
     receipt = json.loads((state / 'receipt.json').read_text())
     assert receipt['human_response'] and receipt['same_agent'] and receipt['detached_pending']

tests/gateway/test_local_session_lineage.py

@@ -17,7 +17,7 @@ async def test_local_lineage_transitions_preserve_owner_or_roll_back(tmp_path, m
     from hermes_state_runtime import (RuntimeStoreError, admit_session_input, claim_session_input,
                                       get_session_admission, list_session_admissions)
 
-    monkeypatch.setattr(run, '_load_gateway_config', lambda: {})
+    monkeypatch.setattr(run, '_load_gateway_config', dict)
     def runner():
         store = SessionStore(tmp_path / 'sessions', GatewayConfig())
         return SimpleNamespace(session_store=store, _session_db=store._db,

tests/gateway/test_local_source_labels.py

@@ -12,7 +12,7 @@ async def _authority(tmp_path, monkeypatch):
     from gateway.config import GatewayConfig
     from gateway.session import SessionStore
     from gateway.session_authority import initialize_session_authority
-    import gateway.run as run
+    from gateway import run
 
     monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {'default': 'fixture'}, 'platform_toolsets': {'cli': []}})
     monkeypatch.setattr(run, '_resolve_gateway_model', lambda config: 'fixture')

tests/gateway/test_managed_turn_shutdown_drain.py

@@ -19,7 +19,7 @@ from tests.gateway.restart_test_helpers import make_restart_runner
 async def test_drain_counts_managed_turns_and_the_timeout_path_stops_them(tmp_path):
     from gateway.session_managed_worker import ManagedWorker
     seen = tmp_path / 'controls'
-    child = subprocess.Popen([sys.executable, '-c', 'import sys; open(sys.argv[1], "wb").write(sys.stdin.buffer.readline())',
+    child = subprocess.Popen([sys.executable, '-c', 'import sys; open(sys.argv[1], "wb").write(sys.stdin.buffer.readline())',  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
                               str(seen)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
     worker = ManagedWorker(child)
     worker.writer.start()

tests/gateway/test_managed_worker_background_output.py

@@ -40,7 +40,7 @@ def test_worker_spawned_process_output_and_exit_survive_the_spawner(tmp_path):
         print(json.dumps(session.id), flush=True)
         os._exit(0)
     ''')
-    out = subprocess.run([sys.executable, '-c', spawner], env=env, cwd=ROOT, capture_output=True, text=True, encoding='utf-8', timeout=60)
+    out = subprocess.run([sys.executable, '-c', spawner], env=env, cwd=ROOT, capture_output=True, text=True, encoding='utf-8', timeout=60, check=False)
     session_id = json.loads(out.stdout.strip().splitlines()[-1])
     adopter = textwrap.dedent(f'''
         import json
@@ -49,7 +49,7 @@ def test_worker_spawned_process_output_and_exit_survive_the_spawner(tmp_path):
         registry.recover_from_checkpoint()
         print(json.dumps(registry.wait({session_id!r}, timeout=30)))
     ''')
-    out = subprocess.run([sys.executable, '-c', adopter], env=env, cwd=ROOT, capture_output=True, text=True, encoding='utf-8', timeout=60)
+    out = subprocess.run([sys.executable, '-c', adopter], env=env, cwd=ROOT, capture_output=True, text=True, encoding='utf-8', timeout=60, check=False)
     result = json.loads(out.stdout.strip().splitlines()[-1])
     assert result['status'] == 'exited', (result, out.stderr[-2000:])
     assert result['exit_code'] == 7 and result['output'] == 'FIRST_42\nSECOND_42\n', result
@@ -72,7 +72,7 @@ class Model(BaseHTTPRequestHandler):
                                                   'arguments': {'action': 'wait', 'session_id': proc, 'timeout': 30}}]}
         else:
             name, args = None, {}
-        message = ({'role': 'assistant', 'content': None, 'tool_calls': [{'id': 'c%d' % len(messages), 'index': 0,
+        message = ({'role': 'assistant', 'content': None, 'tool_calls': [{'id': f'c{len(messages)}', 'index': 0,
                     'type': 'function', 'function': {'name': name, 'arguments': json.dumps(args)}}]}
                    if name else {'role': 'assistant', 'content': 'OK'})
         choice = {'index': 0, 'delta': message, 'finish_reason': 'tool_calls' if name else 'stop'}

tests/gateway/test_managed_worker_bypass.py

@@ -90,7 +90,7 @@ print(json.dumps({{'max_turns': cfg['agent']['max_turns'], 'raw_model': raw['mod
     env = {**child_env(), **{k: os.environ[k] for k in ('TIRITH_ENABLED',) if k in os.environ}}
     env.update(HOME=str(tmp_path), USERPROFILE=str(tmp_path), HERMES_HOME=str(home), PYTHONPATH=str(root))
     result = subprocess.run([sys.executable, str(script)], cwd=root, env=env, stdin=subprocess.DEVNULL,
-                            capture_output=True, text=True, timeout=90)
+                            capture_output=True, text=True, timeout=90, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     receipt = json.loads(result.stdout.strip().splitlines()[-1])
     assert receipt['max_turns'] == 7 and receipt['helper'] == 7

tests/gateway/test_managed_worker_detached_work.py

@@ -82,7 +82,7 @@ def test_managed_turn_runs_detachable_work_inline_and_leaves_no_orphan(tmp_path)
                OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=url)
     created = subprocess.run([sys.executable, '-c', 'import json; from cron.jobs import create_job; '
                               'print(json.dumps(create_job("CRON_CHILD_GOAL", "every 1h", name="inline", deliver="local")["id"]))'],
-                             env=env, cwd=root, capture_output=True, text=True, timeout=60)
+                             env=env, cwd=root, capture_output=True, text=True, timeout=60, check=False)
     peer.job_id = json.loads(created.stdout.strip().splitlines()[-1])
 
     def rows(sql):

tests/gateway/test_managed_worker_drain.py

@@ -15,7 +15,7 @@ async def test_reserved_worker_persists_while_runtime_drains(tmp_path):
     from hermes_state import SessionDB
     from hermes_state_runtime import admit_session_input, begin_runtime_epoch, claim_session_input, RuntimeStoreError
     db = SessionDB(tmp_path / 'state.db')
-    process = subprocess.Popen([sys.executable, '-c', 'import sys; sys.stdin.read()'], stdin=subprocess.PIPE,
+    process = subprocess.Popen([sys.executable, '-c', 'import sys; sys.stdin.read()'], stdin=subprocess.PIPE,  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
     try:
         db.create_session('owned', 'cli')

tests/gateway/test_managed_worker_fencing.py

@@ -60,7 +60,7 @@ async def test_late_bootstrap_handshake_cannot_rearm_a_recovered_admission(tmp_p
         lost = admit_session_input(db, epoch=epoch, principal_id='human', session_id='owned', request_id='lost', payload={})
         row = claim_session_input(db, epoch=epoch, session_id='owned')
         follower = admit_session_input(db, epoch=epoch, principal_id='human', session_id='owned', request_id='next', payload={})
-        child = subprocess.Popen([sys.executable, '-c', 'import sys; sys.stdin.read()'],
+        child = subprocess.Popen([sys.executable, '-c', 'import sys; sys.stdin.read()'],  # noqa: ASYNC220 -- Popen returns immediately; the test drives the child through its handle
             stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
         try:
             scope = reserve_admission_worker(authority, admission_id=row['admission_id'], process=child,

tests/gateway/test_managed_worker_frozen_config.py

@@ -154,7 +154,7 @@ def test_ordinary_worker_binding_freezes_readers_but_keeps_customizations(tmp_pa
     env = {**child_env(), **{k: os.environ[k] for k in ('TIRITH_ENABLED',) if k in os.environ}}
     env.update(HOME=str(tmp_path), USERPROFILE=str(tmp_path), HERMES_HOME=str(home), PYTHONPATH=str(root))
     result = subprocess.run([sys.executable, str(script)], cwd=root, env=env, stdin=subprocess.DEVNULL,
-                            capture_output=True, text=True, timeout=90)
+                            capture_output=True, text=True, timeout=90, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     receipt = json.loads(result.stdout.strip().splitlines()[-1])
     assert receipt == {'safe': False, 'turns': [7, 7, 7], 'mcp': 'FROZEN_A', 'plugin': True}, receipt

tests/gateway/test_managed_worker_turn_flags.py

@@ -23,7 +23,7 @@ class Model(BaseHTTPRequestHandler):
         body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
         last = (body.get('messages') or [{'role': 'system'}])[-1]
         if last['role'] == 'user':
-            message = {'role': 'assistant', 'content': None, 'tool_calls': [{'id': 'call-%d' % len(body['messages']),
+            message = {'role': 'assistant', 'content': None, 'tool_calls': [{'id': f"call-{len(body['messages'])}",
                 'type': 'function', 'function': {'name': 'terminal',
                 'arguments': json.dumps({'command': self.server.command, 'timeout': 10})}}]}
         else:

tests/gateway/test_msgraph_authority.py

@@ -9,7 +9,7 @@ import sys
 
 def test_graph_boundary_commits_only_valid_notifications(tmp_path):
     result = subprocess.run([sys.executable, __file__], env=dict(os.environ, HERMES_HOME=str(tmp_path)),
-                            capture_output=True, text=True, timeout=90)
+                            capture_output=True, text=True, timeout=90, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print((tmp_path / 'graph-receipt.json').read_text())
 

tests/gateway/test_multiplex_cold_profile_sidebar_mutation.py

@@ -83,7 +83,7 @@ def desktop_ensure(env, root, profile):
     """The exact child the Desktop spawns: ``hermes --profile <p> gateway ensure --json``."""
     result = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '--profile', profile,
                              'gateway', 'ensure', '--json', '--timeout', '30'],
-                            cwd=root, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=60)
+                            cwd=root, env=env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=60, check=False)
     payload = json.loads(result.stdout)
     assert result.returncode == 0 and payload['state'] == 'ready', (result.returncode, payload, result.stderr[-500:])
     return payload['endpoint']

tests/gateway/test_multiplex_hot_serve_live.py

@@ -155,7 +155,7 @@ def test_profile_created_under_live_multiplexer_is_served_and_owned(mux):
         assert served_homes(desc) == {root.resolve(), mux['boot'].resolve()}
         # ``hermes profile create`` under the multiplexer's root: the CLI signals the control socket.
         created = subprocess.run([sys.executable, '-m', 'hermes_cli.main', 'profile', 'create', 'fresh'],
-                                 cwd=ROOT, env=env, capture_output=True, text=True, timeout=120)
+                                 cwd=ROOT, env=env, capture_output=True, text=True, timeout=120, check=False)
         assert created.returncode == 0, (created.stdout, created.stderr)
         fresh = root / 'profiles' / 'fresh'
         assert fresh.is_dir()
@@ -173,7 +173,7 @@ def test_profile_created_under_live_multiplexer_is_served_and_owned(mux):
             assert db.execute('SELECT 1 FROM sessions WHERE id=?', (sid,)).fetchone()
         # ``hermes -p fresh`` resolves to the multiplexer instead of spawning a daemon of its own.
         ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'fresh', 'gateway', 'ensure',
-                                  '--timeout', '15'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=60)
+                                  '--timeout', '15'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=60, check=False)
         payload = json.loads(ensured.stdout.strip().splitlines()[-1])
         assert payload['state'] == 'ready' and payload['endpoint']['instance_id'] == desc['instance_id'], (payload, ensured.stderr)
         assert not (fresh / 'gateway.pid').exists()
@@ -214,7 +214,7 @@ def test_named_profile_ensure_during_multiplexer_downtime_starts_the_multiplexer
     assert recorded_served(root) == ['default', 'boot']
     try:
         ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'boot', 'gateway', 'ensure',
-                                  '--json', '--timeout', '90'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=120)
+                                  '--json', '--timeout', '90'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=120, check=False)
         payload = json.loads(ensured.stdout.strip().splitlines()[-1])
         assert payload['state'] == 'ready', (payload, ensured.stderr)
         # The answering owner is the multiplexer at the root, serving boot as a secondary.
@@ -253,7 +253,7 @@ def test_broken_secondary_state_db_parks_only_that_profile(mux):
         started = time.monotonic()
         ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'broken', 'gateway', 'ensure',
                                   '--json', '--timeout', '30'], cwd=ROOT, env=mux['env'], capture_output=True,
-                                 text=True, timeout=60)
+                                 text=True, timeout=60, check=False)
         payload = json.loads(ensured.stdout.strip().splitlines()[-1])
         assert (payload['state'], payload['reason_code']) == ('inaccessible', 'profile_parked'), (payload, ensured.stderr)
         assert 'unusable' in (payload.get('detail') or ''), payload

tests/gateway/test_native_ingress_media.py

@@ -42,7 +42,7 @@ def test_accepted_media_survives_mutation_cleanup_and_owner_kill(tmp_path):
     for calls in (1, 0):
         command[-1] = 'recover'
         result = subprocess.run(command, cwd=repo, env=env, stdin=subprocess.DEVNULL,
-                                capture_output=True, text=True, timeout=60)
+                                capture_output=True, text=True, timeout=60, check=False)
         assert result.returncode == 0, result.stdout + result.stderr
         receipt = json.loads((state / 'recovered.json').read_text())
         assert receipt['model_calls'] == calls, receipt
@@ -52,6 +52,6 @@ def test_accepted_media_survives_mutation_cleanup_and_owner_kill(tmp_path):
 def test_native_context_never_upgrades_metadata_or_transport_trust(tmp_path):
     command, env, repo, state = _peer(tmp_path, 'guards')
     result = subprocess.run(command, cwd=repo, env=env, stdin=subprocess.DEVNULL,
-                            capture_output=True, text=True, timeout=45)
+                            capture_output=True, text=True, timeout=45, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print((state / 'guards.json').read_text())

tests/gateway/test_native_ingress_trust.py

@@ -14,7 +14,7 @@ def _probe(tmp_path, mode):
     env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(state), PYTHONPATH=str(repo))
     result = subprocess.run([sys.executable, str(Path(__file__).parent / 'fixtures' /
                              'native_ingress_trust_peer.py'), mode], cwd=repo, env=env,
-                            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=40)
+                            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=40, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print(result.stdout)
 

tests/gateway/test_native_pause_notice.py

@@ -37,7 +37,7 @@ async def test_paused_fifo_releases_the_messaging_waiter_with_one_notice_per_epi
     from tests.gateway.test_authority_unknown_resolution import Peer, _restarted_owner_with_unknown_head
 
     executed = []
-    store, authority, ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
+    store, authority, ref, unknown, _follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
     try:
         actor = Principal('human', 'p', frozenset({'session:submit'}), 'fixture')
 
@@ -129,14 +129,14 @@ def test_crash_mid_turn_then_platform_message_gets_one_pause_notice_and_reset_un
         wait_for(lambda: rows().get(text, {}).get('status') == status, diagnostic)
 
     try:
-        with daemon(root, home, env, barrier=False) as (proc, desc):
+        with daemon(root, home, env, barrier=False) as (proc, _desc):
             send('WARM_HISTORY', 'terminal')
             send('BLOCK_STARTED', 'started')
             assert model.blocked.wait(20), diagnostic()
             proc.kill()
             assert proc.wait(timeout=10) == -9
         model.release.set()
-        with daemon(root, home, env, barrier=False) as (proc, desc):
+        with daemon(root, home, env, barrier=False) as (proc, _desc):
             wait_for(lambda: rows()['BLOCK_STARTED']['status'] == 'unknown', diagnostic)
             paused_session = rows()['BLOCK_STARTED']['session']
             # First message onto the paused FIFO: admitted (durable), the user is told once,

tests/gateway/test_native_role_reauthorization.py

@@ -25,7 +25,7 @@ def _probe(tmp_path, mode):
     env.update(HOME=str(home), USERPROFILE=str(home), HERMES_HOME=str(state), PYTHONPATH=str(repo))
     result = subprocess.run([sys.executable, str(Path(__file__).parent / 'fixtures' /
                              'native_role_peer.py'), mode], cwd=repo, env=env,
-                            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=60)
+                            stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=60, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     print(result.stdout)
 

tests/gateway/test_native_telegram_startup_recovery.py

@@ -14,6 +14,7 @@ import importlib.machinery
 import pytest
 
 from tests.gateway.fixtures.local_recovery_probe import Model, child_env, daemon
+import itertools
 
 
 class BotAPI(BaseHTTPRequestHandler):
@@ -165,7 +166,7 @@ def test_telegram_fifo_unknown_and_current_authorization_survive_sigkill(tmp_pat
         observed = [next((marker for marker in markers if marker in text), text) for text in texts]
         assert observed == markers, observed
         assert len(set(b['pid'] for b in boots)) == len(boots)
-        assert all(b['epoch'] > a['epoch'] for a, b in zip(boots, boots[1:]))
+        assert all(b['epoch'] > a['epoch'] for a, b in itertools.pairwise(boots))
         assert 'getUpdates' in bot.calls and bot.sent
         print(json.dumps({'boots': boots, 'model_inputs': observed, 'final_rows': rows()}))
     finally:

tests/gateway/test_operator_cross_surface.py

@@ -21,7 +21,7 @@ def test_operator_cross_surface_real_auth_fifo_and_attribution(tmp_path, creator
                PYTHONPATH=str(root), PYTHONUNBUFFERED='1')
     result = subprocess.run(
         [sys.executable, str(root / 'tests/gateway/fixtures/operator_cross_surface_peer.py'), creator],
-        cwd=root, env=env, capture_output=True, text=True, timeout=150)
+        cwd=root, env=env, capture_output=True, text=True, timeout=150, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     receipt = json.loads((home / 'operator-receipt.json').read_text())
     assert receipt['creator'] == creator

tests/gateway/test_relay_authority.py

@@ -15,7 +15,7 @@ import time
 def test_authenticated_relay_uses_canonical_admission(tmp_path):
     env = dict(os.environ, HOME=str(tmp_path), HERMES_HOME=str(tmp_path), PYTHONUNBUFFERED='1')
     result = subprocess.run([sys.executable, __file__, 'peer'], env=env,
-                            capture_output=True, text=True, timeout=100)
+                            capture_output=True, text=True, timeout=100, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     receipt = json.loads((tmp_path / 'relay-receipt.json').read_text())
     assert receipt['same_agent'] and receipt['negative_controls']

tests/gateway/test_retired_media_scan_bound.py

@@ -41,7 +41,7 @@ def test_candidates_share_one_messages_pass(owner, tmp_path, monkeypatch):
         path.write_bytes(b'x')
         references.append({'path': str(path), 'sha256': sha, 'size': 1})
     for reference in references[:3]:
-        db.append_message('keep', 'user', '[Image attached at: %s]' % reference['path'])
+        db.append_message('keep', 'user', '[Image attached at: {}]'.format(reference['path']))
     db.append_message('keep', 'user', 'unrelated history')
     db._execute_write(lambda conn: conn.executemany('INSERT INTO state_meta(key,value) VALUES(?,?)',
         [(PREFIX + str(index), json.dumps(reference)) for index, reference in enumerate(references)]))

tests/gateway/test_runtime_ownership.py

@@ -31,7 +31,7 @@ print('CLAIMED', result)
     witness = tmp_path / 'witness'
     try:
         result = subprocess.run([sys.executable, '-c', code], env={**os.environ, 'WITNESS': str(witness)},
-                                stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30)
+                                stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=30, check=False)
         assert result.returncode == 0, result.stderr
         assert not witness.exists(), result.stdout + result.stderr
         assert 'CLAIMED False' in result.stdout

tests/gateway/test_runtime_worker_shutdown.py

@@ -20,7 +20,7 @@ async def test_managed_claims_participate_in_zero_and_nonzero_drain(tmp_path):
         's': SimpleNamespace(control=lambda frame: stopped.append(frame), closed=threading.Event())})
     runner = SimpleNamespace(session_authority=authority, _running_agents={},
         _running_agent_count=lambda: 0, _active_cron_job_count=lambda: 0, _active_api_run_count=lambda: 0,
-        _active_deferred_agent_worker_count=lambda: 0, _snapshot_running_agents=lambda: {},
+        _active_deferred_agent_worker_count=lambda: 0, _snapshot_running_agents=dict,
         _update_runtime_status=lambda *a: None)
     runner._drain_work_counts = GatewayShutdownMixin._drain_work_counts.__get__(runner)
     authority.runner = runner

tests/gateway/test_runtime_writer_drain.py

@@ -12,7 +12,7 @@ import pytest
 @pytest.mark.platforms("linux")
 @pytest.mark.parametrize('partial_start', [False, True])
 def test_live_runtime_writers_prevent_final_lock_release(tmp_path, monkeypatch, partial_start):
-    import gateway.run as run
+    from gateway import run
     from gateway.run_bootstrap import _start_gateway_start_cron_and_housekeeping
     from gateway.status import acquire_gateway_runtime_lock, release_gateway_runtime_lock
     from cron import scheduler_provider
@@ -58,7 +58,7 @@ if claimed:
 
     def contender():
         result = subprocess.run([sys.executable, '-c', probe], env=os.environ.copy(),
-                                stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=15)
+                                stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=15, check=False)
         assert result.returncode == 0, result.stderr
         return result.stdout.strip()
 

tests/gateway/test_session_a2a.py

@@ -12,7 +12,7 @@ async def test_forwarded_identity_survives_reconnect_without_policy_override(tmp
     from gateway.config import GatewayConfig
     from gateway.session import SessionStore
     from gateway.session_authority import initialize_session_authority
-    import gateway.run as run
+    from gateway import run
     from gateway.session_policy import build_policy
     from hermes_state_runtime import RuntimeStoreError
 
@@ -67,7 +67,7 @@ async def test_forwarding_does_not_merge_lossy_context_or_peer_identity(tmp_path
     from gateway.config import GatewayConfig
     from gateway.session import SessionStore
     from gateway.session_authority import initialize_session_authority
-    import gateway.run as run
+    from gateway import run
 
     monkeypatch.setenv('HERMES_HOME', str(tmp_path))
     monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {'default': 'fixture'}, 'platform_toolsets': {'cli': []}})
@@ -123,7 +123,7 @@ async def test_forwarded_turns_are_authored_by_the_peer_for_memory_attribution(t
     from gateway.session_ingress import row_turn_author
     from agent.turn_author import a2a_key
     from hermes_state_runtime import list_session_admissions
-    import gateway.run as run
+    from gateway import run
 
     monkeypatch.setenv('HERMES_HOME', str(tmp_path))
     monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {'default': 'fixture'}, 'platform_toolsets': {'cli': []}})
@@ -166,8 +166,8 @@ async def test_forward_wakes_on_the_completion_frame_instead_of_a_fixed_100ms_po
     import asyncio
     import time
     from websockets.asyncio.server import serve
-    import hermes_cli.gateway_client as gateway_client
-    import hermes_cli.gateway_runtime as gateway_runtime
+    from hermes_cli import gateway_client
+    from hermes_cli import gateway_runtime
     from gateway.session_a2a import forward_to_owner
 
     calls = []

tests/gateway/test_session_activity_ordering.py

@@ -1,5 +1,5 @@
 """Recovered UTC routes coexist with legacy local-time routes."""
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timedelta, timezone, UTC
 
 from gateway.config import GatewayConfig, Platform
 from gateway.session import SessionSource, SessionStore
@@ -9,7 +9,7 @@ def test_list_sessions_orders_and_filters_mixed_clock_routes(tmp_path):
     store = SessionStore(sessions_dir=tmp_path / 'sessions', config=GatewayConfig())
     recent = store.get_or_create_session(SessionSource(platform=Platform.TELEGRAM, chat_id='recent'))
     old = store.get_or_create_session(SessionSource(platform=Platform.TELEGRAM, chat_id='old'))
-    recent.updated_at = datetime.now(timezone.utc)
+    recent.updated_at = datetime.now(UTC)
     old.updated_at = datetime.now() - timedelta(hours=2)
 
     assert store.list_sessions() == [recent, old]

tests/gateway/test_session_ancillary_reads.py

@@ -16,7 +16,7 @@ def probe(tmp_path, mode):
     env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home),
                PYTHONPATH=str(root), PYTHONUNBUFFERED='1')
     result = subprocess.run([sys.executable, str(root / 'tests/gateway/fixtures/ancillary_reads_peer.py'), mode],
-                            cwd=root, env=env, capture_output=True, text=True, timeout=90)
+                            cwd=root, env=env, capture_output=True, text=True, timeout=90, check=False)
     assert result.returncode == 0, result.stdout + result.stderr
     receipt = json.loads((home / 'receipt.json').read_text())
     assert receipt['mode'] == mode and receipt['authenticated_ws']