chore(config-remote): meet main's ruff rules and the PR's config-reader gate after the rebase

The rebase onto main brought new lint rules (UP006/UP037/RUF100/RUF059/PLW1510/B009) that the
PR's own files did not meet, and main code that reads/writes config.yaml directly, which the PR's
reader gate (scripts/check_config_yaml_readers.py) refuses. No behaviour change in file mode.

- ruff --fix (safe fixes only) on the PR's files; explicit check=False / unused-name fixes in tests.
- gateway/session_config.py (new on main): the multiplexed 'mtime' getter uses config_version /
  config_exists, like tui_gateway's _cfg_get_mtime.
- hermes_cli/prune_advisor.py (new on main): decide() writes sessions.auto_prune through
  config_backend.write_config_key (a remote refusal is the existing "could not save" answer).
- hermes_cli/profiles.py _strip_multiplex_flag (new on main): config_exists, not Path.is_file().
- hermes_cli/config_backups.py / update_receipt.py: annotated as file tooling / pre-swap
  interpreter (both already gated on file tooling or run without the backend importable).

hermes_cli/config_backend.py

@@ -29,6 +29,15 @@ CONFIG_FILENAME = "config.yaml"
 PathLike = Union[str, "os.PathLike[str]"]
 
 
+@dataclass(frozen=True)
+class UserLayer:
+    """One read of a profile's user layer."""
+    doc: Any                    # the parsed root as written (None for an empty document)
+    version: tuple[Any, ...]    # the cache signature this doc was read at (etag for a remote backend)
+    locks: Mapping[str, str] = field(default_factory=dict)   # dotted prefix -> level that locks it
+    provenance: str = ""        # where the doc came from, for diagnostics
+
+
 @dataclass(frozen=True)
 class Changes:
     """A write to one profile's user layer.
@@ -398,8 +407,8 @@ def write_config_document(config_path: PathLike, document: dict, *, extra_conten
 def config_destination(config_path: PathLike) -> str:
     """Where a write to *config_path* lands, for "✓ Set ... in <here>" lines: the file path for
     the file backend and explicit files, the backend's own name for the user layer otherwise."""
-    backend, target, is_layer = _route(config_path)
-    return backend.destination(target) if is_layer else str(target)
+    backend, home = _route(config_path)
+    return backend.destination(home)
 
 
 def write_config_key(config_path: PathLike, key_path: str, value: Any) -> None:
@@ -413,10 +422,9 @@ def write_config_keys(config_path: PathLike, updates: Mapping[str, Any]) -> None
     A remote backend sends them as one CAS PATCH, so a refusal of any of them (a lock, a value
     check) changes none of them: a grouped edit such as a model switch (default + provider + ...)
     never lands half-applied. The file backend applies them key by key, as before."""
-    backend, target, is_layer = _route(config_path)
-    changes = Changes(set={k: v for k, v in updates.items() if v is not None},
-                      unset=tuple(k for k, v in updates.items() if v is None))
-    backend.write_changes(target, changes) if is_layer else backend.write_path(target, changes)
+    backend, home = _route(config_path)
+    backend.write_changes(home, Changes(set={k: v for k, v in updates.items() if v is not None},
+                                        unset=tuple(k for k, v in updates.items() if v is None)))
 
 
 def deployment_env_names() -> frozenset:

hermes_cli/config_migrations.py

@@ -32,7 +32,7 @@ SUPPORT_FLOOR_VERSION = 12
 #: document and nothing else: steps that also touch local files (``.env``, profile ``SOUL.md``,
 #: ``logs/``) skip that part, so reading a profile's remote config, or a refused write's check, never
 #: changes a file on disk — a sibling profile's least of all.
-_CONFIG_ONLY: "contextvars.ContextVar[bool]" = contextvars.ContextVar("config_migrations_config_only", default=False)
+_CONFIG_ONLY: contextvars.ContextVar[bool] = contextvars.ContextVar("config_migrations_config_only", default=False)
 
 
 def _local_files_allowed() -> bool:

hermes_cli/prune_advisor.py

@@ -346,7 +346,7 @@ def decide(enable: bool, home: Optional[Path] = None) -> bool:
     """Persist the user's answer as an explicit ``sessions.auto_prune``; ends the prompt for good.
     False when the save failed or did not take effect (an administrator-managed value wins)."""
     import hermes_yaml as yaml
-    from hermes_cli.config_backend import write_config_key
+    from hermes_cli.config_backend import ConfigWriteError, write_config_key
     from hermes_constants import mkdir_under_hermes_home
     if managed():
         return False
@@ -354,7 +354,7 @@ def decide(enable: bool, home: Optional[Path] = None) -> bool:
     try:
         mkdir_under_hermes_home(path.parent)
         write_config_key(path, "sessions.auto_prune", bool(enable))
-    except (OSError, ValueError, yaml.YAMLError) as exc:
+    except (OSError, ValueError, yaml.YAMLError, ConfigWriteError) as exc:
         logger.warning("Could not save sessions.auto_prune: %s", exc)
         return False
     try:

plugins/config_backends/remote/backend.py

@@ -66,19 +66,19 @@ class _FetchFailed(Exception):
 class _ProfileState:
     home: Path
     profile: str
-    server_config: Dict[str, Any] = field(default_factory=dict)  # EffectiveResponse.config, as sent
-    doc: Dict[str, Any] = field(default_factory=dict)            # what readers get (+ _config_version, migrated)
+    server_config: dict[str, Any] = field(default_factory=dict)  # EffectiveResponse.config, as sent
+    doc: dict[str, Any] = field(default_factory=dict)            # what readers get (+ _config_version, migrated)
     etag: str = ""
     profile_version: int = 0
     # The profile level's stored writerConfigVersion: None = stored null, _UNKNOWN = not reported.
     profile_writer: Any = None
-    other_writers: List[int] = field(default_factory=list)  # the other levels' int stamps (R6)
-    locks: List[Tuple[KeyPath, str]] = field(default_factory=list)
-    provenance: Dict[str, str] = field(default_factory=dict)
+    other_writers: list[int] = field(default_factory=list)  # the other levels' int stamps (R6)
+    locks: list[tuple[KeyPath, str]] = field(default_factory=list)
+    provenance: dict[str, str] = field(default_factory=dict)
     changed_ns: int = 0
     gen: int = 0
     installed: int = 0         # bumped by every _install; a GET started before a newer install is stale
-    base: Optional[Dict[str, Any]] = None  # the migrated doc writes diff against (None = server_config)
+    base: Optional[dict[str, Any]] = None  # the migrated doc writes diff against (None = server_config)
     postprocessed: bool = False   # st.doc is the completed migration output of generation `installed`
     fetched_at: float = 0.0
     last_error: Optional[str] = None
@@ -98,17 +98,17 @@ class _Private:
     everyone else keeps seeing the published doc with its matching diff base. The migrated doc and
     its base are published together, in one step, when the migration ends: a concurrent writer
     never diffs a half-migrated document against the unmigrated base, or the reverse."""
