fix(windows): dirs made while elevated stay usable; an unusable token-lock dir parks once

An elevated Hermes (install, `hermes update` as admin, a gateway started from an
admin prompt) creating a dir with mode=0o700 gets CPython's protected
SYSTEM + Administrators + OWNER RIGHTS DACL, owned by Administrators. The same
user's non-elevated Desktop/gateway/CLI then gets WinError 5 inside it (#139845).

- hermes_dir_acl.make_user_dir: Path.mkdir that, only when the process is
  elevated on Windows, adds an inheriting full-control ACE for the token's user
  SID to every dir it created and to an existing leaf (repairs a dir an earlier
  elevated run left). No-op off Windows / unelevated. Routed through it: the
  gateway lock dir (scoped token locks, host rendezvous, update host
  obligation), bot mailbox, gateway.lock homes, pending_messages, media/attachment
  roots, process logs, spill dirs, vault, outbox, backups, home skeleton and
  mkdir_under_hermes_home (logs).
- W2: acquire_scoped_lock raises ScopedLockDirUnusable when this token may not
  use gateway-locks (or read a stale lock in it), logged WARNING once per dir with
  the repair (`icacls DIR /reset /t`). _acquire_platform_lock parks the adapter
  as non-retryable `lock_dir_unusable` naming the dir and the fix, instead of a
  raw PermissionError that Telegram retried forever.
- Builds on @KoNit-K's update-time ACL repair (#139870), which re-grants the
  whole home after an elevated update.

No Windows host was available: the ACL behavior is covered by
tests/test_hermes_dir_acl_windows.py (Windows CI) and needs a live retest.

Refs #134889 (W8, W2)
Refs #139845
Refs #140337

agent/runtime_session_store.py

@@ -14,6 +14,7 @@ from pathlib import Path
 import queue
 import sys
 import threading
+from hermes_dir_acl import make_user_dir
 
 logger = logging.getLogger(__name__)
 
