fix(registries): normalize scope keys on write paths; repair windows test gaps

The scope-normalization sweep only fixed the read paths
(list_providers/get_provider/registry_generation) in most registries;
the write paths (register/snapshot/restore) still stored under the raw
scope string while reads resolved through hermes_home_key — so a
scoped registration was written under one key and read under another
(agent/terminal_env_registry.py test_scoped_registration_isolated
failed on both Linux and Windows CI).

- add hermes_constants.normalize_scope(scope): hermes_home_key for
  non-None, None preserved for the process-global layer — one named
  function instead of the expression repeated at ~20 call sites
- apply it on both sides of all nine scoped registries
  (browser, terminal_env, image_gen, transcription, tts, video_gen,
  web_search, secret_sources, dashboard_auth, tools.registry)
- agent/deadline: MAX_SAFE_TIMEOUT_S on win32 leaves 60s headroom under
  the DWORD-ms ceiling so margin-adding consumers (human-wait +60s)
  stay platform-safe; tools/approval fail-closed fallback mirrors it
- tools/approval: strip shell escapes everywhere except inside
  drive-anchored paths (the Windows shell is bash, so r\m still spells
  rm; only C:\... backslashes must survive deobfuscation)
- consolidate the duplicated _rmtree_force into hermes_cli/fs_utils
  and repoint both callers
- hermes_cli/kanban + disk-cleanup: shlex.split(posix=False) keeps
  wrapping quotes in tokens; strip them before dispatch/matching
- test gating: SIGKILL durability conformance cells are linux_only
  (TerminateProcess is not SIGKILL), macOS TCC anchor symlink-tree
  classes are linux_only, Linux chromium profile tests resolve the
  fixture home via expanduser patch + posix-joined expecteds so they
  run identically on every host

agent/browser_registry.py

@@ -41,7 +41,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.browser_provider import BrowserProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -69,7 +69,7 @@ def register_provider(provider: BrowserProvider, *, scope: Optional[str] = None)
     if not isinstance(raw_name, str) or not raw_name.strip():
         raise ValueError("Browser provider .name must be a non-empty string")
     name = raw_name.strip()
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     global _generation
     with _lock:
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
@@ -95,7 +95,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[BrowserProvider]:
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -106,13 +106,13 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[BrowserP
         return None
     with _lock:
         key = name.strip()
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
     name: str, *, scope: Optional[str] = None
 ) -> Optional[BrowserProvider]:
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     with _lock:
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name.strip())
@@ -120,7 +120,7 @@ def snapshot_registration(
 
 def registry_generation(*, scope: Optional[str] = None) -> tuple[int, int]:
     """Return a cache fingerprint for the global base and one profile."""
-    active_scope = hermes_home_key(scope)
+    active_scope = normalize_scope(scope)
     with _lock:
         return _generation, _scoped_generations.get(active_scope, 0)
 
@@ -134,7 +134,7 @@ def restore_registration(
 ) -> bool:
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip()
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     global _generation
     with _lock:
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})

agent/deadline.py