-    st: "_ProfileState"
-    doc: Dict[str, Any]
+    st: _ProfileState
+    doc: dict[str, Any]
     serial: int
     gen: int = 0
 
 
-_PRIVATE: "contextvars.ContextVar[Optional[_Private]]" = contextvars.ContextVar("remote_config_private", default=None)
+_PRIVATE: contextvars.ContextVar[Optional[_Private]] = contextvars.ContextVar("remote_config_private", default=None)
 _PRIVATE_SERIAL = itertools.count(1)
 
 
-def _private_for(st: "_ProfileState") -> Optional[_Private]:
+def _private_for(st: _ProfileState) -> Optional[_Private]:
     m = _PRIVATE.get()
     return m if m is not None and m.st is st else None
 
@@ -121,7 +121,7 @@ _VALUE_REFUSALS = frozenset({
 })
 
 
-def _profile_writer(body: Dict[str, Any]) -> Any:
+def _profile_writer(body: dict[str, Any]) -> Any:
     """The profile level's stored ``writerConfigVersion`` (int or None), or ``_UNKNOWN``."""
     for lv in body.get("levels") or []:
         if isinstance(lv, dict) and lv.get("kind") == "profile":
@@ -130,7 +130,7 @@ def _profile_writer(body: Dict[str, Any]) -> Any:
     return _UNKNOWN
 
 
-def _effective_ok(body: Dict[str, Any]) -> bool:
+def _effective_ok(body: dict[str, Any]) -> bool:
     return (isinstance(body.get("config"), dict) and isinstance(body.get("profileVersion"), int)
             and isinstance(body.get("locks"), list))
 
@@ -139,8 +139,8 @@ class RemoteBackend:
     name = "remote"
 
     def __init__(self) -> None:
-        self._states: Dict[str, _ProfileState] = {}
-        self._fetch_locks: Dict[str, threading.Lock] = {}
+        self._states: dict[str, _ProfileState] = {}
+        self._fetch_locks: dict[str, threading.Lock] = {}
         self._lock = threading.Lock()
         self._poller: Optional[threading.Thread] = None
         self._poller_pid: Optional[int] = None
@@ -186,7 +186,7 @@ class RemoteBackend:
         self._ensure_poller()
         return st
 
-    def _roster(self) -> List[_ProfileState]:
+    def _roster(self) -> list[_ProfileState]:
         with self._lock:
             return list(self._states.values())
 
@@ -245,16 +245,16 @@ class RemoteBackend:
         raise _FetchFailed(f"HTTP {resp.status} {resp.error}: {resp.message}", retryable=retryable,
                            retry_after=resp.retry_after, status=resp.status)
 
-    def _install(self, st: _ProfileState, body: Dict[str, Any], etag: Optional[str]) -> None:
+    def _install(self, st: _ProfileState, body: dict[str, Any], etag: Optional[str]) -> None:
         config = body["config"]
         doc = copy.deepcopy(config)
         doc.pop("_config_version", None)  # reserved on the wire (§3.6); never expected here
         # R6: the in-memory schema version is the oldest writer's; migrations run in memory (D12).
