fix(config): gate the readers main added since the rebase; adopt main's lint rules

Rebasing onto current main brought in direct config.yaml readers that the
seam's reader gate (scripts/check_config_yaml_readers.py) now rejects:

- gateway/session_config.py: the session `config.get mtime` getter uses
  config_exists/config_version + the TUI's _version_token, like
  tui_gateway/methods_config.py already does.
- hermes_cli/profiles.py: _strip_multiplex_flag checks config_exists.
- hermes_cli/config_backups.py: the byte comparison sits inside the
  supports_file_tooling()-gated block; marked like its siblings there.
- hermes_cli/update_receipt.py: _bare_collection_enabled runs in the bare
  bootstrap interpreter with no config stack; presence stat marked.

Main also turned on UP006/RUF100/B009 since the branch base: builtin
generics in the seam's new modules, stale noqa comments dropped, and the
newly unused typing imports removed.
This commit is contained in:
kshitijk4poor authored and Teknium committed 2026-10-25 18:54:09 -07:00
1 parent 79bf625904
commit adebe1f8be
11 files changed
+29 -28

No files matched your search

+4 -2
View File
@@ -81,9 +81,11 @@ async def config_get(connection, ref, params):
def read():
from gateway.run import _profile_runtime_scope
from hermes_cli.config import load_config
from hermes_cli.config_backend import config_exists, config_version
from hermes_cli.config_effective import load_user_config_effective
from hermes_constants import DEFAULT_INDICATOR_STYLE
from tools.approval_context import _get_approval_mode
from tui_gateway.methods_config import _version_token
home = Path(connection.authority.profile_id)
with _profile_runtime_scope(home):
cfg = load_config()
@@ -92,8 +94,8 @@ async def config_get(connection, ref, params):
# The user's own settings, not DEFAULT_CONFIG: clients read an absent key as their
# own default (the TUI streams unless display.streaming is explicitly false).
'full': lambda: {'config': _client_config(load_user_config_effective(home / 'config.yaml'))},
'mtime': lambda: {'mtime': (home / 'config.yaml').stat().st_mtime
if (home / 'config.yaml').exists() else 0,
'mtime': lambda: {'mtime': _version_token(config_version(home / 'config.yaml'))
if config_exists(home / 'config.yaml') else 0,
# Frozen tool policy is not a live MCP reload request.
'mcp_rev': mcp_revision(policy, cfg)},
'reasoning': lambda: _reasoning(agent, policy, cfg, connection, ref),
+3 -3
View File
@@ -18,7 +18,7 @@ import tempfile
import zipfile
from contextlib import closing, contextmanager
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, List, Optional, Tuple
from hermes_state_holders import read_only_db_uri
from hermes_cli.backup_sqlite import _safe_copy_db
@@ -658,7 +658,7 @@ def _import_db_member_exclusive(
pass
def _get_config_path_value(data: Dict[str, Any], dotted: Tuple[str, ...]) -> Any:
def _get_config_path_value(data: dict[str, Any], dotted: tuple[str, ...]) -> Any:
node: Any = data
for key in dotted:
if not isinstance(node, dict):
@@ -667,7 +667,7 @@ def _get_config_path_value(data: Dict[str, Any], dotted: Tuple[str, ...]) -> Any
return node
def _set_config_path_value(data: Dict[str, Any], dotted: Tuple[str, ...], value: Any) -> None:
def _set_config_path_value(data: dict[str, Any], dotted: tuple[str, ...], value: Any) -> None:
node = data
for key in dotted[:-1]:
child = node.get(key)
+2 -2
View File
@@ -44,9 +44,9 @@ from hermes_constants import (
# Re-export from hermes_constants — canonical definition lives there.
from hermes_constants import get_hermes_home, get_process_hermes_home
from utils import atomic_replace, fast_safe_load, file_signature, mkstemp_beside
from hermes_cli.config_issues import ConfigIssue, _issue # noqa: F401 ConfigIssue re-exported
from hermes_cli.config_issues import ConfigIssue, _issue
from hermes_cli.config_backend import (
config_exists, config_version, probe_config_readable, read_config_doc, supports_file_tooling, # noqa: F401 config_exists re-exported for facades
config_exists, config_version, probe_config_readable, read_config_doc, supports_file_tooling,
write_config_document)
from hermes_cli.config_read_errors import (
_CONFIG_PARSE_FAILURES, _FIX_PERMS, _FIX_YAML, FailedConfigRead, _backups_dir_display,
+8 -8
View File
@@ -20,7 +20,7 @@ from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping, Optional, Protocol, Tuple, Union, runtime_checkable
from typing import Any, Mapping, Optional, Protocol, Union, runtime_checkable
CONFIG_FILENAME = "config.yaml"
BACKEND_ENV = "HERMES_CONFIG_BACKEND"
@@ -32,7 +32,7 @@ PathLike = Union[str, "os.PathLike[str]"]
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)
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
@@ -47,7 +47,7 @@ class Changes:
"""
document: Optional[dict] = None
set: Mapping[str, Any] = field(default_factory=dict)
unset: Tuple[str, ...] = ()
unset: tuple[str, ...] = ()
extra_content_on_create: Optional[str] = None
@@ -61,7 +61,7 @@ class ConfigBackend(Protocol):
"""``read_user_layer(home).doc`` re-read only when ``version`` changes; never mutate it."""
...
def version(self, home: Path) -> Tuple[Any, ...]:
def version(self, home: Path) -> tuple[Any, ...]:
"""Cheap cache signature; changes whenever the user layer does."""
...
@@ -93,7 +93,7 @@ class FileBackend:
# Path-level primitives: the file backend's whole behaviour, also used for explicit files
# that are not a home's config.yaml (see ``_route``).
@staticmethod
def version_path(path: Path) -> Tuple[int, int, int, int]:
def version_path(path: Path) -> tuple[int, int, int, int]:
st = path.stat() # before the import: a missing file must not need the utils stack (PM runtime)
from utils import file_signature
return file_signature(st)
@@ -130,7 +130,7 @@ class FileBackend:
def read_user_doc_readonly(self, home: Path) -> Any:
return self.read_path_readonly(self.config_path(home))
def version(self, home: Path) -> Tuple[int, int, int, int]:
def version(self, home: Path) -> tuple[int, int, int, int]:
return self.version_path(self.config_path(home))
def exists(self, home: Path) -> bool:
@@ -196,7 +196,7 @@ class _ExplicitFile(FileBackend):
return self._path
def _route(config_path: PathLike) -> Tuple[ConfigBackend, Path]:
def _route(config_path: PathLike) -> tuple[ConfigBackend, Path]:
"""``(backend, home)`` serving a config path through the one ``ConfigBackend`` surface.
``<home>/config.yaml`` is that home's user layer and goes to the selected backend. Any other
@@ -220,7 +220,7 @@ def read_config_doc_readonly(config_path: PathLike) -> Any:
return backend.read_user_doc_readonly(home)
def config_version(config_path: PathLike) -> Tuple[Any, ...]:
def config_version(config_path: PathLike) -> tuple[Any, ...]:
"""Cache signature of a config file; raises ``FileNotFoundError`` / ``OSError`` like ``stat``."""
backend, home = _route(config_path)
return backend.version(home)
+1 -1
View File
@@ -60,7 +60,7 @@ def backup_config(config_path: Path, reason: str, *, keep: int = DEFAULT_KEEP) -
existing = list_config_backups(config_path, reason)
# Not filecmp.cmp: its process-wide cache keys on (size, mtime), so an equal-size edit
# within one mtime tick of an earlier comparison is reported identical and never backed up.
if existing and config_path.read_bytes() == existing[0].read_bytes():
if existing and config_path.read_bytes() == existing[0].read_bytes(): # config-reader: ok — file tooling, gated above
return None
dest = root / f"{config_path.name}.{reason}.{time.strftime('%Y%m%d-%H%M%S')}"
if dest.is_symlink() or dest.exists(): # never write through a planted link
+1 -2
View File
@@ -4,7 +4,6 @@ Its own module so the facade (``hermes_cli.config``) and its check siblings shar
a facade import from the siblings."""
from dataclasses import dataclass
from typing import List
@dataclass
@@ -15,5 +14,5 @@ class ConfigIssue:
hint: str
def _issue(issues: List[ConfigIssue], severity: str, message: str, hint: str) -> None:
def _issue(issues: list[ConfigIssue], severity: str, message: str, hint: str) -> None:
issues.append(ConfigIssue(severity, message, hint))
+4 -4
View File
@@ -1,12 +1,12 @@
"""Value checks for single config sections (voice, timezone, web backends), run by
``hermes_cli.config.validate_config_structure`` for doctor and the startup warnings."""
from typing import Any, Dict, List
from typing import Any
from hermes_cli.config_issues import ConfigIssue, _issue
def _validate_voice(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
def _validate_voice(config: dict[str, Any], issues: list[ConfigIssue]) -> None:
voice_cfg = config.get("voice")
if not (isinstance(voice_cfg, dict) and "submit_mode" in voice_cfg):
return
@@ -17,7 +17,7 @@ def _validate_voice(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
"Set voice.submit_mode to direct (submit immediately) or draft (edit before sending)")
def _validate_timezone(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
def _validate_timezone(config: dict[str, Any], issues: list[ConfigIssue]) -> None:
"""``timezone`` must be an IANA name the runtime can load.
``hermes_time._get_zoneinfo()`` swallows an invalid name behind a single WARNING in the
@@ -50,7 +50,7 @@ def _validate_timezone(config: Dict[str, Any], issues: List[ConfigIssue]) -> Non
_issue(issues, "error", f"timezone {name!r} is not a valid IANA zone name", hint)
def _validate_web_backends(config: Dict[str, Any], issues: List[ConfigIssue]) -> None:
def _validate_web_backends(config: dict[str, Any], issues: list[ConfigIssue]) -> None:
"""A stale web backend selection otherwise fails only at the first web_search/web_extract
call with a generic "no registered provider" error; warn at startup instead."""
# See #99199.
+2 -2
View File
@@ -25,7 +25,7 @@ from hermes_constants import (
named_profile_has_identity, named_profile_is_deleted, named_profile_is_live,
)
from hermes_cli.config_backend import config_exists, config_version, require_file_tooling
from hermes_cli.profiles_cache import _PROFILE_FILE_CACHE, _cached_profile_read, _config_version_or_none # noqa: F401 _PROFILE_FILE_CACHE: tests clear it here
from hermes_cli.profiles_cache import _PROFILE_FILE_CACHE, _cached_profile_read, _config_version_or_none
logger = logging.getLogger(__name__)
@@ -1262,7 +1262,7 @@ def _clone_all_into(source_dir: Path, profile_dir: Path, canon: str) -> Optional
def _strip_multiplex_flag(config_path: Path) -> None:
"""Multiplexing belongs to the default profile; a clone that keeps the flag would start its
own multiplexer (or be refused at boot) the first time it runs a gateway."""
if not config_path.is_file():
if not config_exists(config_path):
return
with contextlib.suppress(Exception): # creation must not fail over an unreadable copy
from hermes_cli.config import atomic_config_replace, read_user_config_raw
+2 -2
View File
@@ -1,7 +1,7 @@
"""The per-profile derived-value cache behind ``list_profiles`` (``hermes_cli.profiles``)."""
from pathlib import Path
from typing import Dict, Optional
from typing import Optional
from hermes_cli.config_backend import config_version
@@ -12,7 +12,7 @@ from hermes_cli.config_backend import config_version
# template, ~119KB) was re-parsed for every bot every five seconds to yield the same two strings.
# Only DERIVED values are cached, never a document a caller could write back: the raw readers
# (`read_user_config_raw`, `_load_yaml_dict`) keep their uncached contract. See #117378.
_PROFILE_FILE_CACHE: Dict[tuple, tuple] = {}
_PROFILE_FILE_CACHE: dict[tuple, tuple] = {}
_PROFILE_FILE_CACHE_MAX = 512
+1 -1
View File
@@ -807,7 +807,7 @@ def _bare_collection_enabled() -> Optional[bool]:
from hermes_constants import get_hermes_home
try:
(get_hermes_home() / "config.yaml").stat()
(get_hermes_home() / "config.yaml").stat() # config-reader: ok — bare interpreter has no config stack; presence probe only
except FileNotFoundError:
return False
except OSError:
+1 -1
View File
@@ -105,7 +105,7 @@ class _Scanner:
if start is None or end is None:
return ""
# col offsets are UTF-8 byte offsets.
c0, c1 = getattr(node, "col_offset"), getattr(node, "end_col_offset")
c0, c1 = node.col_offset, node.end_col_offset
b = self.blines
if start == end:
return b[start - 1][c0:c1].decode("utf-8", "replace")