refactor(utils): move mode+owner preservation into atomic_write_text

Follow-up to the salvaged #79323 commits. The three hand-rolled
stat -> atomic_write_text -> chmod blocks (xai migration, uninstaller
shell-rc rewrite, dashboard SOUL.md editor) collapse into an opt-in
preserve_mode=True kwarg on utils.atomic_write_text, plus create_mode=
on both atomic_write_text and atomic_yaml_write for first-create paths
(SOUL.md first save, write_manifest's allowlist create path).

Beyond deduplication this closes two gaps the hand-rolled copies had:

- Owner preservation: the old in-place writes kept the inode, so file
  ownership survived root-run rewrites for free. atomic_write_text
  swaps in a new inode owned by the writing user, and the hand-rolled
  blocks restored only the mode -- a root-run 'hermes migrate xai' or
  sudo uninstall on a user-owned Docker/NAS volume would flip
  config.yaml / ~/.zshrc ownership to root. preserve_mode now routes
  through the same _preserve_file_owner/_restore_file_owner helpers
  atomic_yaml_write and atomic_json_write already use.

- chmod-after-replace window: the mode is applied to the temp fd via
  fchmod BEFORE the replace (mirroring atomic_json_write's mode= param),
  so the target never transits through mkstemp's 0600.

Also removes write_manifest's caller-side existed/chmod block (and its
small TOCTOU) in favor of atomic_yaml_write(create_mode=0o644), and
corrects the SOUL.md mode comment (the default profile's runtime seeder
does run it through _secure_file; named profiles do not).

preserve_mode defaults to False so the existing callers (memory store,
skill manager, cron, agent importer) keep their current semantics.

New tests in tests/test_atomic_write_text_metadata.py cover mode
preservation, owner restore through symlinks, fchmod-before-replace,
create_mode on both writers, and no-behavior-change without opt-in;
all mutation-checked.

hermes_cli/profile_distribution.py

@@ -61,7 +61,6 @@ Update semantics:
 
 from __future__ import annotations
 
-import os
 import re
 import shutil
 import subprocess
@@ -274,26 +273,20 @@ def write_manifest(profile_dir: Path, manifest: DistributionManifest) -> Path:
     # tracking and env_requires with no error surfaced anywhere.
     from utils import atomic_yaml_write
 
-    # atomic_yaml_write preserves an existing file's mode, but a file it
-    # *creates* keeps mkstemp's 0600. _materialize() reaches this line with no
-    # manifest on disk whenever a distribution declares an explicit
-    # `distribution_owned` allowlist that does not list distribution.yaml
-    # itself, so the file is never copied out of the staged tree. The manifest
-    # is a shareable descriptor rather than a secret and used to land at the
-    # umask default, so restore 0644 on the create path only.
-    existed = mf_path.exists()
-
+    # create_mode=0o644: _materialize() reaches this line with no manifest on
+    # disk whenever a distribution declares an explicit `distribution_owned`
+    # allowlist that does not list distribution.yaml itself, so the file is
+    # never copied out of the staged tree. The manifest is a shareable
+    # descriptor rather than a secret and used to land at the umask default,
+    # so don't leave a freshly created one at mkstemp's 0600. An existing
+    # file's mode is preserved as before.
     atomic_yaml_write(
         mf_path,
         manifest.to_dict(),
         sort_keys=False,
         default_flow_style=False,
+        create_mode=0o644,
     )
-    if not existed:
-        try:
-            os.chmod(mf_path, 0o644)
-        except OSError:
-            pass
     return mf_path
 
 

hermes_cli/uninstall.py

@@ -8,7 +8,6 @@ Provides options for:
 
 import os
 import shutil
-import stat
 import subprocess
 import sys
 from pathlib import Path
@@ -98,16 +97,10 @@ def remove_path_from_shell_configs():
                 # that to a warning, so the next login just starts a bare
                 # shell. atomic_replace also resolves a symlinked rc file, so a
                 # dotfiles-repo setup keeps the symlink instead of having it
-                # replaced by a regular file.
-                prior_mode = stat.S_IMODE(config_path.stat().st_mode)
-                atomic_write_text(config_path, new_content)
-                # atomic_write_text swaps in a fresh 0600 temp file; shell rc
-                # files are normally 0644 and removing Hermes' PATH block must
-                # not quietly change their permissions.
-                try:
-                    os.chmod(config_path, prior_mode)
-                except OSError:
-                    pass
+                # replaced by a regular file. preserve_mode keeps the rc's
+                # permission bits (normally 0644) and owner (sudo-run
+                # uninstalls) instead of mkstemp's 0600/root.
+                atomic_write_text(config_path, new_content, preserve_mode=True)
                 removed_from.append(config_path)
                 
         except Exception as e:

hermes_cli/web_routers/profiles.py

@@ -14,8 +14,6 @@ late-binding seam in :mod:`hermes_cli.web_deps` so tests that
 
 import asyncio  # noqa: F401 — used by handlers
 import logging
-import os
-import stat
 import subprocess  # noqa: F401
 import sys  # noqa: F401
 import time  # noqa: F401
@@ -623,28 +621,17 @@ async def update_profile_soul(name: str, body: ProfileSoulUpdate):
         # ``{"content": "", "exists": False}`` -- so an interrupted save shows
         # up as "your persona was never set" and the editor's next Save
         # persists that empty document over it.
-        try:
-            prior_mode = stat.S_IMODE(soul_path.stat().st_mode)
-        except FileNotFoundError:
-            # First save for this profile -- there is no prior file to match,
-            # so fall back to the mode profile creation itself produces:
-            # hermes_cli.profiles seeds SOUL.md with a bare write_text() and
-            # deliberately chmods only .env to 0600. Without this the create
-            # path keeps mkstemp's 0600 and the atomicity fix would silently
-            # tighten the persona document.
-            prior_mode = 0o644
-        except OSError:
-            # stat() failed for some other reason -- do not guess a mode.
-            prior_mode = None
-
-        atomic_write_text(soul_path, body.content)
-        # atomic_write_text swaps in a fresh 0600 temp file; profile SOUL.md is
-        # created 0644 and is not run through _secure_file, so re-apply.
-        if prior_mode is not None:
-            try:
-                os.chmod(soul_path, prior_mode)
-            except OSError:
-                pass
+        #
+        # preserve_mode carries an existing file's permission bits and owner
+        # across the replace. create_mode=0o644 covers the first save: named
+        # profiles seed SOUL.md at the umask default (hermes_cli.profiles
+        # chmods only .env to 0600), and SOUL.md is not a secret. (The default
+        # profile's runtime seeder does run it through _secure_file, but that
+        # seeder fires on every load_config, so the file already exists there
+        # and preserve_mode keeps whatever mode it set.)
+        atomic_write_text(
+            soul_path, body.content, preserve_mode=True, create_mode=0o644
+        )
     except OSError as e:
         _log.exception("PUT /api/profiles/%s/soul failed", name)
         raise HTTPException(status_code=500, detail=f"Could not write SOUL.md: {e}")

hermes_cli/xai_retirement.py

@@ -138,8 +138,6 @@ def format_issue(issue: RetirementIssue) -> str:
 
 import datetime as _dt
 import io
-import os
-import stat
 from pathlib import Path
 import shutil
 
@@ -262,22 +260,11 @@ def apply_migration(
     buf = io.StringIO()
     yaml.dump(doc, buf)
 
-    # atomic_write_text swaps in a fresh 0600 temp file, so carry the existing
-    # permission bits across: _secure_file deliberately leaves config.yaml
-    # alone under managed (NixOS 0640) and container installs, and a migration
-    # must not silently tighten what those setups widened.
-    try:
-        prior_mode = stat.S_IMODE(config_path.stat().st_mode)
-    except OSError:
-        prior_mode = None
-
-    atomic_write_text(config_path, buf.getvalue())
-
-    if prior_mode is not None:
-        try:
-            os.chmod(config_path, prior_mode)
-        except OSError:
-            pass
+    # preserve_mode carries the existing permission bits AND owner across the
+    # replace: _secure_file deliberately leaves config.yaml alone under managed
+    # (NixOS 0640) and container installs, and a root-run migration on a
+    # user-owned volume must not flip ownership to root.
+    atomic_write_text(config_path, buf.getvalue(), preserve_mode=True)
 
     return ApplyResult(
         file_path=config_path,

tests/test_atomic_write_text_metadata.py

@@ -0,0 +1,159 @@
+"""``atomic_write_text``'s opt-in metadata preservation (mode + owner).
+
+``os.replace`` swaps mkstemp's 0600 temp file (owned by the writing user)
+onto the target, so a bare atomic rewrite of an existing user-authored file
+tightens its permission bits and — for root-run callers on Docker/NAS
+volumes — flips its ownership.  ``preserve_mode=True`` carries both across
+the replace, exactly like ``atomic_yaml_write`` does unconditionally;
+``create_mode=`` sets the bits when the target does not exist yet.
+
+These guard the follow-up to PR #79323, which collapsed three hand-rolled
+stat/write/chmod blocks (xai migration, uninstaller shell-rc rewrite,
+dashboard SOUL.md editor) into these kwargs.
+"""
+
+from __future__ import annotations
+
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+from utils import atomic_write_text, atomic_yaml_write
+
+
+pytestmark = pytest.mark.skipif(
+    sys.platform == "win32", reason="POSIX permission bits"
+)
+
+
+class TestPreserveMode:
+    def test_existing_mode_survives_the_rewrite(self, tmp_path: Path) -> None:
+        """A 0640 managed config must not tighten to mkstemp's 0600."""
+        target = tmp_path / "config.yaml"
+        target.write_text("old: true\n", encoding="utf-8")
+        os.chmod(target, 0o640)
+
+        atomic_write_text(target, "new: true\n", preserve_mode=True)
+
+        assert target.read_text(encoding="utf-8") == "new: true\n"
+        assert stat.S_IMODE(target.stat().st_mode) == 0o640
+
+    def test_default_still_leaves_mkstemp_mode(self, tmp_path: Path) -> None:
+        """Without opt-in, behavior is unchanged: the file lands 0600."""
+        target = tmp_path / "notes.md"
+        target.write_text("old\n", encoding="utf-8")
+        os.chmod(target, 0o644)
+
+        atomic_write_text(target, "new\n")
+
+        assert stat.S_IMODE(target.stat().st_mode) == 0o600
+
+    def test_mode_is_applied_before_the_replace(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """The temp fd gets fchmod'd, so the target never transits 0600."""
+        target = tmp_path / "config.yaml"
+        target.write_text("old\n", encoding="utf-8")
+        os.chmod(target, 0o640)
+
+        import utils as utils_mod
+
+        real_replace = utils_mod.atomic_replace
+        seen: list[int] = []
+
+        def spying_replace(tmp, dst):
+            seen.append(stat.S_IMODE(os.stat(tmp).st_mode))
+            return real_replace(tmp, dst)
+
+        monkeypatch.setattr(utils_mod, "atomic_replace", spying_replace)
+        atomic_write_text(target, "new\n", preserve_mode=True)
+
+        assert seen == [0o640]
+
+    def test_owner_is_restored_on_the_real_symlink_target(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Root-run rewrites of a user-owned file must not flip ownership.
+
+        Mirrors test_atomic_yaml_write_restores_owner_on_real_symlink_target:
+        forces a preserved uid/gid so the test does not need root.
+        """
+        real = tmp_path / "zshrc"
+        link = tmp_path / ".zshrc"
+        real.write_text("export A=1\n", encoding="utf-8")
+        link.symlink_to(real)
+
+        chown_calls: list[tuple[Path, int, int]] = []
+        monkeypatch.setattr("utils._preserve_file_owner", lambda _p: (123, 456))
+        monkeypatch.setattr(
+            "utils.os.chown",
+            lambda path, uid, gid: chown_calls.append((Path(path), uid, gid)),
+        )
+
+        atomic_write_text(link, "export B=2\n", preserve_mode=True)
+
+        assert chown_calls == [(real, 123, 456)]
+        assert link.is_symlink()
+        assert real.read_text(encoding="utf-8") == "export B=2\n"
+
+    def test_no_owner_calls_without_opt_in(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        target = tmp_path / "mem.md"
+        target.write_text("old\n", encoding="utf-8")
+
+        chown_calls: list[tuple] = []
+        monkeypatch.setattr("utils._preserve_file_owner", lambda _p: (123, 456))
+        monkeypatch.setattr(
+            "utils.os.chown", lambda *a: chown_calls.append(a)
+        )
+
+        atomic_write_text(target, "new\n")
+
+        assert chown_calls == []
+
+
+class TestCreateMode:
+    def test_create_mode_applies_when_target_is_new(self, tmp_path: Path) -> None:
+        target = tmp_path / "SOUL.md"
+        assert not target.exists()
+
+        atomic_write_text(
+            target, "# Persona\n", preserve_mode=True, create_mode=0o644
+        )
+
+        assert stat.S_IMODE(target.stat().st_mode) == 0o644
+
+    def test_existing_mode_beats_create_mode(self, tmp_path: Path) -> None:
+        target = tmp_path / "SOUL.md"
+        target.write_text("old\n", encoding="utf-8")
+        os.chmod(target, 0o600)
+
+        atomic_write_text(
+            target, "new\n", preserve_mode=True, create_mode=0o644
+        )
+
+        assert stat.S_IMODE(target.stat().st_mode) == 0o600
+
+    def test_atomic_yaml_write_create_mode(self, tmp_path: Path) -> None:
+        """write_manifest's create path: new file lands 0644, not 0600."""
+        target = tmp_path / "distribution.yaml"
+        assert not target.exists()
+
+        atomic_yaml_write(target, {"name": "t"}, create_mode=0o644)
+
+        assert stat.S_IMODE(target.stat().st_mode) == 0o644
+
+    def test_atomic_yaml_write_existing_mode_beats_create_mode(
+        self, tmp_path: Path
+    ) -> None:
+        target = tmp_path / "distribution.yaml"
+        target.write_text("name: old\n", encoding="utf-8")
+        os.chmod(target, 0o600)
+
+        atomic_yaml_write(target, {"name": "new"}, create_mode=0o644)
+
+        assert stat.S_IMODE(target.stat().st_mode) == 0o600

utils.py

@@ -142,6 +142,8 @@ def atomic_write_text(
     *,
     encoding: str = "utf-8",
     tmp_prefix: str = ".tmp_",
+    preserve_mode: bool = False,
+    create_mode: "int | None" = None,
 ) -> None:
     """Write *content* to *path* via temp file + fsync + atomic rename.
 
@@ -151,18 +153,45 @@ def atomic_write_text(
 
     Used by the memory store, skill manager, and agent importer so that
     every destructive file rewrite in the codebase shares one implementation.
+
+    Args:
+        preserve_mode: When True, carry an existing target's permission bits
+            and (POSIX, best-effort) owner across the replace, like
+            ``atomic_yaml_write`` does unconditionally.  ``os.replace`` swaps
+            in mkstemp's 0600 temp file owned by the writing user, so without
+            this a root-run rewrite of a user-owned file flips its owner and
+            tightens its mode.  The mode is applied to the temp fd *before*
+            the replace, so the file never transits through 0600.  Off by
+            default: the historical callers (memory store, skill manager,
+            cron) own their 0600-is-fine files.
+        create_mode: Permission bits to apply when the target does not yet
+            exist (otherwise the new file keeps mkstemp's 0600).  Ignored
+            when ``preserve_mode`` found an existing mode to carry over.
     """
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
+
+    original_mode = _preserve_file_mode(path) if preserve_mode else None
+    original_owner = _preserve_file_owner(path) if preserve_mode else None
+    effective_mode = original_mode if original_mode is not None else create_mode
+
     fd, tmp_path = tempfile.mkstemp(
         dir=str(path.parent), prefix=tmp_prefix, suffix=".tmp"
     )
     try:
+        if effective_mode is not None and hasattr(os, "fchmod"):
+            # fchmod is Unix-only; on Windows the post-replace chmod below
+            # applies the final mode instead.
+            os.fchmod(fd, effective_mode)
         with os.fdopen(fd, "w", encoding=encoding) as handle:
             handle.write(content)
             handle.flush()
             os.fsync(handle.fileno())
-        atomic_replace(tmp_path, path)
+        real_path = atomic_replace(tmp_path, path)
+        if preserve_mode:
+            _restore_file_owner(Path(real_path), original_owner)
+        if effective_mode is not None and not hasattr(os, "fchmod"):
+            _restore_file_mode(Path(real_path), effective_mode)
     except BaseException:
         try:
             os.unlink(tmp_path)
@@ -307,6 +336,7 @@ def atomic_yaml_write(
     default_flow_style: bool = False,
     sort_keys: bool = False,
     extra_content: str | None = None,
+    create_mode: "int | None" = None,
 ) -> None:
     """Write YAML data to a file atomically.
 
@@ -321,12 +351,17 @@ def atomic_yaml_write(
         sort_keys: Whether to sort dict keys (default False).
         extra_content: Optional string to append after the YAML dump
             (e.g. commented-out sections for user reference).
+        create_mode: Permission bits to apply when the target does not yet
+            exist (a created file otherwise keeps mkstemp's 0600).  An
+            existing file's mode is always preserved and wins over this.
     """
     path = Path(path)
     path.parent.mkdir(parents=True, exist_ok=True)
 
     original_mode = _preserve_file_mode(path)
     original_owner = _preserve_file_owner(path)
+    if original_mode is None:
+        original_mode = create_mode
 
     fd, tmp_path = tempfile.mkstemp(
         dir=str(path.parent),