-        writer_versions: List[int] = [
+        writer_versions: list[int] = [
             lv["writerConfigVersion"] for lv in body.get("levels") or []
             if isinstance(lv, dict) and isinstance(lv.get("writerConfigVersion"), int)]
         doc["_config_version"] = min(writer_versions) if writer_versions else _latest_config_version()
-        locks: List[Tuple[KeyPath, str]] = []
+        locks: list[tuple[KeyPath, str]] = []
         for entry in body.get("locks") or []:
             try:
                 locks.append((decode(entry["path"]), str(entry["level"])))
@@ -337,7 +337,7 @@ class RemoteBackend:
                 st.postprocessed = True
         return True
 
-    def _run_private(self, st: _ProfileState, doc: Dict[str, Any], current: int) -> Dict[str, Any]:
+    def _run_private(self, st: _ProfileState, doc: dict[str, Any], current: int) -> dict[str, Any]:
         """Run the migrations from schema *current* over a private copy of *doc* and return it.
         Nothing is published: only this thread sees the copy while the steps run (:class:`_Private`)."""
         from hermes_cli.config_migrations import run_migrations
@@ -354,7 +354,7 @@ class RemoteBackend:
             _PRIVATE.reset(private_token)
         return m.doc
 
-    def _migrate_in_memory(self, st: _ProfileState, started: int, current: int, doc: Dict[str, Any]) -> None:
+    def _migrate_in_memory(self, st: _ProfileState, started: int, current: int, doc: dict[str, Any]) -> None:
         """Migrate doc generation *started* (*doc*, schema version *current*) in memory (D12).
 
         The steps work on a private copy; the result is published with its diff base in one step,
@@ -379,7 +379,7 @@ class RemoteBackend:
         logger.info("Remote Config: migrated profile %r in memory from config version %d", st.profile, current)
 
     @staticmethod
-    def _publish_locked(st: _ProfileState, doc: Dict[str, Any], base: Dict[str, Any]) -> None:
+    def _publish_locked(st: _ProfileState, doc: dict[str, Any], base: dict[str, Any]) -> None:
         st.doc = doc
         st.base = base
         st.changed_ns = time.time_ns()
@@ -421,7 +421,7 @@ class RemoteBackend:
         return self._snapshot(st, take)
 
     @staticmethod
-    def _version_of(st: _ProfileState) -> Tuple[Any, ...]:
+    def _version_of(st: _ProfileState) -> tuple[Any, ...]:
         # Leads with a nanosecond timestamp like the file backend's st_mtime_ns (the TUI's config
         # watcher reads element 0 as a time); the etag and local generation make it exact. A
         # migration's own reads see its private copy under a version of their own, so no config
@@ -432,7 +432,7 @@ class RemoteBackend:
             return (-m.serial, m.gen, st.changed_ns, st.etag, st.gen)
         return (st.changed_ns, st.etag, st.gen)
 
-    def version(self, home: Path) -> Tuple[Any, ...]:
+    def version(self, home: Path) -> tuple[Any, ...]:
         return self._version_of(self._state(home))
 
     def exists(self, home: Path) -> bool:
@@ -458,7 +458,7 @@ class RemoteBackend:
         re-read doc, with the re-read locks; it never replays the caller's whole document, which
         would turn every key another writer changed meanwhile into an edit of ours."""
         st = self._state(home)
-        intent: Optional[Tuple[Dict[KeyPath, Any], List[KeyPath]]] = None
+        intent: Optional[tuple[dict[KeyPath, Any], list[KeyPath]]] = None
         for attempt in (1, 2):
             for _ in range(self._MAX_PREPARE):
                 # Diff against the doc readers get, i.e. migrated in memory (D12) — also after the
@@ -501,12 +501,12 @@ class RemoteBackend:
                 raise ConfigWriteError("Remote Config write not sent: the profile's config kept changing "
                                        "while the write was prepared", code="config_version_conflict")
 
-    def _diff_base(self, st: _ProfileState) -> Dict[str, Any]:
+    def _diff_base(self, st: _ProfileState) -> dict[str, Any]:
         base = copy.deepcopy(st.base if st.base is not None else st.server_config)
         base.pop("_config_version", None)
         return base
 
-    def _intent(self, st: _ProfileState, changes: Changes) -> Tuple[Dict[KeyPath, Any], List[KeyPath]]:
+    def _intent(self, st: _ProfileState, changes: Changes) -> tuple[dict[KeyPath, Any], list[KeyPath]]:
         """The key-level edit ``changes`` makes to the last read doc (§15.2 steps 1-5)."""
         base = self._diff_base(st)
         if changes.document is not None:
@@ -531,8 +531,8 @@ class RemoteBackend:
         # retry after 409 keeps the siblings another writer put in that section meanwhile.
         return intent_diff(base, new)
 
-    def _patch_body(self, st: _ProfileState, sets: Dict[KeyPath, Any], unsets: List[KeyPath]
-                    ) -> Optional[Tuple[Dict[str, Any], Dict[KeyPath, Any], List[KeyPath]]]:
+    def _patch_body(self, st: _ProfileState, sets: dict[KeyPath, Any], unsets: list[KeyPath]
+                    ) -> Optional[tuple[dict[str, Any], dict[KeyPath, Any], list[KeyPath]]]:
         """The PATCH body applying the intent to the current doc, or None when that is no change."""
         base = self._diff_base(st)
         new = copy.deepcopy(base)
@@ -552,7 +552,7 @@ class RemoteBackend:
         if not sets and not unsets:
             return None
         set_body, unset_body = encode_changes(sets, unsets)
-        body: Dict[str, Any] = {"expectedVersion": st.profile_version}
+        body: dict[str, Any] = {"expectedVersion": st.profile_version}
         # The stamp describes the profile level's STORED data (contract §5.3: absent = unchanged).
         # Stamp only when that data is already at this agent's version, or unstamped (null: readers
         # already treat it as current, R6). An older stamp means the level still holds pre-migration
@@ -567,8 +567,8 @@ class RemoteBackend:
             body["unset"] = unset_body
         return body, sets, unsets
 
-    def _migration_check_input(self, st: _ProfileState, body: Dict[str, Any], sets: Dict[KeyPath, Any],
-                               unsets: List[KeyPath]) -> Optional[Tuple[Dict[str, Any], int]]:
+    def _migration_check_input(self, st: _ProfileState, body: dict[str, Any], sets: dict[KeyPath, Any],
+                               unsets: list[KeyPath]) -> Optional[tuple[dict[str, Any], int]]:
         """``(stored doc after this write, its schema version)`` when readers will migrate it in
         memory after the write (R6: the oldest level's stamp), else None. Call under st.lock."""
         if not sets:
@@ -591,8 +591,8 @@ class RemoteBackend:
             _pop_path(stored, p)
         return stored, after
 
-    def _check_migration_keeps(self, st: _ProfileState, sets: Dict[KeyPath, Any],
-                               stored: Dict[str, Any], after: int) -> None:
+    def _check_migration_keeps(self, st: _ProfileState, sets: dict[KeyPath, Any],
+                               stored: dict[str, Any], after: int) -> None:
         """Refuse a write whose values every later read would migrate away.
 
         The plane stores the write as sent, but under a level stamp older than this agent's schema
@@ -634,7 +634,7 @@ class RemoteBackend:
                                 code=resp.error)
 
     @staticmethod
-    def _warn_inherited(st: _ProfileState, unsets: List[KeyPath]) -> None:
+    def _warn_inherited(st: _ProfileState, unsets: list[KeyPath]) -> None:
         """R5: removing a key at the profile level cannot remove a value an upper level supplies."""
         for p in unsets:
             node: Any = st.server_config
@@ -719,7 +719,7 @@ class RemoteBackend:
                     if st.next_poll <= now:
                         self.poll_one(st)
                 wake = min((st.next_poll for st in self._roster()), default=now + poll_interval())
-            except Exception:  # noqa: BLE001 — the only poller thread must survive anything
+            except Exception:
                 logger.warning("Remote Config: poll loop error; retrying", exc_info=True)
                 wake = time.monotonic() + MIN_POLL_SECONDS
             self._stop.wait(max(1.0, wake - time.monotonic()))
@@ -736,7 +736,7 @@ class RemoteBackend:
             logger.log(level, "Remote Config: poll for profile %r failed (%s); keeping the loaded config",
                        st.profile, exc)
             changed = False
-        except Exception as exc:  # noqa: BLE001 — the poller thread must survive anything
+        except Exception as exc:
             st.last_error = str(exc)
             logger.warning("Remote Config: poll for profile %r failed", st.profile, exc_info=True)
             changed = False
@@ -775,7 +775,7 @@ def _first_change(path: KeyPath, sent: Any, found: bool, got: Any) -> Optional[K
         return path
 
 
-def _set_path(doc: Dict[str, Any], path: KeyPath, value: Any) -> None:
+def _set_path(doc: dict[str, Any], path: KeyPath, value: Any) -> None:
     node = doc
     for seg in path[:-1]:
         if not isinstance(node.get(seg), dict):
@@ -784,7 +784,7 @@ def _set_path(doc: Dict[str, Any], path: KeyPath, value: Any) -> None:
     node[path[-1]] = value
 
 
-def _pop_path(doc: Dict[str, Any], path: KeyPath) -> None:
+def _pop_path(doc: dict[str, Any], path: KeyPath) -> None:
     node: Any = doc
     for seg in path[:-1]:
         node = node.get(seg) if isinstance(node, dict) else None

plugins/config_backends/remote/client.py

@@ -26,7 +26,7 @@ _HTTP_TIMEOUT_S = 15.0
 @dataclass(frozen=True)
 class Response:
     status: int
-    body: Dict[str, Any]
+    body: dict[str, Any]
     etag: Optional[str]
     retry_after: Optional[float]
 
@@ -53,7 +53,7 @@ def instance_id() -> str:
     return os.environ.get(INSTANCE_ENV, "").strip()
 
 
-def _parse_body(raw: bytes) -> Dict[str, Any]:
+def _parse_body(raw: bytes) -> dict[str, Any]:
     if not raw:
         return {}
     try:
@@ -71,7 +71,7 @@ def _retry_after(value: Optional[str]) -> Optional[float]:
 
 
 def request(method: str, home: Path, profile: str, *, etag: Optional[str] = None,
-            body: Optional[Dict[str, Any]] = None) -> Response:
+            body: Optional[dict[str, Any]] = None) -> Response:
     """One call to ``/v1/config/self?profile=<profile>`` as ``home``'s agent.
 
     The destination is checked before the bearer is resolved, and the request never follows a

plugins/config_backends/remote/credentials.py

@@ -41,11 +41,11 @@ class PlaneCredentialError(RuntimeError):
         self.retryable = retryable
 
 
-_IDP_CACHE: Dict[Tuple[str, str, str], Tuple[float, str]] = {}
+_IDP_CACHE: dict[tuple[str, str, str], tuple[float, str]] = {}
 _IDP_LOCK = threading.Lock()
 
 
-def _idp_env() -> Dict[str, str]:
+def _idp_env() -> dict[str, str]:
     return {k: os.environ.get(IDP_ENV_PREFIX + k.upper(), "").strip() for k in _IDP_KEYS}
 
 
@@ -63,12 +63,12 @@ def _nous_token(home: Path) -> str:
 
     try:
         from hermes_cli.auth import resolve_nous_access_token
-    except Exception as exc:  # noqa: BLE001 — an import failure is a broken install, not transient
+    except Exception as exc:
         raise PlaneCredentialError(f"cannot load the Nous auth module: {exc}", retryable=False) from exc
     token = set_hermes_home_override(home)  # auth.json is per profile home
     try:
         return resolve_nous_access_token()
-    except Exception as exc:  # noqa: BLE001 — AuthError and network errors alike
+    except Exception as exc:
         # A missing login or a revoked session needs a human; a refresh that hit the network can retry.
         fatal = bool(getattr(exc, "relogin_required", False)) or getattr(exc, "code", None) == "nous_auth_missing"
         raise PlaneCredentialError(
@@ -104,7 +104,7 @@ def _idp_token() -> str:
             payload = json.loads(resp.read().decode() or "{}")
     except CredentialDestinationError:
         raise
-    except Exception as exc:  # noqa: BLE001 — HTTP and network errors alike; the message has no secret
+    except Exception as exc:  # HTTP and network errors alike; the message has no secret
         raise PlaneCredentialError(f"IdP token request to {token_url} failed: {exc}", retryable=True) from exc
     access_token = payload.get("access_token") if isinstance(payload, dict) else None
     if not isinstance(access_token, str) or not access_token:

plugins/config_backends/remote/destination.py

@@ -41,7 +41,7 @@ def checked_url(url: str, *, env_name: str) -> str:
     try:
         parts = urllib.parse.urlsplit(url)
         host = (parts.hostname or "").lower()
-        parts.port  # noqa: B018 — raises ValueError on a malformed port
+        _ = parts.port  # raises ValueError on a malformed port
     except ValueError as exc:
         raise CredentialDestinationError(f"{env_name}={url!r} is not a valid URL: {exc}") from exc
     if not host:

plugins/config_backends/remote/diff.py

@@ -11,7 +11,7 @@ from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple
 
 from .paths import Path, covers, encode
 
-Locks = Sequence[Tuple[Path, str]]
+Locks = Sequence[tuple[Path, str]]
 
 
 def json_equal(a: Any, b: Any) -> bool:
@@ -27,11 +27,11 @@ def json_equal(a: Any, b: Any) -> bool:
     return type(a) is type(b) and a == b
 
 
-def diff(base: Dict[str, Any], new: Dict[str, Any], prefix: Path = ()) -> Tuple[Dict[Path, Any], List[Path]]:
+def diff(base: dict[str, Any], new: dict[str, Any], prefix: Path = ()) -> tuple[dict[Path, Any], list[Path]]:
     """``(set, unset)`` turning ``base`` into ``new`` (§15.2 step 5). A changed list is one
     whole-list ``set`` (D8); a key equal to its base value is not sent."""
-    sets: Dict[Path, Any] = {}
-    unsets: List[Path] = []
+    sets: dict[Path, Any] = {}
+    unsets: list[Path] = []
     for key, value in new.items():
         p = prefix + (key,)
         if key not in base:
@@ -46,7 +46,7 @@ def diff(base: Dict[str, Any], new: Dict[str, Any], prefix: Path = ()) -> Tuple[
     return sets, unsets
 
 
-def intent_diff(base: Dict[str, Any], new: Dict[str, Any], prefix: Path = ()) -> Tuple[Dict[Path, Any], List[Path]]:
+def intent_diff(base: dict[str, Any], new: dict[str, Any], prefix: Path = ()) -> tuple[dict[Path, Any], list[Path]]:
     """:func:`diff` as a caller's *intent*: the key-level edit a write means, to be re-applied to a
     newer document after a CAS conflict (§15.3). A mapping the base did not hold as a mapping
     (absent, or a scalar) is expanded to its leaves: writing ``terminal.timeout`` into a profile
@@ -54,8 +54,8 @@ def intent_diff(base: Dict[str, Any], new: Dict[str, Any], prefix: Path = ()) ->
     the time the retry runs" — another writer's ``terminal.persistent`` must survive it. Only an
     empty mapping stays a whole value (there is no leaf to name); :func:`apply_intent` adds it
     without replacing a mapping that appeared meanwhile."""
-    sets: Dict[Path, Any] = {}
-    unsets: List[Path] = []
+    sets: dict[Path, Any] = {}
+    unsets: list[Path] = []
     for key, value in new.items():
         p = prefix + (key,)
         old = base.get(key, _ABSENT)
@@ -77,7 +77,7 @@ def intent_diff(base: Dict[str, Any], new: Dict[str, Any], prefix: Path = ()) ->
 _ABSENT = object()
 
 
-def apply_intent(doc: Dict[str, Any], sets: Dict[Path, Any], unsets: Sequence[Path]) -> None:
+def apply_intent(doc: dict[str, Any], sets: dict[Path, Any], unsets: Sequence[Path]) -> None:
     """Apply an :func:`intent_diff` edit to *doc* in place."""
     for p, v in sets.items():
         if isinstance(v, dict) and not v:
@@ -89,7 +89,7 @@ def apply_intent(doc: Dict[str, Any], sets: Dict[Path, Any], unsets: Sequence[Pa
         _pop_path(doc, p)
 
 
-def _set_path(doc: Dict[str, Any], path: Path, value: Any) -> None:
+def _set_path(doc: dict[str, Any], path: Path, value: Any) -> None:
     node = doc
     for seg in path[:-1]:
         if not isinstance(node.get(seg), dict):
@@ -98,7 +98,7 @@ def _set_path(doc: Dict[str, Any], path: Path, value: Any) -> None:
     node[path[-1]] = value
 
 
-def _pop_path(doc: Dict[str, Any], path: Path) -> bool:
+def _pop_path(doc: dict[str, Any], path: Path) -> bool:
     node: Any = doc
     for seg in path[:-1]:
         if not isinstance(node, dict) or not isinstance(node.get(seg), dict):
@@ -110,7 +110,7 @@ def _pop_path(doc: Dict[str, Any], path: Path) -> bool:
     return False
 
 
-def _get_path(doc: Dict[str, Any], path: Path) -> Tuple[bool, Any]:
+def _get_path(doc: dict[str, Any], path: Path) -> tuple[bool, Any]:
     node: Any = doc
     for seg in path:
         if not isinstance(node, dict) or seg not in node:
@@ -119,12 +119,12 @@ def _get_path(doc: Dict[str, Any], path: Path) -> Tuple[bool, Any]:
     return True, node
 
 
-def strip_locked(base: Dict[str, Any], new: Dict[str, Any], locks: Locks) -> Tuple[Dict[str, Any], Dict[str, Any], List[Path]]:
+def strip_locked(base: dict[str, Any], new: dict[str, Any], locks: Locks) -> tuple[dict[str, Any], dict[str, Any], list[Path]]:
     """Drop every locked path from BOTH documents (§15.2 step 1), so a bulk write neither sets
     nor unsets a key it may not touch. Returns the copies and the locked paths whose new value
     differed from the effective one (the ones the caller tried to change)."""
     base, new = copy.deepcopy(base), copy.deepcopy(new)
-    changed: List[Path] = []
+    changed: list[Path] = []
     for lock, _level in locks:
         in_base, base_val = _get_path(base, lock)
         in_new, new_val = _get_path(new, lock)
@@ -135,7 +135,7 @@ def strip_locked(base: Dict[str, Any], new: Dict[str, Any], locks: Locks) -> Tup
     return base, new, changed
 
 
-def _touched(path: Path, value: Any) -> Iterator[Tuple[Path, Any]]:
+def _touched(path: Path, value: Any) -> Iterator[tuple[Path, Any]]:
     """``(touched path, value written there)`` for ``set path = value`` (§6.4 rule 2): the path
     itself and every path inside ``value``, descending through objects only."""
     yield path, value
@@ -144,7 +144,7 @@ def _touched(path: Path, value: Any) -> Iterator[Tuple[Path, Any]]:
             yield from _touched(path + (k,), v)
 
 
-def write_check(sets: Dict[Path, Any], unsets: Sequence[Path], locks: Locks) -> Optional[Tuple[Path, str]]:
+def write_check(sets: dict[Path, Any], unsets: Sequence[Path], locks: Locks) -> Optional[tuple[Path, str]]:
     """The ``(lock path, level)`` that refuses this write (§6.4 rules 1-3), or None."""
     for p in unsets:
         for lock, level in locks:
@@ -162,5 +162,5 @@ def write_check(sets: Dict[Path, Any], unsets: Sequence[Path], locks: Locks) ->
     return None
 
 
-def encode_changes(sets: Dict[Path, Any], unsets: Sequence[Path]) -> Tuple[Dict[str, Any], List[str]]:
+def encode_changes(sets: dict[Path, Any], unsets: Sequence[Path]) -> tuple[dict[str, Any], list[str]]:
     return {encode(p): v for p, v in sets.items()}, [encode(p) for p in unsets]

plugins/config_backends/remote/paths.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 from typing import Iterable, List, Optional, Sequence, Tuple
 
-Path = Tuple[str, ...]
+Path = tuple[str, ...]
 
 
 class PathError(ValueError):
@@ -20,8 +20,8 @@ def encode(segments: Sequence[str]) -> str:
 
 
 def decode(encoded: str) -> Path:
-    segments: List[str] = []
-    current: List[str] = []
+    segments: list[str] = []
+    current: list[str] = []
     i = 0
     while i < len(encoded):
         ch = encoded[i]
@@ -72,7 +72,7 @@ def covers(lock: Sequence[str], path: Sequence[str]) -> bool:
     return len(lock) <= len(path) and tuple(path[:len(lock)]) == tuple(lock)
 
 
-def covering_lock(locks: Iterable[Tuple[Path, str]], path: Sequence[str]) -> Optional[Tuple[Path, str]]:
+def covering_lock(locks: Iterable[tuple[Path, str]], path: Sequence[str]) -> Optional[tuple[Path, str]]:
     """The first ``(lock, level)`` whose lock covers ``path``, or None."""
     for lock, level in locks:
         if covers(lock, path):

plugins/config_backends/remote/values.py

@@ -47,7 +47,7 @@ def to_wire(value: Any, path: Path = ()) -> Any:
     if isinstance(value, list | tuple):
         return [to_wire(v, path) for v in value]
     if isinstance(value, dict):
-        out: Dict[str, Any] = {}
+        out: dict[str, Any] = {}
         for k, v in value.items():
             key = _wire_key(k, path)
             if key in out:
@@ -72,7 +72,7 @@ def _wire_key(key: Any, path: Path) -> str:
 # --- secret-shaped keys (§14) -----------------------------------------------------------------
 
 @lru_cache(maxsize=1)
-def _secret_rules() -> Dict[str, Any]:
+def _secret_rules() -> dict[str, Any]:
     data = json.loads(SECRET_KEYS_FIXTURE.read_text(encoding="utf-8-sig"))
     return {
         "secret_keys": frozenset(data["secretKeys"]),
@@ -88,7 +88,7 @@ def _secret_rules() -> Dict[str, Any]:
     }
 
 
-def _is_env_config_key(key: str, rules: Dict[str, Any]) -> bool:
+def _is_env_config_key(key: str, rules: dict[str, Any]) -> bool:
     if "." in key:
         return False
     up = key.upper()
@@ -107,7 +107,7 @@ def is_secret_path(segments: Sequence[str]) -> bool:
     return leaf in rules["secret_keys"] or leaf.endswith(rules["secret_suffixes"])
 
 
-def _walk(value: Any, path: Path) -> Iterator[Tuple[Path, Any]]:
+def _walk(value: Any, path: Path) -> Iterator[tuple[Path, Any]]:
     yield path, value
     if isinstance(value, dict):
         for k, v in value.items():

tests/plugins/config_backends/stub_plane.py

@@ -30,7 +30,7 @@ CLIENT_SECRET = "agent-secret"
 INSTANCE = "inst-0001"
 
 
-def deep_merge(base: Dict[str, Any], over: Dict[str, Any]) -> Dict[str, Any]:
+def deep_merge(base: dict[str, Any], over: dict[str, Any]) -> dict[str, Any]:
     """contract.md §6.2: lists replace, ``null`` over a mapping is ignored, ``null`` over anything
     else sets ``null``."""
     out = copy.deepcopy(base)
@@ -56,15 +56,15 @@ def _depth(value: Any) -> int:
 
 class StubPlane:
     def __init__(self) -> None:
-        self.upper: Dict[str, Any] = {}
-        self.upper_locks: List[Dict[str, str]] = []
+        self.upper: dict[str, Any] = {}
+        self.upper_locks: list[dict[str, str]] = []
         # portal-groups contract §6.7: a member's group levels ({"groupId", "priority", "version"}) and
         # groupProvenance, emitted as-is so tests prove the agent ignores fields it does not use.
-        self.groups: List[Dict[str, Any]] = []
-        self.group_provenance: Optional[Dict[str, Any]] = None
-        self.profiles: Dict[str, Dict[str, Any]] = {}
+        self.groups: list[dict[str, Any]] = []
+        self.group_provenance: Optional[dict[str, Any]] = None
+        self.profiles: dict[str, dict[str, Any]] = {}
         self.fail_status: Optional[int] = None   # every /self request returns this
-        self.requests: List[Dict[str, Any]] = []
+        self.requests: list[dict[str, Any]] = []
         self.token_requests = 0
         self.home: Optional[Path] = None  # the test's HERMES_HOME, set by the fixture
         self.lock = threading.Lock()
@@ -72,7 +72,7 @@ class StubPlane:
         self.url = f"http://127.0.0.1:{self._server.server_address[1]}"
         self._thread = threading.Thread(target=self._server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True)
 
-    def __enter__(self) -> "StubPlane":
+    def __enter__(self) -> StubPlane:
         self._thread.start()
         return self
 
@@ -82,10 +82,10 @@ class StubPlane:
 
     # --- model ---
 
-    def profile(self, name: str) -> Dict[str, Any]:
+    def profile(self, name: str) -> dict[str, Any]:
         return self.profiles.setdefault(name, {"values": {}, "version": 0, "writer": None})
 
-    def effective(self, name: str) -> Dict[str, Any]:
+    def effective(self, name: str) -> dict[str, Any]:
         prof = self.profile(name)
         body = {
             "instanceId": INSTANCE, "profile": name,
@@ -103,7 +103,7 @@ class StubPlane:
         body["etag"] = f'"{digest}"'
         return body
 
-    def patches(self) -> List[Dict[str, Any]]:
+    def patches(self) -> list[dict[str, Any]]:
         return [r for r in self.requests if r["method"] == "PATCH"]
 
     # --- HTTP ---
@@ -112,10 +112,10 @@ class StubPlane:
         plane = self
 
         class Handler(BaseHTTPRequestHandler):
-            def log_message(self, format, *args) -> None:  # noqa: A002 — base signature
+            def log_message(self, format, *args) -> None:
                 pass
 
-            def _send(self, status: int, body: Optional[Dict[str, Any]] = None, etag: Optional[str] = None):
+            def _send(self, status: int, body: Optional[dict[str, Any]] = None, etag: Optional[str] = None):
                 raw = json.dumps(body).encode() if body is not None else b""
                 self.send_response(status)
                 self.send_header("Content-Type", "application/json")
@@ -125,7 +125,7 @@ class StubPlane:
                 self.end_headers()
                 self.wfile.write(raw)
 
-            def _body(self) -> Dict[str, Any]:
+            def _body(self) -> dict[str, Any]:
                 n = int(self.headers.get("Content-Length") or 0)
                 return json.loads(self.rfile.read(n) or b"{}")
 
@@ -165,7 +165,7 @@ class StubPlane:
                         return self._send(200, eff, eff["etag"])
                     return self._patch(profile, body or {})
 
-            def _patch(self, profile: str, body: Dict[str, Any]):
+            def _patch(self, profile: str, body: dict[str, Any]):
                 prof = plane.profile(profile)
                 if body.get("expectedVersion") != prof["version"]:
                     return self._send(409, {"error": "config_version_conflict", "message": "stale",
@@ -220,7 +220,7 @@ class StubPlane:
         return Handler
 
 
-def remote_env(plane: StubPlane) -> Dict[str, str]:
+def remote_env(plane: StubPlane) -> dict[str, str]:
     return {
         "HERMES_CONFIG_BACKEND": "remote",
         "HERMES_CONFIG_REMOTE_URL": plane.url,

tests/plugins/config_backends/test_remote_backend.py

@@ -800,7 +800,7 @@ def test_migration_of_a_replaced_doc_does_not_touch_the_replacement(plane, monke
     def read():
         try:
             read_raw_config()
-        except BaseException as exc:  # noqa: BLE001 — surfaced below
+        except BaseException as exc:
             failures.append(exc)
 
     reader = threading.Thread(target=read)
@@ -996,7 +996,7 @@ def test_edit_during_an_in_memory_migration_never_sends_the_migration(plane, mon
     def migrate():  # a lock-free reader (read_config_doc) starts the migration
         try:
             read_config_doc(plane.home / "config.yaml")
-        except BaseException as exc:  # noqa: BLE001 — surfaced below
+        except BaseException as exc:
             failures.append(exc)
 
     migrator = threading.Thread(target=migrate, name="migrator")
@@ -1119,7 +1119,7 @@ def test_child_for_another_profile_keeps_the_remote_deployment(plane, monkeypatc
             "from hermes_cli.config import load_config\n"
             "print('RESULT=' + json.dumps([get_config_backend().name, load_config()['display']['personality']]))\n")
     proc = subprocess.run([sys.executable, "-c", code], env=child_env, capture_output=True, text=True,
-                          timeout=120, stdin=subprocess.DEVNULL, cwd=str(tmp_path))
+                          timeout=120, stdin=subprocess.DEVNULL, cwd=str(tmp_path), check=False)
     assert proc.returncode == 0, proc.stderr[-3000:]
     line = [ln for ln in (proc.stdout + proc.stderr).splitlines() if ln.startswith("RESULT=")][-1]
     assert json.loads(line[len("RESULT="):]) == ["remote", "remote-child"]
@@ -1223,7 +1223,7 @@ def test_roster_model_follows_the_remote_layer_not_the_local_file(profile_plane)
 
 
 def _tui_server(monkeypatch, home):
-    import hermes_cli.banner as banner
+    from hermes_cli import banner
     monkeypatch.setattr(banner, "prefetch_update_check", lambda: None)
     from tui_gateway import server
     monkeypatch.setattr(server, "_hermes_home", home)
@@ -1324,7 +1324,7 @@ def test_cas_retry_keeps_a_sibling_added_to_a_section_the_doc_lacked(plane, monk
         write_config_key(plane.home / "config.yaml", "terminal.timeout", 321)
 
     assert plane.profile("default")["values"]["terminal"] == {"persistent": True, "timeout": 321}
-    first, retry = plane.patches()
+    _first, retry = plane.patches()
     assert retry["body"]["expectedVersion"] == 2
     assert retry["body"]["set"] == {"terminal.timeout": 321} and "unset" not in retry["body"]
 
@@ -1551,7 +1551,7 @@ def _fresh_interpreter_boot(tmp_path, env, project_env):
     child = {k: v for k, v in os.environ.items() if k in ("PATH", "LANG", "TMPDIR", "SYSTEMROOT")}
     child.update(env, PYTHONPATH=str(Path(__file__).resolve().parents[3]))
     return subprocess.run([sys.executable, "-c", code], env=child, capture_output=True, text=True,
-                          timeout=120, stdin=subprocess.DEVNULL, cwd=str(tmp_path))
+                          timeout=120, stdin=subprocess.DEVNULL, cwd=str(tmp_path), check=False)
 
 
 def test_plane_credential_from_the_project_dotenv_boots(plane, tmp_path):

tests/plugins/config_backends/test_remote_credential_destination.py

@@ -31,7 +31,7 @@ class _Listener:
         listener = self
 
         class Handler(BaseHTTPRequestHandler):
-            def log_message(self, format, *args) -> None:  # noqa: A002 — base signature
+            def log_message(self, format, *args) -> None:
                 pass
 
             def _any(self):

tests/plugins/config_backends/test_remote_grouped_and_migration_reads.py

@@ -40,7 +40,7 @@ def plane(tmp_path, monkeypatch):
 
 
 def _tui_server(monkeypatch, home):
-    import hermes_cli.banner as banner
+    from hermes_cli import banner
     monkeypatch.setattr(banner, "prefetch_update_check", lambda: None)
     from tui_gateway import server
     monkeypatch.setattr(server, "_hermes_home", home)
@@ -187,7 +187,7 @@ def _start(target, name):
     def run():
         try:
             results.append(target())
-        except BaseException as exc:  # noqa: BLE001 — surfaced by the caller
+        except BaseException as exc:
             failures.append(exc)
 
     t = threading.Thread(target=run, name=name)

tests/plugins/config_backends/test_remote_multiprocess.py

@@ -25,7 +25,7 @@ print("RESULT=" + json.dumps(load_config()["display"]["personality"]))
 
 def _run(args, env, timeout=120):
     return subprocess.run(args, cwd=REPO, env=env, capture_output=True, text=True, timeout=timeout,
-                          stdin=subprocess.DEVNULL)
+                          stdin=subprocess.DEVNULL, check=False)
 
 
 @pytest.fixture
@@ -69,7 +69,7 @@ def test_cli_gateway_and_dashboard_share_the_remote_config(plane_env):
 
 
 def test_process_exits_non_zero_when_plane_is_down(plane_env, monkeypatch):
-    plane, env, home = plane_env
+    plane, env, _home = plane_env
     plane.fail_status = 403  # a refusal is not retried, so this is fast
     proc = _run([sys.executable, "-m", "hermes_cli.main", "config", "get", "display.personality"], env)
     assert proc.returncode != 0

tests/tools/test_stage2_hook_remote_config_skip.py

@@ -53,7 +53,7 @@ def _run(home: Path, backend_env: str | None) -> str:
         + _block()
     )
     proc = subprocess.run(["sh", "-c", script], capture_output=True, text=True, timeout=60,
-                          stdin=subprocess.DEVNULL, env=_env(root, backend_env))
+                          stdin=subprocess.DEVNULL, env=_env(root, backend_env), check=False)
     assert proc.returncode == 0, proc.stderr
     assert "could not determine the config backend" not in proc.stdout, proc.stdout
     return proc.stdout
@@ -79,7 +79,7 @@ def _runtime_selects(home: Path, backend_env: str | None) -> str:
     env = _env(home.parent, backend_env)
     env["HERMES_HOME"] = str(home)
     proc = subprocess.run([sys.executable, "-c", code, str(home)], capture_output=True, text=True,
-                          timeout=60, env=env, cwd=str(home.parent))
+                          timeout=60, env=env, cwd=str(home.parent), check=False)
     assert proc.returncode == 0, proc.stderr
     return proc.stdout.strip().splitlines()[-1]