@@ -206,7 +207,7 @@ class RuntimeSessionStore(RuntimeSessionCompressionMixin, RuntimeSessionLifecycl
         self.lock = threading.RLock()
         self.failure = None
         self.path = Path(outbox_dir) / 'pending.json'
-        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        make_user_dir(self.path.parent, mode=0o700, parents=True, exist_ok=True)
         if self.path.parent.is_symlink() or self.path.is_symlink():
             raise WorkerPersistenceError('unsafe_outbox')
         os.chmod(self.path.parent, 0o700)

agent/vault_store.py

@@ -28,6 +28,7 @@ from datetime import datetime, timezone, UTC
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 from urllib.parse import urlsplit
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import get_hermes_home
 from utils import atomic_write_bytes
@@ -203,7 +204,7 @@ class VaultStore:
     # -- key / crypto ------------------------------------------------------
 
     def _ensure_dir(self) -> None:
-        self._base.mkdir(mode=0o700, parents=True, exist_ok=True)
+        make_user_dir(self._base, mode=0o700, parents=True, exist_ok=True)
         # Route through the canonical securer (honors managed/NixOS
         # group-share mode and HERMES_UID/GID ownership) rather than a
         # bespoke chmod — same requirement as the browser-profile snapshot

cron/bot_chat_legacy.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 import json
 import logging
 from pathlib import Path
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import get_hermes_home
 from utils import atomic_json_write
@@ -24,7 +25,7 @@ def defer(key: str, job: dict, content: str, home: Path, *, for_failure: bool) -
     """Park one never-admitted output under its delivery id; a later tick hands it to the owner."""
     import time
     root = get_hermes_home().resolve() / "cron" / "bot_chat_pending"
-    root.mkdir(parents=True, exist_ok=True, mode=0o700)
+    make_user_dir(root, parents=True, exist_ok=True, mode=0o700)
     path = root / f"{key}.json"
     if not path.exists():  # an existing record (queued or already transferred) keeps its status
         atomic_json_write(path, dict(

gateway/host_rendezvous.py

@@ -44,6 +44,7 @@ from dataclasses import dataclass
 from datetime import datetime, timezone, UTC
 from pathlib import Path
 from typing import Any, Optional, Sequence
+from hermes_dir_acl import make_user_dir
 
 from utils import atomic_json_write
 
@@ -175,7 +176,7 @@ def ensure_host_state_dir() -> Path:
     Ownership of the dir is ours, so widening is repaired rather than refused.
     """
     directory = host_state_dir()
-    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
+    make_user_dir(directory, parents=True, exist_ok=True, mode=0o700)
     if sys.platform != "win32":
         with contextlib.suppress(OSError):
             if stat.S_IMODE(directory.stat().st_mode) & 0o077:

gateway/hosted_room_attachments.py

@@ -10,6 +10,7 @@ room's frozen recipient roster.
 from __future__ import annotations
 
 from gateway.hosted_rooms import EventAttachmentConflictError
+from hermes_dir_acl import make_user_dir
 
 import base64
 import binascii
@@ -343,8 +344,8 @@ class HostedRoomAttachmentStore:
         return str(self.db_path.resolve()), str(self.root.resolve()), info.st_dev, info.st_ino
 
     def _prepare_private_root(self) -> None:
-        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
-        self.blob_root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        make_user_dir(self.root, parents=True, exist_ok=True, mode=0o700)
+        make_user_dir(self.blob_root, parents=True, exist_ok=True, mode=0o700)
         for path in (self.root, self.blob_root):
             try:
                 os.chmod(path, 0o700)

gateway/platforms/base.py

@@ -19,6 +19,7 @@ import uuid
 import weakref
 from abc import ABC, abstractmethod
 from urllib.parse import urlsplit
+from hermes_dir_acl import make_user_dir, unusable_dir_remedy
 
 from utils import normalize_proxy_url
 from agent.i18n import t
@@ -605,7 +606,7 @@ def _secure_media_cache_dir(cache_dir: Path) -> None:
         if managed:
             cache_dir.mkdir(parents=True, exist_ok=True)
             return
-        cache_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+        make_user_dir(cache_dir, parents=True, exist_ok=True, mode=0o700)
         try:
             from hermes_cli.config import _secure_dir
 
@@ -2212,10 +2213,23 @@ class BasePlatformAdapter(ABC):
         holder is replaced only when the runner armed this adapter for its initial
         ``--replace`` connect (the status module validates ownership and terminates)."""
         from gateway.status import (
-            acquire_scoped_lock, scoped_lock_owner_label, take_over_scoped_lock_holder)
+            ScopedLockDirUnusable, acquire_scoped_lock, scoped_lock_owner_label, take_over_scoped_lock_holder)
         self._platform_lock_scope, self._platform_lock_identity = scope, identity
         lock_meta = {"platform": self.platform.value}
-        acquired, existing = acquire_scoped_lock(scope, identity, metadata=lock_meta)
+        try:
+            acquired, existing = acquire_scoped_lock(scope, identity, metadata=lock_meta)
+        except ScopedLockDirUnusable as exc:
+            # An ACL no retry can change (an elevated install created the lock dir, #134889 W2):
+            # stay down with the repair named instead of a reconnect loop that fails identically.
+            # The status module already logged the WARNING (once per dir).
+            self._platform_lock_identity = None
+            self._set_fatal_error(
+                "lock_dir_unusable",
+                f"{resource_desc} lock cannot be taken: the gateway lock directory {exc.filename} is not "
+                f"usable by this user (likely created by an elevated Hermes). To repair, "
+                f"{unusable_dir_remedy(exc.filename)}, then restart the gateway.",
+                retryable=False)
+            return False
         if acquired:
             return True
         if (self._platform_lock_takeover_allowed and not self._platform_lock_takeover_attempted

gateway/runtime_ownership.py

@@ -13,6 +13,7 @@ import os
 from pathlib import Path
 import stat
 import threading
+from hermes_dir_acl import make_user_dir
 
 
 class OwnershipConflict(RuntimeError):
@@ -156,7 +157,7 @@ class ProfileOwnership:
                 for home in sorted({canonical_home(p) for p in homes}):
                     if home in self._handles:
                         continue
-                    home.mkdir(mode=0o700, parents=True, exist_ok=True)
+                    make_user_dir(home, mode=0o700, parents=True, exist_ok=True)
                     path = home / 'gateway.lock'
                     fd = _open_lock(path)
                     handle = os.fdopen(fd, 'r+', encoding='utf-8')  # windows-footgun: ok — write-only

gateway/session_ingress_media.py

@@ -16,6 +16,7 @@ import re
 import stat
 import tempfile
 import threading
+from hermes_dir_acl import make_user_dir
 
 from hermes_state_runtime import RuntimeStoreError
 from utils import fsync_directory
@@ -244,7 +245,7 @@ def capture_native_media(paths):
             root = target.parent.parent
             if target.parent.resolve() != target.parent:
                 raise RuntimeStoreError('invalid_params')
-            target.parent.mkdir(mode=0o700, exist_ok=True)
+            make_user_dir(target.parent, mode=0o700, exist_ok=True)
             with _lease_lock:
                 scope = _lease_scope.get()
                 if scope is not None:
@@ -273,7 +274,7 @@ def _capture_file(path, limit, total, references, staged):
         root = _media_root()
         if root.resolve() != root:
             raise RuntimeStoreError('invalid_params')
-        root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        make_user_dir(root, mode=0o700, parents=True, exist_ok=True)
         fd, name = tempfile.mkstemp(prefix='.capture-', dir=root)
         temporary = Path(name)
         staged.append(temporary)

gateway/shutdown_flush.py

@@ -22,6 +22,7 @@ import time
 import uuid
 from pathlib import Path
 from typing import Any, Dict, Optional
+from hermes_dir_acl import make_user_dir
 
 logger = logging.getLogger(__name__)
 
@@ -46,7 +47,7 @@ def _get_flush_dir():
     flush_dir = get_hermes_home() / "pending_messages"
     from hermes_constants import assert_named_profile_home_live
     assert_named_profile_home_live(flush_dir)
-    flush_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+    make_user_dir(flush_dir, parents=True, exist_ok=True, mode=0o700)
     if os.name == "posix":
         os.chmod(flush_dir, 0o700)
     return flush_dir

gateway/status.py

@@ -20,6 +20,7 @@ from dataclasses import dataclass
 from datetime import datetime, timezone, UTC
 from pathlib import Path
 from typing import Any, Callable, NamedTuple, Optional
+from hermes_dir_acl import make_user_dir, unusable_dir_remedy
 
 from gateway.status_inline_source import (
     command_line_runs_inline_source,
@@ -207,7 +208,7 @@ def record_start_and_check_storm(
     Best-effort: a broken ``gateway-starts.log`` ledger is logged and swallowed, never fatal."""
     try:
         path = get_hermes_home() / "gateway-starts.log"
-        path.parent.mkdir(parents=True, exist_ok=True)
+        make_user_dir(path.parent, parents=True, exist_ok=True)
         now = datetime.now(UTC).timestamp()
         existing: list[float] = []
         if path.exists():
@@ -1056,7 +1057,7 @@ def write_pid_file() -> None:
     """Write this process's PID record via O_CREAT|O_EXCL; a racing gateway's FileExistsError
     propagates for the caller to decide."""
     path = _get_pid_path()
-    path.parent.mkdir(parents=True, exist_ok=True)
+    make_user_dir(path.parent, parents=True, exist_ok=True)
     _write_json_excl(path, _build_pid_record())
     _clear_running_pid_cache()
 
@@ -1541,12 +1542,56 @@ def _process_is_stopped(pid: int) -> bool:
     return False
 
 
+class ScopedLockDirUnusable(PermissionError):
+    """This token cannot create, read or replace locks in the scoped-lock dir (#134889 W2).
+
+    On Windows an elevated install/update creates ``gateway-locks`` owned by Administrators with
+    no entry for the user; the same user's non-elevated gateway then gets ``Permission denied`` on
+    every token lock. Retrying cannot fix an ACL, so the adapter reports it once, names the repair
+    and stays down instead of crash-looping. A ``PermissionError`` subclass so existing callers
+    keep their handling.
+    """
+
+
+# Lock dirs already reported unusable by this process: WARNING once per dir, DEBUG after, since the
+# reconnect watcher and every adapter sharing the dir re-enter the acquire path.
+_warned_unusable_lock_dirs: set[Path] = set()
+
+
+def _unusable_lock_dir(lock_dir: Path, exc: PermissionError) -> ScopedLockDirUnusable:
+    level = logging.DEBUG if lock_dir in _warned_unusable_lock_dirs else logging.WARNING
+    _warned_unusable_lock_dirs.add(lock_dir)
+    logger.log(level, "Gateway lock directory %s is not usable by this process (%s); it was likely created "
+               "by an elevated (administrator) Hermes. Platforms that need a token lock stay down until it "
+               "is repaired. To repair, %s", lock_dir, exc, unusable_dir_remedy(lock_dir))
+    return ScopedLockDirUnusable(exc.errno, f"gateway lock directory is not usable: {exc}", str(lock_dir))
+
+
 def acquire_scoped_lock(
     scope: str, identity: str, metadata: Optional[dict[str, Any]] = None
 ) -> tuple[bool, Optional[dict[str, Any]]]:
-    """Acquire a machine-local lock keyed by scope + identity (one Telegram token across homes)."""
+    """Acquire a machine-local lock keyed by scope + identity (one Telegram token across homes).
+
+    Raises :class:`ScopedLockDirUnusable` (logged once per dir with the repair) when this token may
+    not use the lock dir at all."""
     lock_path = _get_scope_lock_path(scope, identity)
-    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        acquired = _acquire_scoped_lock_at(lock_path, scope, identity, metadata)
+    except PermissionError as exc:
+        raise _unusable_lock_dir(lock_path.parent, exc) from exc
+    _warned_unusable_lock_dirs.discard(lock_path.parent)
+    return acquired
+
+
+def _acquire_scoped_lock_at(
+    lock_path: Path, scope: str, identity: str, metadata: Optional[dict[str, Any]]
+) -> tuple[bool, Optional[dict[str, Any]]]:
+    from hermes_dir_acl import make_user_dir
+    make_user_dir(lock_path.parent, parents=True, exist_ok=True)
+    # An existing lock this token may not read (left by an elevated gateway) must raise here:
+    # ``_read_json_file`` would call it empty and O_EXCL "in use", a conflict with no holder.
+    with contextlib.suppress(FileNotFoundError), open(lock_path, "rb"):
+        pass
     record = {
         **_build_pid_record(), "scope": scope, "identity_hash": _scope_hash(identity),
         "metadata": metadata or {}, "updated_at": _utc_now_iso(),

hermes_cli/backup.py

@@ -16,6 +16,7 @@ from contextlib import contextmanager, suppress
 from datetime import datetime, timezone, UTC
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Tuple
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import (
     LOCAL_RUNTIME_ROOT_DIRS, _get_platform_default_hermes_home, get_default_hermes_root, get_hermes_home,
@@ -1305,10 +1306,10 @@ def _create_quick_snapshot_locked(
     snap_dir = root / snap_id
     staging_dir = root / f".{snap_id}.{os.getpid()}.partial"
     shutil.rmtree(staging_dir, ignore_errors=True)
-    root.mkdir(parents=True, exist_ok=True, mode=0o700)
+    make_user_dir(root, parents=True, exist_ok=True, mode=0o700)
     if os.name != "nt":
         os.chmod(root, 0o700)
-    staging_dir.mkdir(mode=0o700, exist_ok=False)
+    make_user_dir(staging_dir, mode=0o700, exist_ok=False)
     logger.info("quick snapshot phase=copy status=started id=%s", snap_id)
 
     manifest: dict[str, int] = {}  # rel_path -> file size

hermes_cli/config_home.py

@@ -2,6 +2,7 @@
 
 import os
 from pathlib import Path
+from hermes_dir_acl import make_user_dir
 
 
 class HomeInitializationError(RuntimeError):
@@ -36,7 +37,7 @@ def _ensure_directory(path: Path, *, create: bool, secure: bool, home: Path) ->
             if not link.is_dir():
                 raise FileNotFoundError(f"Directory link is unavailable: {link}")
         if create:
-            path.mkdir(parents=True, exist_ok=True)
+            make_user_dir(path, parents=True, exist_ok=True)
         elif not path.is_dir():
             raise FileNotFoundError(f"Required directory does not exist: {path}")
         # The operator owns permissions beyond a link, including logs/curator.

hermes_cli/update_host_obligation.py

@@ -33,6 +33,7 @@ import tempfile
 import time
 from pathlib import Path
 from typing import Any, Callable, Iterable, Optional
+from hermes_dir_acl import make_user_dir
 
 logger = logging.getLogger("hermes_cli.update_cmd")
 
@@ -244,7 +245,7 @@ def _write_locked(path: Path, *, expected_sha: str, runtimes: Optional[list], pr
     if runtimes is not None:
         payload["inventory"] = {"version": 1, "runtimes": runtimes}
     try:
-        path.parent.mkdir(parents=True, exist_ok=True)
+        make_user_dir(path.parent, parents=True, exist_ok=True)
         _write_record(path, payload)
     except Exception as exc:  # health: allow BLE001 -- never raises: the caller falls back to the per-home marker
         logger.debug("Could not write host update-restart obligation: %s", exc)
@@ -261,7 +262,7 @@ def _record_mutex(path: Path):
     """
     # One guard for every install's record (and the unkeyed one an older release still locks).
     guard = path.parent / f".{Path(HOST_OBLIGATION_NAME).stem}.mutex"
-    guard.mkdir(parents=True, exist_ok=True)
+    make_user_dir(guard, parents=True, exist_ok=True)
     with _marker_mutex()(guard / "record"):
         yield
 

hermes_constants.py

@@ -429,7 +429,12 @@ def mkdir_under_hermes_home(path: str | Path) -> Path:
     """Create *path*, but never materialize a deleted/missing named profile."""
     target = Path(path)
     assert_named_profile_home_live(target)
-    target.mkdir(parents=True, exist_ok=True)
+    try:  # lazy: this module is in the launcher's bootstrap closure and must import alone
+        from hermes_dir_acl import make_user_dir
+    except ImportError:
+        target.mkdir(parents=True, exist_ok=True)
+    else:
+        make_user_dir(target, parents=True, exist_ok=True)
     return target
 
 

hermes_dir_acl.py

@@ -0,0 +1,150 @@
+"""Directories an elevated Windows process creates must stay usable by the same user unelevated.
+
+Since CPython 3.13, ``os.mkdir(path, 0o700)`` on Windows applies a protected DACL of exactly
+SYSTEM, Administrators and OWNER RIGHTS. An elevated process's default owner is
+``BUILTIN\\Administrators``, so OWNER RIGHTS resolves to Administrators too, and the user's own
+non-elevated token (where Administrators is deny-only) gets ``WinError 5`` on the directory and on
+everything created inside it (#139845, #139485, the ``gateway-locks`` reports in #134889 W2). A plain
+``mkdir`` inherits its parent's ACL instead, which under ``%LOCALAPPDATA%`` names the user, so the
+``mode=0o700`` sites are the ones that strand the user.
+
+:func:`make_user_dir` is the one place Hermes state/lock/log directories are created. Off Windows,
+or in a non-elevated process, it is exactly ``Path.mkdir``. In an elevated Windows process it adds
+an explicit, inheriting full-control ACE for the token's user SID to the directory, so the directory
+and every child created later stay usable by that user after elevation ends. An elevated call on an
+existing directory re-grants the same ACE, which repairs a directory an earlier elevated run left
+behind. A non-elevated process cannot change such a DACL; callers that hit one report it once with
+:func:`unusable_dir_remedy`.
+"""
+from __future__ import annotations
+
+import functools
+import logging
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+log = logging.getLogger(__name__)
+
+_TOKEN_QUERY = 0x0008
+_TOKEN_USER = 1
+_TOKEN_ELEVATION = 20
+
+
+@functools.lru_cache(maxsize=1)
+def process_is_elevated() -> bool:
+    """True for an elevated (UAC full-token) Windows process; False elsewhere or on any error."""
+    if sys.platform != "win32":
+        return False
+    import ctypes
+    from ctypes import wintypes
+
+    advapi = ctypes.WinDLL("advapi32", use_last_error=True)
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel.GetCurrentProcess.restype = wintypes.HANDLE
+    advapi.OpenProcessToken.argtypes = [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)]
+    advapi.GetTokenInformation.argtypes = [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD,
+                                           ctypes.POINTER(wintypes.DWORD)]
+    kernel.CloseHandle.argtypes = [wintypes.HANDLE]
+    token = wintypes.HANDLE()
+    if not advapi.OpenProcessToken(kernel.GetCurrentProcess(), _TOKEN_QUERY, ctypes.byref(token)):
+        return False
+    try:
+        elevated, size = wintypes.DWORD(), wintypes.DWORD()
+        if not advapi.GetTokenInformation(token, _TOKEN_ELEVATION, ctypes.byref(elevated),
+                                          ctypes.sizeof(elevated), ctypes.byref(size)):
+            return False
+        return bool(elevated.value)
+    finally:
+        kernel.CloseHandle(token)
+
+
+def _grant_with_pywin32(path: Path) -> None:
+    import ntsecuritycon
+    import win32api
+    import win32security
+
+    token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32security.TOKEN_QUERY)
+    user_sid = win32security.GetTokenInformation(token, win32security.TokenUser)[0]
+    descriptor = win32security.GetFileSecurity(str(path), win32security.DACL_SECURITY_INFORMATION)
+    dacl = descriptor.GetSecurityDescriptorDacl()
+    inherit = win32security.OBJECT_INHERIT_ACE | win32security.CONTAINER_INHERIT_ACE
+    if dacl is not None:
+        for index in range(dacl.GetAceCount()):
+            (ace_type, flags, *_), mask, sid = dacl.GetAce(index)[:3]
+            if (ace_type == win32security.ACCESS_ALLOWED_ACE_TYPE and sid == user_sid
+                    and mask & ntsecuritycon.FILE_ALL_ACCESS == ntsecuritycon.FILE_ALL_ACCESS
+                    and flags & inherit == inherit and not flags & win32security.INHERIT_ONLY_ACE):
+                return  # already granted: never grow the DACL on repeat calls
+    else:
+        dacl = win32security.ACL()
+    dacl.AddAccessAllowedAceEx(win32security.ACL_REVISION_DS, inherit, ntsecuritycon.FILE_ALL_ACCESS, user_sid)
+    descriptor.SetSecurityDescriptorDacl(True, dacl, False)
+    win32security.SetFileSecurity(str(path), win32security.DACL_SECURITY_INFORMATION, descriptor)
+
+
+def _grant_with_icacls(path: Path) -> None:
+    whoami = subprocess.run(["whoami", "/user", "/fo", "csv", "/nh"], capture_output=True, text=True,
+                            check=True, encoding="utf-8", errors="replace")
+    sid = whoami.stdout.strip().rsplit(",", 1)[-1].strip().strip('"')
+    if not sid.upper().startswith("S-1-"):
+        raise OSError(f"could not resolve the current user SID from whoami: {whoami.stdout!r}")
+    subprocess.run(["icacls", str(path), "/grant", f"*{sid}:(OI)(CI)F", "/Q"], capture_output=True,
+                   check=True)
+
+
+def grant_user_access(path: str | Path) -> bool:
+    """Give the token's user an inheriting full-control ACE on *path* when this process is elevated.
+
+    True when a grant was attempted and did not fail; False off Windows, when not elevated, or when
+    the grant failed (logged; the directory itself still exists)."""
+    if sys.platform != "win32" or not process_is_elevated():
+        return False
+    target = Path(path)
+    try:
+        try:
+            _grant_with_pywin32(target)
+        except ImportError:
+            _grant_with_icacls(target)
+    except Exception as exc:  # health: allow BLE001 -- pywintypes.error is not an OSError; never fail the mkdir
+        log.warning("Could not grant the current user access to %s created while elevated (%s); a "
+                    "non-elevated Hermes may not be able to use it. To repair, from an elevated prompt run: "
+                    'icacls "%s" /reset /t', target, exc, target)
+        return False
+    return True
+
+
+# Dirs this elevated process already granted: the grant reads and may rewrite a DACL, and the
+# callers (log handlers, lock acquires) re-enter on hot paths.
+_granted: set[Path] = set()
+
+
+def make_user_dir(path: str | Path, mode: int = 0o777, *, parents: bool = False, exist_ok: bool = False) -> Path:
+    """``Path.mkdir`` (same arguments and defaults) that keeps an elevated Windows creation usable
+    by the same user unelevated.
+
+    Off Windows or unelevated this is ``Path.mkdir`` and nothing else. Elevated, every directory
+    this call created (missing parents included) and the leaf, even when it already existed (the
+    repair case), get the user's inheriting full-control ACE once per process."""
+    target = Path(path)
+    if sys.platform != "win32" or not process_is_elevated():
+        target.mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
+        return target
+    created = []
+    probe = target
+    while not probe.exists() and probe.parent != probe:
+        created.append(probe)
+        probe = probe.parent
+    target.mkdir(mode=mode, parents=parents, exist_ok=exist_ok)
+    for directory in reversed(created or [target]):
+        if directory not in _granted and grant_user_access(directory):
+            _granted.add(directory)
+    return target
+
+
+def unusable_dir_remedy(path: str | Path) -> str:
+    """The one-line repair for a directory this process may not use (owner from an elevated run)."""
+    if os.name == "nt":
+        return f'from an elevated prompt run: icacls "{path}" /reset /t'
+    return f'run: sudo chown -R "$(id -un)" "{path}"'

hermes_state.py

@@ -23,6 +23,7 @@ import uuid
 from collections import deque
 from contextlib import contextmanager, suppress
 from pathlib import Path
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import get_hermes_home, mkdir_under_hermes_home
 from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TypeVar, cast
@@ -434,7 +435,7 @@ def divert_session_transcript_jsonl(session_id: str, messages) -> Optional[Path]
     if not sid or not messages:
         return None
     sessions_dir = get_hermes_home() / "sessions"
-    sessions_dir.mkdir(parents=True, exist_ok=True)
+    make_user_dir(sessions_dir, parents=True, exist_ok=True)
     path = sessions_dir / f"{sid}.jsonl"
     with path.open("a", encoding="utf-8") as handle:
         for msg in messages:

tests/gateway/test_scoped_lock_dir_unusable.py

@@ -0,0 +1,103 @@
+"""#134889 W2: a token-lock dir this user may not write gives one named warning, never a crash loop.
+
+The Windows reports were ``Permission denied`` on ``gateway-locks\\telegram-bot-token-*.lock``; the
+leading cause is a ``gateway-locks`` dir created by an elevated install. The POSIX analogue used
+here is real, not patched: a lock dir without write permission for this uid. The Windows ACL shape
+itself is proved by ``test_scoped_lock_dir_windows_acl.py`` on Windows CI.
+"""
+import logging
+import os
+import stat
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from gateway import status
+from gateway.platforms.base import BasePlatformAdapter
+from gateway.restart import is_global_startup_conflict
+
+pytestmark = pytest.mark.skipif(os.name == "nt" or os.geteuid() == 0,  # windows-footgun: ok — skipped on nt
+                                reason="POSIX mode bits; root ignores them")
+
+
+class _StubAdapter(BasePlatformAdapter):
+    platform = MagicMock(value="telegram")
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        pass
+
+    async def send(self, *args: Any, **kwargs: Any) -> None:
+        pass
+
+    async def get_chat_info(self, chat_id: str) -> dict[str, Any]:
+        return {}
+
+
+def _adapter():
+    obj = _StubAdapter.__new__(_StubAdapter)
+    obj._running = True
+    obj._fatal_error_code = obj._fatal_error_message = obj._fatal_error_handler = None
+    obj._fatal_error_retryable = True
+    obj._platform_lock_scope = obj._platform_lock_identity = None
+    obj._platform_lock_takeover_allowed = obj._platform_lock_takeover_attempted = False
+    obj._status_write_logged = None
+    obj._write_runtime_status_safe = lambda *a, **k: None
+    return obj
+
+
+@pytest.fixture()
+def locked_out_dir(tmp_path, monkeypatch):
+    lock_dir = tmp_path / "gateway-locks"
+    lock_dir.mkdir()
+    lock_dir.chmod(stat.S_IRUSR | stat.S_IXUSR)  # r-x: lists, cannot create the token lock
+    monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(lock_dir))
+    getattr(status, "_warned_unusable_lock_dirs", set()).clear()
+    yield lock_dir
+    lock_dir.chmod(stat.S_IRWXU)
+
+
+def test_unusable_lock_dir_parks_the_adapter_with_the_repair(locked_out_dir, caplog):
+    adapter = _adapter()
+    with caplog.at_level(logging.DEBUG, logger="gateway.status"):
+        # On main this raised PermissionError out of connect() and the adapter retried forever.
+        assert adapter._acquire_platform_lock("telegram-bot-token", "tok", "Telegram bot token") is False
+        assert _adapter()._acquire_platform_lock("telegram-bot-token", "tok", "Telegram bot token") is False
+
+    assert adapter._fatal_error_code == "lock_dir_unusable"
+    assert adapter._fatal_error_retryable is False
+    assert not is_global_startup_conflict(adapter._fatal_error_code)
+    assert str(locked_out_dir) in adapter._fatal_error_message
+    assert "chown" in adapter._fatal_error_message
+    assert adapter._platform_lock_identity is None  # nothing to release
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING and "not usable" in r.getMessage()]
+    assert len(warnings) == 1, [r.getMessage() for r in caplog.records]
+    assert str(locked_out_dir) in warnings[0].getMessage()
+
+
+def test_lock_dir_repair_is_picked_up_without_restart(locked_out_dir):
+    with pytest.raises(status.ScopedLockDirUnusable):
+        status.acquire_scoped_lock("telegram-bot-token", "tok")
+    locked_out_dir.chmod(stat.S_IRWXU)
+    assert status.acquire_scoped_lock("telegram-bot-token", "tok") == (True, None)
+    assert locked_out_dir not in status._warned_unusable_lock_dirs
+    status.release_scoped_lock("telegram-bot-token", "tok")
+
+
+def test_unreadable_stale_token_lock_is_named_not_reported_as_a_holderless_conflict(tmp_path, monkeypatch):
+    lock_dir = tmp_path / "gateway-locks"
+    lock_dir.mkdir()
+    monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(lock_dir))
+    stale = status._get_scope_lock_path("telegram-bot-token", "tok")
+    stale.write_text('{"pid": 1}', encoding="utf-8")
+    stale.chmod(0)
+    lock_dir.chmod(stat.S_IRUSR | stat.S_IXUSR)  # the stale file cannot be read or unlinked either
+    try:
+        with pytest.raises(status.ScopedLockDirUnusable):
+            status.acquire_scoped_lock("telegram-bot-token", "tok")
+    finally:
+        lock_dir.chmod(stat.S_IRWXU)
+        stale.chmod(stat.S_IRUSR | stat.S_IWUSR)

tests/hermes_cli/test_update_host_obligation.py

@@ -229,7 +229,7 @@ def test_one_installs_completed_restart_never_erases_another_installs_debt(tmp_p
     from pathlib import Path
 
     source = Path(__file__).resolve().parents[2]
-    files = ("hermes_constants.py", "hermes_cli/__init__.py", "hermes_cli/update_host_obligation.py",
+    files = ("hermes_constants.py", "hermes_dir_acl.py", "hermes_cli/__init__.py", "hermes_cli/update_host_obligation.py",
              "hermes_cli/update_restart_recovery.py", "hermes_cli/update_lock.py")
     roots = {}
     for name in ("install-a", "install-b"):

tests/test_hermes_dir_acl.py

@@ -0,0 +1,61 @@
+"""``make_user_dir`` contract off Windows and its grant selection (the Windows ACL effect itself is
+proved by ``test_hermes_dir_acl_windows.py`` on Windows CI)."""
+import os
+
+import pytest
+
+import hermes_dir_acl
+
+
+def test_off_windows_it_is_plain_mkdir(tmp_path, monkeypatch):
+    calls = []
+    monkeypatch.setattr(hermes_dir_acl, "grant_user_access", lambda p: calls.append(p))
+    target = hermes_dir_acl.make_user_dir(tmp_path / "a" / "b", 0o700, parents=True)
+    assert target.is_dir() and calls == []
+    if os.name != "nt":
+        assert (target.stat().st_mode & 0o777) == 0o700
+    with pytest.raises(FileExistsError):
+        hermes_dir_acl.make_user_dir(target)
+    with pytest.raises(FileNotFoundError):
+        hermes_dir_acl.make_user_dir(tmp_path / "x" / "y")
+
+
+def test_elevated_grants_every_created_dir_and_an_existing_leaf_once(tmp_path, monkeypatch):
+    granted = []
+    monkeypatch.setattr(hermes_dir_acl.sys, "platform", "win32")
+    monkeypatch.setattr(hermes_dir_acl, "process_is_elevated", lambda: True)
+    monkeypatch.setattr(hermes_dir_acl, "grant_user_access", lambda p: granted.append(p) or True)
+    monkeypatch.setattr(hermes_dir_acl, "_granted", set())
+
+    leaf = hermes_dir_acl.make_user_dir(tmp_path / "state" / "gateway-locks", parents=True, exist_ok=True)
+    assert granted == [tmp_path / "state", leaf]  # parents first, so the leaf inherits from a granted parent
+    hermes_dir_acl.make_user_dir(leaf, exist_ok=True)
+    assert granted == [tmp_path / "state", leaf]  # memoized per process
+
+    existing = tmp_path / "left-by-an-earlier-elevated-run"
+    existing.mkdir()
+    hermes_dir_acl.make_user_dir(existing, exist_ok=True)
+    assert granted[-1] == existing  # the repair case: re-grant an existing leaf
+
+
+def test_a_failed_grant_never_fails_the_mkdir_and_is_retried(tmp_path, monkeypatch):
+    monkeypatch.setattr(hermes_dir_acl.sys, "platform", "win32")
+    monkeypatch.setattr(hermes_dir_acl, "process_is_elevated", lambda: True)
+    monkeypatch.setattr(hermes_dir_acl, "_granted", set())
+    attempts = []
+
+    def boom(path):
+        attempts.append(path)
+        raise OSError("access denied")
+
+    monkeypatch.setattr(hermes_dir_acl, "_grant_with_pywin32", boom)
+    target = hermes_dir_acl.make_user_dir(tmp_path / "d", exist_ok=True)
+    hermes_dir_acl.make_user_dir(target, exist_ok=True)
+    assert target.is_dir() and attempts == [target, target]
+
+
+def test_remedy_names_the_platform_repair(monkeypatch):
+    monkeypatch.setattr(hermes_dir_acl.os, "name", "nt")
+    assert hermes_dir_acl.unusable_dir_remedy("C:\\x") == 'from an elevated prompt run: icacls "C:\\x" /reset /t'
+    monkeypatch.setattr(hermes_dir_acl.os, "name", "posix")
+    assert "chown" in hermes_dir_acl.unusable_dir_remedy("/x")

tests/test_hermes_dir_acl_windows.py

@@ -0,0 +1,203 @@
+"""#139845 / #134889 W8 + W2 on real Windows ACLs.
+
+The reporters' dirs were created by an elevated process: a DACL of SYSTEM + Administrators only, no
+entry for the user. Their own non-elevated token holds Administrators as deny-only, so it gets
+``WinError 5`` inside. Each probe runs in a child that builds that filtered token itself
+(``CreateRestrictedToken``: Administrators deny-only, every privilege but SeChangeNotify removed, so
+no SeBackup/SeRestore bypass), impersonates it, and then touches the dir the way production does.
+
+The parent needs an elevated token (the hosted runner has one) to stand in for the elevated
+install; ``make_user_dir`` only grants when the creating process is elevated.
+"""
+import json
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.platforms("windows")
+
+_REPO = Path(__file__).resolve().parent.parent
+
+_FILTERED_TOKEN = textwrap.dedent(r"""
+    import ctypes, ctypes.wintypes as w, json, os, sys
+    from pathlib import Path
+
+    advapi = ctypes.WinDLL("advapi32", use_last_error=True)
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+
+    class SID_AND_ATTRIBUTES(ctypes.Structure):
+        _fields_ = [("Sid", ctypes.c_void_p), ("Attributes", w.DWORD)]
+
+    kernel.GetCurrentProcess.restype = w.HANDLE
+    advapi.OpenProcessToken.argtypes = [w.HANDLE, w.DWORD, ctypes.POINTER(w.HANDLE)]
+    advapi.ConvertStringSidToSidW.argtypes = [w.LPCWSTR, ctypes.POINTER(ctypes.c_void_p)]
+    advapi.CreateRestrictedToken.argtypes = [w.HANDLE, w.DWORD, w.DWORD, ctypes.POINTER(SID_AND_ATTRIBUTES),
+                                             w.DWORD, ctypes.c_void_p, w.DWORD, ctypes.c_void_p,
+                                             ctypes.POINTER(w.HANDLE)]
+    advapi.ImpersonateLoggedOnUser.argtypes = [w.HANDLE]
+    token = w.HANDLE()
+    # TOKEN_DUPLICATE | TOKEN_QUERY | TOKEN_ASSIGN_PRIMARY | TOKEN_IMPERSONATE
+    if not advapi.OpenProcessToken(kernel.GetCurrentProcess(), 0x0002 | 0x0008 | 0x0001 | 0x0004,
+                                   ctypes.byref(token)):
+        raise ctypes.WinError(ctypes.get_last_error())
+    admins = ctypes.c_void_p()
+    if not advapi.ConvertStringSidToSidW("S-1-5-32-544", ctypes.byref(admins)):
+        raise ctypes.WinError(ctypes.get_last_error())
+    disable = (SID_AND_ATTRIBUTES * 1)(SID_AND_ATTRIBUTES(admins, 0))
+    filtered = w.HANDLE()
+    # DISABLE_MAX_PRIVILEGE: what a UAC-filtered token keeps, i.e. no backup/restore bypass.
+    if not advapi.CreateRestrictedToken(token, 0x1, 1, disable, 0, None, 0, None, ctypes.byref(filtered)):
+        raise ctypes.WinError(ctypes.get_last_error())
+    if not advapi.ImpersonateLoggedOnUser(filtered):
+        raise ctypes.WinError(ctypes.get_last_error())
+
+    def usable(directory):
+        try:
+            fd = os.open(Path(directory) / "probe.lock", os.O_CREAT | os.O_WRONLY, 0o600)
+        except PermissionError:
+            return False
+        os.close(fd)
+        return True
+""")
+
+_DROP_PRIVILEGES = textwrap.dedent(r"""
+    import ctypes, ctypes.wintypes as w, json, os, sys
+    from pathlib import Path
+
+    advapi = ctypes.WinDLL("advapi32", use_last_error=True)
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+
+    class LUID(ctypes.Structure):
+        _fields_ = [("LowPart", w.DWORD), ("HighPart", w.LONG)]
+
+    class LUID_AND_ATTRIBUTES(ctypes.Structure):
+        _fields_ = [("Luid", LUID), ("Attributes", w.DWORD)]
+
+    class TOKEN_PRIVILEGES(ctypes.Structure):
+        _fields_ = [("PrivilegeCount", w.DWORD), ("Privileges", LUID_AND_ATTRIBUTES * 1)]
+
+    kernel.GetCurrentProcess.restype = w.HANDLE
+    advapi.OpenProcessToken.argtypes = [w.HANDLE, w.DWORD, ctypes.POINTER(w.HANDLE)]
+    advapi.LookupPrivilegeValueW.argtypes = [w.LPCWSTR, w.LPCWSTR, ctypes.POINTER(LUID)]
+    advapi.AdjustTokenPrivileges.argtypes = [w.HANDLE, w.BOOL, ctypes.POINTER(TOKEN_PRIVILEGES),
+                                             w.DWORD, ctypes.c_void_p, ctypes.c_void_p]
+    token = w.HANDLE()
+    if not advapi.OpenProcessToken(kernel.GetCurrentProcess(), 0x0020 | 0x0008, ctypes.byref(token)):
+        raise ctypes.WinError(ctypes.get_last_error())
+    for name in ("SeBackupPrivilege", "SeRestorePrivilege"):
+        luid = LUID()
+        if not advapi.LookupPrivilegeValueW(None, name, ctypes.byref(luid)):
+            raise ctypes.WinError(ctypes.get_last_error())
+        tp = TOKEN_PRIVILEGES(1, (LUID_AND_ATTRIBUTES * 1)(LUID_AND_ATTRIBUTES(luid, 0x00000004)))
+        if not advapi.AdjustTokenPrivileges(token, False, ctypes.byref(tp), 0, None, None):
+            raise ctypes.WinError(ctypes.get_last_error())
+""")
+
+_REPORTERS_DACL = ["/inheritance:r", "/grant:r", "*S-1-5-18:(OI)(CI)F", "*S-1-5-32-544:(OI)(CI)F"]
+
+
+def _elevated() -> bool:
+    sys.path.insert(0, str(_REPO))
+    from hermes_dir_acl import process_is_elevated
+    return process_is_elevated()
+
+
+def _child(code: str, *args: str, prelude: str = _FILTERED_TOKEN) -> dict:
+    child = subprocess.run([sys.executable, "-c", prelude + textwrap.dedent(code), *args], cwd=_REPO,
+                           capture_output=True, text=True, check=False, timeout=180)
+    assert child.returncode == 0, child.stderr
+    return json.loads(child.stdout.strip().splitlines()[-1])
+
+
+@pytest.fixture()
+def reporters_parent(tmp_path):
+    if not _elevated():
+        pytest.skip("needs an elevated token to stand in for the elevated install")
+    parent = tmp_path / "hermes"
+    parent.mkdir()
+    subprocess.run(["icacls", str(parent), *_REPORTERS_DACL], check=True, capture_output=True)
+    yield parent
+    subprocess.run(["icacls", str(parent), "/reset", "/t", "/c", "/q"], check=False, capture_output=True)
+
+
+def test_dir_made_while_elevated_stays_usable_by_the_filtered_token(reporters_parent):
+    from hermes_dir_acl import make_user_dir
+
+    plain = reporters_parent / "plain"
+    plain.mkdir()  # inherits the reporters' SYSTEM + Administrators DACL
+    granted = make_user_dir(reporters_parent / "gateway-locks", 0o700)
+    result = _child("""
+        print(json.dumps({"plain": usable(sys.argv[1]), "granted": usable(sys.argv[2])}))
+    """, str(plain), str(granted))
+    assert result["plain"] is False, "the filtered token could use the reporters' DACL: the probe is not faithful"
+    assert result["granted"] is True
+
+
+def test_elevated_rerun_repairs_a_dir_an_earlier_elevated_run_left(reporters_parent):
+    from hermes_dir_acl import make_user_dir
+
+    left = reporters_parent / "state"
+    left.mkdir()
+    before = _child("print(json.dumps({'usable': usable(sys.argv[1])}))", str(left))
+    make_user_dir(left)
+    make_user_dir(left)  # idempotent: no second ACE
+    after = _child("print(json.dumps({'usable': usable(sys.argv[1])}))", str(left))
+    acl = subprocess.run(["icacls", str(left)], capture_output=True, text=True, check=True).stdout
+    assert before["usable"] is False and after["usable"] is True, acl
+
+
+def test_token_lock_on_an_unusable_lock_dir_parks_with_one_named_warning(tmp_path):
+    """W2: the gateway meets a ``gateway-locks`` it may not create in.
+
+    Same shape as ``tests/tools/test_bot_live_delivery_windows_acl.py`` (proven on the hosted
+    runner): an Everyone DENY ACE that no identity escapes, and a child with SeBackup/SeRestore
+    removed so ``os.open`` has no bypass. Needs no elevation."""
+    lock_dir = tmp_path / "gateway-locks"
+    lock_dir.mkdir()
+    subprocess.run(["icacls", str(lock_dir), "/deny", "*S-1-1-0:(F)"], check=True, capture_output=True)
+    try:
+        _w2(lock_dir)
+    finally:
+        subprocess.run(["icacls", str(lock_dir), "/remove:d", "*S-1-1-0"], check=False, capture_output=True)
+
+
+def _w2(lock_dir):
+    result = _child(r"""
+        import logging
+        from unittest.mock import MagicMock
+        os.environ["HERMES_GATEWAY_LOCK_DIR"] = sys.argv[1]
+        records = []
+        handler = logging.Handler()
+        handler.emit = lambda record: records.append((record.levelname, record.getMessage()))
+        logging.getLogger().addHandler(handler)
+        logging.getLogger().setLevel(logging.DEBUG)
+        from gateway.platforms.base import BasePlatformAdapter
+
+        class Stub(BasePlatformAdapter):
+            platform = MagicMock(value="telegram")
+            async def connect(self, *, is_reconnect=False): return True
+            async def disconnect(self): pass
+            async def send(self, *a, **k): pass
+            async def get_chat_info(self, chat_id): return {}
+
+        codes = []
+        for _ in range(3):  # the reconnect watcher re-enters; one WARNING, never a raise
+            a = Stub.__new__(Stub)
+            a._running, a._fatal_error_handler, a._status_write_logged = True, None, None
+            a._fatal_error_code = a._fatal_error_message = None
+            a._fatal_error_retryable = True
+            a._platform_lock_scope = a._platform_lock_identity = None
+            a._platform_lock_takeover_allowed = a._platform_lock_takeover_attempted = False
+            a._write_runtime_status_safe = lambda *x, **k: None
+            ok = a._acquire_platform_lock("telegram-bot-token", "tok", "Telegram bot token")
+            codes.append([ok, a._fatal_error_code, a._fatal_error_retryable, a._fatal_error_message])
+        print(json.dumps({"codes": codes, "warnings": [m for lvl, m in records
+                                                      if lvl == "WARNING" and "not usable" in m]}))
+    """, str(lock_dir), prelude=_DROP_PRIVILEGES)
+    for ok, code, retryable, message in result["codes"]:
+        assert (ok, code, retryable) == (False, "lock_dir_unusable", False)
+        assert "icacls" in message and "/reset" in message and str(lock_dir) in message
+    assert len(result["warnings"]) == 1, result

tools/bot_live_delivery.py

@@ -13,6 +13,7 @@ import re
 import time
 import uuid
 from contextlib import ExitStack, contextmanager
+from hermes_dir_acl import make_user_dir, unusable_dir_remedy
 
 from utils import atomic_json_write, atomic_write_text, fsync_directory
 from pathlib import Path
@@ -158,10 +159,7 @@ def _dir_owner(path: Path) -> str:
 
 
 def _unusable(root: Path, exc: PermissionError) -> MailboxUnusable:
-    if os.name == "nt":
-        remedy = f'from an elevated prompt run: icacls "{root}" /reset /t'
-    else:
-        remedy = f'run: sudo chown -R "$(id -un)" "{root}"'
+    remedy = unusable_dir_remedy(root)
     level = logging.DEBUG if root in _warned_unusable else logging.WARNING
     _warned_unusable.add(root)
     log.log(level, "bot_live_delivery: mailbox directory %s (owner %s) is not usable by this process (%s); "
@@ -173,8 +171,8 @@ def _unusable(root: Path, exc: PermissionError) -> MailboxUnusable:
 def _prepare(root: Path) -> Path:
     """Create/harden the mailbox dir and its ``.lock`` file; returns the lock path."""
     created = not root.is_dir()
-    root.parent.mkdir(parents=True, exist_ok=True)
-    root.mkdir(mode=0o700, exist_ok=True)
+    make_user_dir(root.parent, parents=True, exist_ok=True)
+    make_user_dir(root, mode=0o700, exist_ok=True)
     try:
         root.chmod(0o700)
     except OSError as exc:

tools/computer_use/tool.py

@@ -20,6 +20,7 @@ from collections import namedtuple
 from functools import partial
 from types import SimpleNamespace
 from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
+from hermes_dir_acl import make_user_dir
 
 from agent.image_eviction_policy import image_dimensions_from_bytes
 from tools.computer_use.backend import ActionResult, CaptureResult, ComputerUseBackend, UIElement
@@ -788,7 +789,7 @@ def _secure_dir_policy(cache_dir) -> None:
         if managed:
             cache_dir.mkdir(parents=True, exist_ok=True)
             return
-        cache_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+        make_user_dir(cache_dir, parents=True, exist_ok=True, mode=0o700)
         try:
             from hermes_cli.config import _secure_dir
 

tools/process_registry_output_log.py

@@ -14,6 +14,7 @@ import os
 from pathlib import Path
 import time
 from typing import Optional
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import get_hermes_home
 
@@ -34,7 +35,7 @@ def worker_output_log(session_id: str) -> Optional[Path]:
     if not is_worker_process():
         return None
     directory = output_dir()
-    directory.mkdir(mode=0o700, parents=True, exist_ok=True)
+    make_user_dir(directory, mode=0o700, parents=True, exist_ok=True)
     from tools.process_registry_results import RESULT_RETENTION_SECONDS
     # Only a recorded exit retires a log: a quiet process can run longer than the retention.
     cutoff = time.time() - RESULT_RETENTION_SECONDS

tools/process_registry_results.py

@@ -10,6 +10,7 @@ import logging
 import re
 import sqlite3
 import time
+from hermes_dir_acl import make_user_dir
 
 from hermes_constants import get_hermes_home
 from utils import atomic_json_write
@@ -59,7 +60,7 @@ def save_completed_result(session) -> None:
     try:
         from hermes_constants import assert_named_profile_home_live
         assert_named_profile_home_live(directory)
-        directory.mkdir(mode=0o700, parents=True, exist_ok=True)
+        make_user_dir(directory, mode=0o700, parents=True, exist_ok=True)
         atomic_json_write(directory / f"{session.id}.json", record, mode=0o600)
         _result_paths()
     except OSError:

tools/spill_safety.py

@@ -12,6 +12,7 @@ import os
 import stat
 from pathlib import Path
 from typing import IO
+from hermes_dir_acl import make_user_dir
 
 __all__ = ["ensure_spill_dir", "open_exclusive", "write_text_exclusive"]
 
@@ -24,7 +25,7 @@ def ensure_spill_dir(path: Path, *, private: bool = True) -> Path:
     creates the leaf ``0o700`` and tightens an existing leaf. Raises ``OSError`` if the leaf
     is not a real directory."""
     path = Path(path)
-    path.mkdir(mode=0o700 if private else 0o777, parents=True, exist_ok=True)
+    make_user_dir(path, mode=0o700 if private else 0o777, parents=True, exist_ok=True)
     st = os.lstat(path)
     if not stat.S_ISDIR(st.st_mode):
         raise OSError(f"spill dir is not a directory (symlink?): {path}")

tools/vision_tools.py

@@ -18,6 +18,7 @@ from pathlib import Path
 from typing import Any, Awaitable, Callable, Dict, NamedTuple, Optional
 from urllib.parse import urlparse
 import httpx
+from hermes_dir_acl import make_user_dir
 
 # ``agent.auxiliary_client`` costs ~50 ms cold; only the handlers need it. Both names stay
 # module attributes so tests can patch ``tools.vision_tools.async_call_llm`` (truthy-skip: mocks win).
@@ -215,7 +216,7 @@ def _secure_cache_dir(new_subpath: str, old_name: str) -> Path:
         return cache_dir
     # mode= is honored only for the leaf and is not further masked by umask
     # for the bits we care about; _secure_dir reconciles anything unusual.
-    cache_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+    make_user_dir(cache_dir, parents=True, exist_ok=True, mode=0o700)
     try:
         from hermes_cli.config import _secure_dir