@@ -100,7 +100,10 @@ __all__ = [
 #   * Windows — ``WaitForSingleObject`` takes a DWORD of milliseconds
 #     (2**32 - 1 ms ≈ 49.7 days), so 365 days would overflow it.
 if sys.platform == "win32":
-    MAX_SAFE_TIMEOUT_S = 4_294_967.0  # just under DWORD-millisecond ceiling
+    # Leave headroom for consumers that add a small margin on top (e.g. the
+    # 60s human-wait margin): MAX + margin must stay under the DWORD-millisecond
+    # ceiling (4294967.295s), so 4_294_907 + 60 rounds to exactly 4_294_967s.
+    MAX_SAFE_TIMEOUT_S = 4_294_907.0
 else:
     MAX_SAFE_TIMEOUT_S = 31_536_000.0  # 365 days
 

agent/image_gen_registry.py

@@ -25,7 +25,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.image_gen_provider import ImageGenProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +52,7 @@ def register_provider(provider: ImageGenProvider, *, scope: Optional[str] = None
         raise ValueError("Image gen provider .name must be a non-empty string")
     name = raw_name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(name)
         target[name] = provider
@@ -65,7 +66,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[ImageGenProvider]:
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -76,13 +77,14 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[ImageGen
         return None
     with _lock:
         key = name.strip()
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
     name: str, *, scope: Optional[str] = None
 ) -> Optional[ImageGenProvider]:
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name.strip())
 
@@ -97,6 +99,7 @@ def restore_registration(
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

agent/secret_sources/registry.py

@@ -44,7 +44,7 @@ from agent.secret_sources.base import (
     reset_source_environment,
     set_source_environment,
 )
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -138,7 +138,7 @@ def register_source(
             name, getattr(source, "shape", None),
         )
         return False
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     with _REGISTRY_LOCK:
         effective = dict(_SOURCES)
         if scope is not None:
@@ -170,7 +170,7 @@ def register_source(
 def get_source(name: str, *, scope: Optional[str] = None) -> Optional[SecretSource]:
     _ensure_builtin_sources()
     with _REGISTRY_LOCK:
-        return _SCOPED_SOURCES.get(hermes_home_key(scope), {}).get(
+        return _SCOPED_SOURCES.get(normalize_scope(scope), {}).get(
             name
         ) or _SOURCES.get(name)
 
@@ -180,7 +180,7 @@ def snapshot_registration(
 ) -> Optional[SecretSource]:
     """Return the registration owned by exactly one registry layer."""
     _ensure_builtin_sources()
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     with _REGISTRY_LOCK:
         target = _SOURCES if scope is None else _SCOPED_SOURCES.get(scope, {})
         return target.get(name)
@@ -195,7 +195,7 @@ def restore_registration(
 ) -> bool:
     """Restore a host-owned source registration if it is still current."""
     _ensure_builtin_sources()
-    scope = hermes_home_key(scope) if scope is not None else None
+    scope = normalize_scope(scope)
     with _REGISTRY_LOCK:
         target = _SOURCES if scope is None else _SCOPED_SOURCES.setdefault(scope, {})
         if target.get(name) is not current:
@@ -213,7 +213,7 @@ def list_sources(*, scope: Optional[str] = None) -> List[SecretSource]:
     _ensure_builtin_sources()
     with _REGISTRY_LOCK:
         merged = dict(_SOURCES)
-        merged.update(_SCOPED_SOURCES.get(hermes_home_key(scope), {}))
+        merged.update(_SCOPED_SOURCES.get(normalize_scope(scope), {}))
         return list(merged.values())
 
 

agent/terminal_env_registry.py

@@ -30,7 +30,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.terminal_env_provider import TerminalEnvironmentProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -78,6 +78,7 @@ def register_provider(
         )
     global _generation
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(name)
         target[name] = provider
@@ -101,7 +102,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[TerminalEnvironmentPr
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -115,7 +116,7 @@ def get_provider(
     key = name.strip().lower()
     with _lock:
         return (
-            _scoped_providers.get(hermes_home_key(scope), {}).get(key)
+            _scoped_providers.get(normalize_scope(scope), {}).get(key)
             or _providers.get(key)
         )
 
@@ -173,13 +174,14 @@ def snapshot_registration(
     name: str, *, scope: Optional[str] = None
 ) -> Optional[TerminalEnvironmentProvider]:
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name.strip().lower())
 
 
 def registry_generation(*, scope: Optional[str] = None) -> tuple:
     """Return a cache fingerprint for the global base and one profile."""
-    active_scope = hermes_home_key(scope)
+    active_scope = normalize_scope(scope)
     with _lock:
         return _generation, _scoped_generations.get(active_scope, 0)
 
@@ -195,6 +197,7 @@ def restore_registration(
     key = name.strip().lower()
     global _generation
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

agent/transcription_registry.py

@@ -24,7 +24,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.transcription_provider import TranscriptionProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -87,6 +87,7 @@ def register_provider(provider: TranscriptionProvider, *, scope: Optional[str] =
         )
         return
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(key)
         target[key] = provider
@@ -106,7 +107,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[TranscriptionProvider
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -122,7 +123,7 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[Transcri
         return None
     key = name.strip().lower()
     with _lock:
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
@@ -130,6 +131,7 @@ def snapshot_registration(
 ) -> Optional[TranscriptionProvider]:
     key = name.strip().lower()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(key)
 
@@ -144,6 +146,7 @@ def restore_registration(
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip().lower()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

agent/tts_registry.py

@@ -33,7 +33,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.tts_provider import TTSProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -97,6 +97,7 @@ def register_provider(provider: TTSProvider, *, scope: Optional[str] = None) ->
         )
         return
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(key)
         target[key] = provider
@@ -116,7 +117,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[TTSProvider]:
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -132,7 +133,7 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[TTSProvi
         return None
     key = name.strip().lower()
     with _lock:
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
@@ -140,6 +141,7 @@ def snapshot_registration(
 ) -> Optional[TTSProvider]:
     key = name.strip().lower()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(key)
 
@@ -154,6 +156,7 @@ def restore_registration(
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip().lower()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

agent/video_gen_registry.py

@@ -29,7 +29,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.video_gen_provider import VideoGenProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -56,6 +56,7 @@ def register_provider(provider: VideoGenProvider, *, scope: Optional[str] = None
         raise ValueError("Video gen provider .name must be a non-empty string")
     name = raw_name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(name)
         target[name] = provider
@@ -69,7 +70,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[VideoGenProvider]:
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -80,13 +81,14 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[VideoGen
         return None
     with _lock:
         key = name.strip()
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
     name: str, *, scope: Optional[str] = None
 ) -> Optional[VideoGenProvider]:
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name.strip())
 
@@ -101,6 +103,7 @@ def restore_registration(
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

agent/web_search_registry.py

@@ -37,7 +37,7 @@ import threading
 from typing import Dict, List, Optional
 
 from agent.web_search_provider import WebSearchProvider
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -64,6 +64,7 @@ def register_provider(provider: WebSearchProvider, *, scope: Optional[str] = Non
         raise ValueError("Web provider .name must be a non-empty string")
     name = raw_name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         existing = target.get(name)
         target[name] = provider
@@ -83,7 +84,7 @@ def list_providers(*, scope: Optional[str] = None) -> List[WebSearchProvider]:
     """Return all registered providers, sorted by name."""
     with _lock:
         merged = dict(_providers)
-        merged.update(_scoped_providers.get(hermes_home_key(scope), {}))
+        merged.update(_scoped_providers.get(normalize_scope(scope), {}))
         items = list(merged.values())
     return sorted(items, key=lambda p: p.name)
 
@@ -94,13 +95,14 @@ def get_provider(name: str, *, scope: Optional[str] = None) -> Optional[WebSearc
         return None
     with _lock:
         key = name.strip()
-        return _scoped_providers.get(hermes_home_key(scope), {}).get(key) or _providers.get(key)
+        return _scoped_providers.get(normalize_scope(scope), {}).get(key) or _providers.get(key)
 
 
 def snapshot_registration(
     name: str, *, scope: Optional[str] = None
 ) -> Optional[WebSearchProvider]:
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name.strip())
 
@@ -115,6 +117,7 @@ def restore_registration(
     """Restore a plugin registration only when *current* is still installed."""
     key = name.strip()
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(key) is not current:
             return False

hermes_cli/dashboard_auth/registry.py

@@ -10,7 +10,7 @@ import logging
 import threading
 from typing import List, Optional
 
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 from hermes_cli.dashboard_auth.base import (
     DashboardAuthProvider,
     assert_protocol_compliance,
@@ -24,7 +24,7 @@ _scoped_providers: dict[str, dict[str, DashboardAuthProvider]] = {}
 
 def _merged(scope: Optional[str] = None) -> dict[str, DashboardAuthProvider]:
     providers = dict(_providers)
-    providers.update(_scoped_providers.get(hermes_home_key(scope), {}))
+    providers.update(_scoped_providers.get(normalize_scope(scope), {}))
     return providers
 
 
@@ -41,6 +41,7 @@ def register_provider(
     """
     assert_protocol_compliance(type(provider))
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         effective = target if scope is None else _merged(scope)
         if provider.name in effective:
@@ -70,6 +71,7 @@ def snapshot_registration(
     scope: Optional[str] = None,
 ) -> Optional[DashboardAuthProvider]:
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.get(scope, {})
         return target.get(name)
 
@@ -83,6 +85,7 @@ def restore_registration(
 ) -> bool:
     """Restore a host-owned provider registration if it is still current."""
     with _lock:
+        scope = normalize_scope(scope)
         target = _providers if scope is None else _scoped_providers.setdefault(scope, {})
         if target.get(name) is not current:
             return False

hermes_cli/fs_utils.py

@@ -0,0 +1,23 @@
+"""Filesystem helpers shared across Hermes CLI subsystems."""
+
+import os
+import shutil
+import stat
+from pathlib import Path
+
+
+def rmtree_force(path: Path) -> None:
+    """``shutil.rmtree`` that clears read-only bits before deleting.
+
+    Windows refuses to delete a tree containing read-only files — git clones
+    store objects read-only — raising ``PermissionError`` where POSIX unlinks
+    them fine. The ``onerror`` hook chmods the file writable and retries.
+    """
+    def _onerror(func, p, _exc_info):
+        try:
+            os.chmod(p, stat.S_IWRITE)
+        except OSError:
+            pass
+        func(p)
+
+    shutil.rmtree(path, onerror=_onerror)

hermes_cli/kanban.py

@@ -3474,7 +3474,15 @@ def run_slash(rest: str) -> str:
     import io
     import contextlib
 
-    tokens = shlex.split(rest, posix=os.name == "posix") if rest and rest.strip() else []
+    # Non-posix split (Windows) keeps backslashes as path separators but
+    # leaves quote characters in the tokens — strip a fully wrapping pair
+    # so `"my task"` reaches argparse as `my task`, not `"my task"`.
+    tokens = []
+    if rest and rest.strip():
+        for tok in shlex.split(rest, posix=os.name == "posix"):
+            if len(tok) >= 2 and tok[0] == tok[-1] and tok[0] in ("'", '"'):
+                tok = tok[1:-1]
+            tokens.append(tok)
 
     # Bare ``/kanban`` or ``/kanban help`` / ``--help`` / ``-h`` / ``?``:
     # show the curated short-help block instead of dumping argparse's full

hermes_cli/plugins_cmd.py

@@ -28,30 +28,12 @@ from hermes_constants import get_hermes_home
 from hermes_cli._subprocess_compat import noninteractive_git_env
 from hermes_cli.config import cfg_get
 from hermes_cli.secret_prompt import masked_secret_prompt
+from hermes_cli.fs_utils import rmtree_force as _rmtree_force
 from utils import atomic_write_text
 
 logger = logging.getLogger(__name__)
 
 
-def _rmtree_force(path: Path) -> None:
-    """``shutil.rmtree`` that clears read-only bits before deleting.
-
-    Windows refuses to delete a tree containing read-only files — git clones
-    store objects read-only — raising ``PermissionError`` where POSIX unlinks
-    them fine.  The ``onerror`` hook chmods the file writable and retries.
-    """
-    import stat
-
-    def _onerror(func, p, _exc_info):
-        try:
-            os.chmod(p, stat.S_IWRITE)
-        except OSError:
-            pass
-        func(p)
-
-    shutil.rmtree(path, onerror=_onerror)
-
-
 @functools.lru_cache(maxsize=1)
 def _resolve_git_executable() -> Optional[str]:
     """Resolve a git binary for subprocess use when ``PATH`` may be minimal.

hermes_constants.py

@@ -152,6 +152,16 @@ def hermes_home_key(path: str | Path | None = None) -> str:
     return os.path.normcase(str(resolved))
 
 
+def normalize_scope(scope: str | Path | None) -> str | None:
+    """Normalize a registry scope key, preserving ``None``.
+
+    ``None`` means the process-global layer and must stay ``None``; any other
+    value is normalized through :func:`hermes_home_key` so writes and reads
+    agree on the key (normcase on Windows, resolved absolute path).
+    """
+    return hermes_home_key(scope) if scope is not None else None
+
+
 def get_process_hermes_home() -> Path:
     """Return the Hermes home for the running process, ignoring task overrides.
 

plugins/disk-cleanup/__init__.py

@@ -113,8 +113,12 @@ def _extract_paths_from_terminal(args: Dict[str, Any], result: str) -> Set[str]:
         # Tokenise the command — catches `touch /tmp/hermes-x/test_foo.py`.
         # ``posix`` follows the host so Windows backslash paths survive
         # (``shlex.split(posix=True)`` would eat them as escapes).
+        # Non-posix mode keeps quote characters in tokens, so strip a fully
+        # wrapping pair (`"C:\file"` → `C:\file`) before matching.
         try:
             for tok in shlex.split(cmd, posix=os.name == "posix"):
+                if len(tok) >= 2 and tok[0] == tok[-1] and tok[0] in ("'", '"'):
+                    tok = tok[1:-1]
                 if tok.startswith(("/", "~")) or re.match(r"^[A-Za-z]:[\\/]", tok):
                     paths.add(tok)
         except ValueError:

tests/conformance/persistence/test_cell1_prefix_durability.py

@@ -75,6 +75,7 @@ def _recover(db_path: Path) -> list[tuple]:
 
 
 @pytest.mark.parametrize("requested_mode", [None, "DELETE", "WAL"])
+@pytest.mark.linux_only
 def test_acknowledged_appends_survive_sigkill(tmp_path, requested_mode):
     db_path = tmp_path / "state.db"
     journal = tmp_path / "acked.jsonl"

tests/conformance/persistence/test_cell3_rotation_atomicity.py

@@ -78,6 +78,7 @@ def _completed_rotations(journal: Path) -> int:
 
 
 @pytest.mark.parametrize("requested_mode", [None, "DELETE", "WAL"])
+@pytest.mark.linux_only
 def test_rotation_is_atomic_under_sigkill(tmp_path, requested_mode):
     db_path = tmp_path / "state.db"
     journal = tmp_path / "rotations.jsonl"

tests/hermes_cli/test_browser_connect_default_chromium.py

@@ -8,6 +8,8 @@ from unittest.mock import patch
 
 import pytest
 
+import posixpath
+
 import hermes_cli.browser_connect as bc
 
 
@@ -146,31 +148,37 @@ class TestDetectDefaultLinux:
 
 class TestLinuxProfileDir:
     def _env(self, monkeypatch, home):
-        monkeypatch.setenv("HOME", str(home))
+        # The code under test resolves the user's home via
+        # os.path.expanduser("~"), which reads HOME on POSIX but USERPROFILE
+        # on Windows — patch expanduser to a POSIX-FORM home so the
+        # Linux-target path resolution is exercised identically on every
+        # host (posixpath.join only inserts '/' between components; a
+        # backslash-drive home would leak host separators into the result).
+        monkeypatch.setattr(bc.os.path, "expanduser", lambda _p: home.as_posix())
         monkeypatch.delenv("XDG_CONFIG_HOME", raising=False)
 
     def test_native_path_when_nothing_exists(self, tmp_path, monkeypatch):
         self._env(monkeypatch, tmp_path)
-        assert bc.real_profile_data_dir("chromium", "Linux") == str(tmp_path / ".config" / "chromium")
+        assert bc.real_profile_data_dir("chromium", "Linux") == posixpath.join(tmp_path.as_posix(), ".config", "chromium")
 
     def test_snap_chromium_profile_is_found(self, tmp_path, monkeypatch):
         self._env(monkeypatch, tmp_path)
         snap = tmp_path / "snap" / "chromium" / "common" / "chromium"
         snap.mkdir(parents=True)
-        assert bc.real_profile_data_dir("chromium", "Linux") == str(snap)
+        assert bc.real_profile_data_dir("chromium", "Linux") == snap.as_posix()
 
     def test_flatpak_chrome_profile_is_found(self, tmp_path, monkeypatch):
         self._env(monkeypatch, tmp_path)
         flatpak = tmp_path / ".var" / "app" / "com.google.Chrome" / "config" / "google-chrome"
         flatpak.mkdir(parents=True)
-        assert bc.real_profile_data_dir("chrome", "Linux") == str(flatpak)
+        assert bc.real_profile_data_dir("chrome", "Linux") == flatpak.as_posix()
 
     def test_native_profile_wins_when_present(self, tmp_path, monkeypatch):
         self._env(monkeypatch, tmp_path)
         native = tmp_path / ".config" / "BraveSoftware" / "Brave-Browser"
         native.mkdir(parents=True)
         (tmp_path / ".var" / "app" / "com.brave.Browser" / "config" / "BraveSoftware" / "Brave-Browser").mkdir(parents=True)
-        assert bc.real_profile_data_dir("brave", "Linux") == str(native)
+        assert bc.real_profile_data_dir("brave", "Linux") == native.as_posix()
 
     def test_xdg_config_home_is_honoured(self, tmp_path, monkeypatch):
         monkeypatch.setenv("HOME", str(tmp_path))

tests/hermes_cli/test_macos_tcc_anchor.py

@@ -127,6 +127,7 @@ class TestUvStoreDetection:
         assert not tcc._is_uv_macos_store(path)
 
 
+@pytest.mark.linux_only
 class TestEnsureTccAnchor:
     def test_noop_on_non_macos(self, tmp_path, monkeypatch):
         _linux(monkeypatch)
@@ -367,6 +368,7 @@ class TestEnsureTccAnchor:
         assert f"/{_RUNTIME_DIR_NAME}/python/" in tcc._STORE_ROOT_MARKERS
 
 
+@pytest.mark.linux_only
 class TestBootGate:
     """Direct branch coverage for _passes_boot_gate.
 
@@ -453,6 +455,7 @@ class TestBootGate:
         assert tcc._passes_boot_gate(tmp_path / "staged", venv)
 
 
+@pytest.mark.linux_only
 class TestTccAnchorState:
     def test_state_active_through_unpatched_home_symlink(self, tmp_path, monkeypatch):
         # The managed-runtime layout symlinks cpython-3.11-macos-* →
@@ -523,6 +526,7 @@ class TestTccAnchorState:
         assert status == "active"
 
 
+@pytest.mark.linux_only
 class TestDoctorCheck:
     def test_missing_warns_without_fix(self, monkeypatch, capsys):
         monkeypatch.setattr(

tests/hermes_cli/test_plugins_cmd.py

@@ -448,7 +448,7 @@ class TestCmdRemove:
 
     @patch("hermes_cli.plugins_cmd._sanitize_plugin_name")
     @patch("hermes_cli.plugins_cmd._plugins_dir")
-    @patch("hermes_cli.plugins_cmd.shutil.rmtree")
+    @patch("hermes_cli.fs_utils.shutil.rmtree")
     def test_remove_deletes_plugin(self, mock_rmtree, mock_plugins_dir, mock_sanitize):
         from hermes_cli.plugins_cmd import cmd_remove
 
@@ -459,7 +459,12 @@ class TestCmdRemove:
 
         cmd_remove("test-plugin")
 
-        mock_rmtree.assert_called_once_with(mock_target)
+        # Deletion goes through the read-only-clearing rmtree wrapper: the
+        # target is deleted with an onerror hook attached (Windows read-only
+        # git objects), never a bare rmtree.
+        assert mock_rmtree.call_count == 1
+        assert mock_rmtree.call_args.args == (mock_target,)
+        assert callable(mock_rmtree.call_args.kwargs.get("onerror"))
 
     @patch("hermes_cli.plugins_cmd._sanitize_plugin_name")
     @patch("hermes_cli.plugins_cmd._plugins_dir")

tests/tools/test_approval_timeout_overflow.py

@@ -77,7 +77,10 @@ class TestApprovalTimeoutOverflowClamp:
         monkeypatch.setattr(builtins, "__import__", _blocked)
         with _with_configured_timeout(10**18):
             value = _get_approval_timeout()
-        assert value == 365 * 24 * 3600
+        # The fallback mirrors the per-host MAX_SAFE_TIMEOUT_S ceiling
+        # (imported BEFORE the block — the check itself must not depend on
+        # the patched-away import).
+        assert value == int(MAX_SAFE_TIMEOUT_S)
         # Still platform-safe for the crashing primitive.
         lock = threading.Lock()
         assert lock.acquire(timeout=value)

tools/approval.py

@@ -2168,7 +2168,19 @@ def _replace_simple_shell_expansions(word: str) -> str:
     return _PARAM_DEFAULT_RE.sub(lambda match: match.group("default"), word)
 
 
-_IS_WINDOWS = sys.platform == "win32"
+def _is_windows_drive_relative_backslash(word: str, i: int) -> bool:
+    """True when ``word[i] == '\\'`` is a path separator, not a shell escape.
+
+    On Windows drive-relative paths (``C:\\Users\\x``, ``C:Users`` forms), a
+    backslash between word characters is a separator the shell leaves alone.
+    The codebase's Windows shell is still bash, so ``r\\m`` spells ``rm``
+    exactly as on POSIX — the escape must keep being stripped everywhere
+    EXCEPT inside a drive-anchored path, where stripping would corrupt the
+    path (``C:\\Users`` → ``C:Users``) and open false-positive approvals.
+    """
+    if i <= 0 or not word[i - 1].isalnum():
+        return False
+    return bool(re.match(r"^[A-Za-z]:", word))
 
 
 def _strip_shell_word_syntax(word: str) -> str:
@@ -2178,7 +2190,12 @@ def _strip_shell_word_syntax(word: str) -> str:
     while i < len(word):
         ch = word[i]
         if quote:
-            if ch == "\\" and quote == '"' and not _IS_WINDOWS and i + 1 < len(word):
+            if (
+                ch == "\\"
+                and quote == '"'
+                and i + 1 < len(word)
+                and not _is_windows_drive_relative_backslash(word, i)
+            ):
                 chars.append(word[i + 1])
                 i += 2
                 continue
@@ -2193,7 +2210,11 @@ def _strip_shell_word_syntax(word: str) -> str:
             quote = ch
             i += 1
             continue
-        if ch == "\\" and not _IS_WINDOWS and i + 1 < len(word):
+        if (
+            ch == "\\"
+            and i + 1 < len(word)
+            and not _is_windows_drive_relative_backslash(word, i)
+        ):
             chars.append(word[i + 1])
             i += 2
             continue
@@ -3520,9 +3541,11 @@ def _get_approval_timeout() -> int:
         safe_cap = int(MAX_SAFE_TIMEOUT_S)
     except Exception:
         # Fail CLOSED: returning the raw value here would re-open the exact
-        # time_t overflow this clamp exists to prevent. ~1 year, matching
-        # agent.deadline.MAX_SAFE_TIMEOUT_S.
-        safe_cap = 365 * 24 * 3600
+        # time_t overflow this clamp exists to prevent. Mirror the per-host
+        # ceiling of agent.deadline.MAX_SAFE_TIMEOUT_S — 1 year on POSIX,
+        # but the DWORD-millisecond ceiling on Windows (WaitForSingleObject
+        # caps at ~49.7 days; a year overflows it).
+        safe_cap = 4_294_907 if sys.platform == "win32" else 365 * 24 * 3600
     if raw > safe_cap:
         logger.warning(
             "approvals.timeout=%s exceeds the platform-safe maximum; "

tools/checkpoint_manager.py

@@ -2203,22 +2203,9 @@ def store_status(checkpoint_base: Optional[Path] = None) -> Dict:
 
 
 def _rmtree_force(path: Path) -> None:
-    """``shutil.rmtree`` that clears read-only bits before deleting.
+    from hermes_cli.fs_utils import rmtree_force
 
-    Windows refuses to delete a tree containing read-only files (git objects
-    are commonly read-only), raising ``PermissionError`` where POSIX unlinks
-    them fine. The ``onerror`` hook chmods the file writable and retries.
-    """
-    import stat
-
-    def _onerror(func, p, _exc_info):
-        try:
-            os.chmod(p, stat.S_IWRITE)
-        except OSError:
-            pass
-        func(p)
-
-    shutil.rmtree(path, onerror=_onerror)
+    rmtree_force(path)
 
 
 def clear_all(checkpoint_base: Optional[Path] = None) -> Dict[str, int]:

tools/registry.py

@@ -25,7 +25,7 @@ import time
 from pathlib import Path
 from typing import Callable, Dict, List, Optional, Set
 
-from hermes_constants import hermes_home_key
+from hermes_constants import hermes_home_key, normalize_scope
 
 logger = logging.getLogger(__name__)
 
@@ -550,7 +550,7 @@ class ToolRegistry:
         scope: Optional[str] = None,
     ) -> Optional[ToolEntry]:
         """Return the local slot state without following global fallback."""
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             target = self._tools if scope is None else self._scoped_tools.get(scope, {})
             return target.get(name)
@@ -608,7 +608,7 @@ class ToolRegistry:
         The identity-bearing result lets plugin unload/reload revoke a stale
         authorization without losing durable module-to-profile attribution.
         """
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             policy = _PluginOverridePolicy(allowed)
             self._plugin_override_policy[(scope, module_namespace)] = policy
@@ -622,7 +622,7 @@ class ToolRegistry:
         scope: Optional[str] = None,
     ) -> Optional[_PluginOverridePolicy]:
         """Return one local authorization generation without fallback."""
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             return self._plugin_override_policy.get((scope, module_namespace))
 
@@ -635,7 +635,7 @@ class ToolRegistry:
         scope: Optional[str] = None,
     ) -> bool:
         """CAS-restore policy state while retaining durable scope attribution."""
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             key = (scope, module_namespace)
             if self._plugin_override_policy.get(key) is not current:
@@ -793,7 +793,7 @@ class ToolRegistry:
         owner = caller_owner or handler_owner
         if scope is None and owner is not None:
             scope = self._plugin_scope_of(owner)
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             target = (
                 self._tools
@@ -991,7 +991,7 @@ class ToolRegistry:
         newer entry under the same name, in which case unloading this entry
         must leave the newer entry untouched.
         """
-        scope = hermes_home_key(scope) if scope is not None else None
+        scope = normalize_scope(scope)
         with self._lock:
             target = (
                 self._tools