refactor: extract atomic_write_text to utils.py; fix write-failure error handling

Deduplicates the mkstemp→fsync→atomic_replace pattern that existed in
three places: agent_import.py (added by #72983), MemoryStore._write_file,
and skill_manager_tool._atomic_write_text. All three now call a single
utils.atomic_write_text helper.

Also wraps the atomic_write_text call in _merge_memory_entries with
try/except OSError so a write failure records a per-item error instead
of propagating uncaught and aborting the entire import with no record.

Follow-up to #72983.

hermes_cli/agent_import.py

@@ -40,16 +40,14 @@ from __future__ import annotations
 
 import json
 import logging
-import os
 import re
 import shutil
 import sys
-import tempfile
 import time
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Sequence, Tuple
 
-from utils import atomic_replace
+from utils import atomic_write_text
 
 logger = logging.getLogger(__name__)
 
@@ -249,30 +247,6 @@ def backup_memory_file(path: Path) -> Optional[Path]:
     return backup
 
 
-def atomic_write_text(path: Path, content: str) -> None:
-    """Write ``content`` to ``path`` via temp file + atomic rename.
-
-    Mirrors ``MemoryStore._write_file``: an interrupted or failed write can
-    never leave a truncated memory store on disk, and readers always see
-    either the old complete file or the new one.  ``atomic_replace`` also
-    keeps a symlinked destination a symlink.
-    """
-    fd, tmp_path = tempfile.mkstemp(
-        dir=str(path.parent), suffix=".tmp", prefix=".import_"
-    )
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            handle.write(content)
-            handle.flush()
-            os.fsync(handle.fileno())
-        atomic_replace(tmp_path, path)
-    except BaseException:
-        try:
-            os.unlink(tmp_path)
-        except OSError:
-            pass
-        raise
-
 
 def merge_entries(
     existing: Sequence[str],
@@ -582,10 +556,16 @@ class AgentImporter:
                 return
             if backup is not None:
                 details["backup"] = str(backup)
-            atomic_write_text(
-                destination,
-                ENTRY_DELIMITER.join(merged) + ("\n" if merged else ""),
-            )
+            try:
+                atomic_write_text(
+                    destination,
+                    ENTRY_DELIMITER.join(merged) + ("\n" if merged else ""),
+                )
+            except OSError as exc:
+                self.record(kind, source, destination, "error",
+                            f"Could not write merged memory file: {exc}",
+                            **details)
+                return
             self.record(kind, source, destination, "imported", **details)
         else:
             self.record(kind, source, destination, "imported",

tools/memory_tool.py

@@ -25,15 +25,13 @@ Design:
 
 import json
 import logging
-import os
-import tempfile
 import time
 from contextlib import contextmanager
 from pathlib import Path
 from hermes_constants import get_hermes_home
 from typing import Dict, Any, List, Optional, Tuple
 
-from utils import atomic_replace
+from utils import atomic_write_text
 
 # fcntl is Unix-only; on Windows use msvcrt for file locking
 msvcrt = None
@@ -873,23 +871,7 @@ class MemoryStore:
         """
         content = ENTRY_DELIMITER.join(entries) if entries else ""
         try:
-            # Write to temp file in same directory (same filesystem for atomic rename)
-            fd, tmp_path = tempfile.mkstemp(
-                dir=str(path.parent), suffix=".tmp", prefix=".mem_"
-            )
-            try:
-                with os.fdopen(fd, "w", encoding="utf-8") as f:
-                    f.write(content)
-                    f.flush()
-                    os.fsync(f.fileno())
-                atomic_replace(tmp_path, path)
-            except BaseException:
-                # Clean up temp file on any failure
-                try:
-                    os.unlink(tmp_path)
-                except OSError:
-                    pass
-                raise
+            atomic_write_text(path, content, tmp_prefix=".mem_")
         except (OSError, IOError) as e:
             raise RuntimeError(f"Failed to write memory file {path}: {e}")
 

tools/skill_manager_tool.py

@@ -34,16 +34,14 @@ Directory layout for user skills:
 
 import json
 import logging
-import os
 import re
 import shutil
-import tempfile
 import contextvars as _ctxvars
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
 from hermes_constants import get_hermes_home, display_hermes_home
-from utils import atomic_replace, is_truthy_value
+from utils import atomic_write_text, is_truthy_value
 from hermes_cli.config import cfg_get
 from agent.skill_utils import (
     extract_skill_description,
@@ -817,35 +815,13 @@ def _resolve_skill_target(skill_dir: Path, file_path: str) -> Tuple[Optional[Pat
 
 
 def _atomic_write_text(file_path: Path, content: str, encoding: str = "utf-8") -> None:
+    """Atomically write text content to a file.
+
+    Thin wrapper around :func:`utils.atomic_write_text` so that every
+    destructive file rewrite in the codebase shares one implementation.
     """
-    Atomically write text content to a file.
-    
-    Uses a temporary file in the same directory and os.replace() to ensure
-    the target file is never left in a partially-written state if the process
-    crashes or is interrupted.
-    
-    Args:
-        file_path: Target file path
-        content: Content to write
-        encoding: Text encoding (default: utf-8)
-    """
-    file_path.parent.mkdir(parents=True, exist_ok=True)
-    fd, temp_path = tempfile.mkstemp(
-        dir=str(file_path.parent),
-        prefix=f".{file_path.name}.tmp.",
-        suffix="",
-    )
-    try:
-        with os.fdopen(fd, "w", encoding=encoding) as f:
-            f.write(content)
-        atomic_replace(temp_path, file_path)
-    except Exception:
-        # Clean up temp file on error
-        try:
-            os.unlink(temp_path)
-        except OSError:
-            logger.error("Failed to remove temporary file %s during atomic write", temp_path, exc_info=True)
-        raise
+    atomic_write_text(file_path, content, encoding=encoding,
+                      tmp_prefix=f".{file_path.name}.tmp.")
 
 
 # =============================================================================

utils.py

@@ -136,6 +136,41 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
     return real_path
 
 
+def atomic_write_text(
+    path: Union[str, Path],
+    content: str,
+    *,
+    encoding: str = "utf-8",
+    tmp_prefix: str = ".tmp_",
+) -> None:
+    """Write *content* to *path* via temp file + fsync + atomic rename.
+
+    Ensures the target file is never left in a partially-written state if
+    the process crashes or is interrupted.  ``atomic_replace`` preserves
+    symlinks and handles cross-device / busy-file fallbacks.
+
+    Used by the memory store, skill manager, and agent importer so that
+    every destructive file rewrite in the codebase shares one implementation.
+    """
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_path = tempfile.mkstemp(
+        dir=str(path.parent), prefix=tmp_prefix, suffix=".tmp"
+    )
+    try:
+        with os.fdopen(fd, "w", encoding=encoding) as handle:
+            handle.write(content)
+            handle.flush()
+            os.fsync(handle.fileno())
+        atomic_replace(tmp_path, path)
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
 def atomic_json_write(
     path: Union[str, Path],
     data: Any,