refactor: share one atomic file write helper across stores

Replace the per-module sibling-file-and-rename copies with
common/atomic_write.py, which uses a unique dot-prefixed temp name,
fsyncs, and keeps the target's permissions.

Co-authored-by: cowagent <cow@cowagent.ai>

agent/memory/summarizer.py

@@ -9,11 +9,11 @@ Handles memory persistence when conversation context is trimmed or overflows:
 - Deep Dream: periodically distills daily memories → refined MEMORY.md + dream diary
 """
 
-import os
 import threading
 from typing import Optional, Callable, Any, List, Dict
 from pathlib import Path
 from datetime import datetime
+from common.atomic_write import write_text_atomic
 from common.log import logger
 
 
@@ -191,32 +191,6 @@ def _is_empty_sentinel(text: str) -> bool:
     return s == "" or s == "无" or s.lower() == "none"
 
 
-def _write_text_atomic(path: Path, text: str) -> None:
-    """Write ``text`` to ``path`` through a sibling file, then swap it in.
-
-    Both callers replace a file the user only has one copy of. Writing straight
-    into ``path`` truncates it first, so anything that fails halfway -- a full
-    disk, the process being killed -- leaves a partial file behind and the
-    previous content is unrecoverable. Building the text beside the file and
-    replacing it means a failed save leaves whatever was there before.
-    """
-    temporary = f"{path}.tmp"
-    try:
-        with open(temporary, "w", encoding="utf-8") as handle:
-            handle.write(text)
-            handle.flush()
-            os.fsync(handle.fileno())
-        os.replace(temporary, path)
-    except Exception:
-        try:
-            if os.path.exists(temporary):
-                os.remove(temporary)
-        except OSError:
-            pass
-        raise
-
-
-
 class MemoryFlushManager:
     """
     Manages memory flush operations.
@@ -583,7 +557,7 @@ class MemoryFlushManager:
         try:
             main_file = self.get_main_memory_file(user_id)
             old_size = len(memory_content)
-            _write_text_atomic(main_file, new_memory + "\n")
+            write_text_atomic(main_file, new_memory + "\n")
             logger.info(
                 f"[DeepDream] Updated MEMORY.md "
                 f"({old_size} → {len(new_memory)} chars)"
@@ -672,7 +646,7 @@ class MemoryFlushManager:
 
         today = datetime.now().strftime("%Y-%m-%d")
         diary_file = dreams_dir / f"{today}.md"
-        _write_text_atomic(diary_file, f"# Dream Diary: {today}\n\n{content}\n")
+        write_text_atomic(diary_file, f"# Dream Diary: {today}\n\n{content}\n")
         logger.info(f"[DeepDream] Wrote dream diary to {diary_file}")
 
     # ---- Internal helpers ----

agent/team.py

@@ -20,11 +20,10 @@ first write migrates them and clears the old keys.
 from __future__ import annotations
 
 import json
-import os
-import tempfile
 from pathlib import Path
 from typing import Any, Dict, Iterable, Mapping, Optional
 
+from common.atomic_write import write_text_atomic
 from common.log import logger
 from common.utils import expand_path
 
@@ -150,20 +149,7 @@ def write(settings: Mapping[str, Any], roster: Mapping[str, Any]) -> Path:
     )
     if bootstrapped:
         payload["channel_instances"] = bootstrapped
-    fd, tmp_name = tempfile.mkstemp(prefix=f".{FILE_NAME}.", suffix=".tmp", dir=str(path.parent))
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            json.dump(payload, handle, indent=4, ensure_ascii=False)
-            handle.write("\n")
-            handle.flush()
-            os.fsync(handle.fileno())
-        os.replace(tmp_name, path)
-    except Exception:
-        try:
-            os.unlink(tmp_name)
-        except OSError:
-            pass
-        raise
+    write_text_atomic(path, json.dumps(payload, indent=4, ensure_ascii=False) + "\n")
     return path
 
 
@@ -233,32 +219,6 @@ def adopt_legacy_channels(settings: Mapping[str, Any]) -> Optional[Path]:
     return written
 
 
-def _write_atomically(path: Path, payload: str) -> None:
-    """Put ``payload`` into ``path`` through a sibling file, then swap it in.
-
-    The same reason ``write()`` above does it: opening ``path`` for writing cuts
-    it short before the new bytes are there, so anything that stops the write
-    leaves a partial config.json behind. ``load_config`` then reads it as
-    corruption, and on the desktop client the self-heal path quarantines the
-    file and reinitialises from config-template.json, which takes every API key,
-    channel credential and custom provider with it; a source deployment raises
-    instead of starting at all.
-    """
-    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            handle.write(payload)
-            handle.flush()
-            os.fsync(handle.fileno())
-        os.replace(tmp_name, path)
-    except Exception:
-        try:
-            os.unlink(tmp_name)
-        except OSError:
-            pass
-        raise
-
-
 def retire_legacy(config_path: Optional[Path]) -> None:
     """Take the roster keys out of ``config.json``, once the file has them.
 
@@ -278,7 +238,7 @@ def retire_legacy(config_path: Optional[Path]) -> None:
             return
         for key in TEAM_KEYS:
             settings.pop(key, None)
-        _write_atomically(
+        write_text_atomic(
             path, json.dumps(settings, indent=4, ensure_ascii=False) + "\n"
         )
     except (OSError, ValueError) as e:

agent/tools/scheduler/task_store.py

@@ -7,6 +7,7 @@ import os
 import threading
 from datetime import datetime
 from typing import Dict, List, Optional
+from common.atomic_write import write_text_atomic
 from common.utils import expand_path
 
 
@@ -66,32 +67,6 @@ class TaskStore:
         """Ensure the storage directory exists"""
         store_dir = os.path.dirname(self.store_path)
         os.makedirs(store_dir, exist_ok=True)
-
-    @staticmethod
-    def _write_atomic(path: str, text: str) -> None:
-        """Write ``text`` to ``path`` through a sibling file, then swap it in.
-
-        Writing straight into ``path`` truncates it first, so anything that
-        fails while serialising -- a value json cannot encode, a full disk --
-        leaves it empty and every stored task is gone on the next load, which
-        swallows the decode error and reports no tasks at all. Building the
-        result beside the file and replacing it means a failed save leaves
-        whatever was there before.
-        """
-        temporary = f"{path}.tmp"
-        try:
-            with open(temporary, "w", encoding="utf-8") as handle:
-                handle.write(text)
-                handle.flush()
-                os.fsync(handle.fileno())
-            os.replace(temporary, path)
-        except Exception:
-            try:
-                if os.path.exists(temporary):
-                    os.remove(temporary)
-            except OSError:
-                pass
-            raise
     
     def load_tasks(self) -> Dict[str, dict]:
         """
@@ -133,7 +108,7 @@ class TaskStore:
                     try:
                         with open(self.store_path, 'r', encoding='utf-8') as src:
                             previous = src.read()
-                        self._write_atomic(backup_path, previous)
+                        write_text_atomic(backup_path, previous)
                     except Exception:
                         pass
 
@@ -144,7 +119,7 @@ class TaskStore:
                     "tasks": tasks
                 }
 
-                self._write_atomic(
+                write_text_atomic(
                     self.store_path, json.dumps(data, ensure_ascii=False, indent=2)
                 )
             except Exception as e:

cli/commands/skill.py

@@ -3,7 +3,6 @@
 import os
 import re
 import sys
-import threading
 import json
 import hashlib
 import shutil
@@ -19,6 +18,7 @@ import click
 import requests
 
 from cli.utils import (
+    _ensure_project_on_path,
     get_skills_dir,
     get_builtin_skills_dir,
     load_skills_config,
@@ -350,41 +350,11 @@ def _install_local(path: str, result: InstallResult, agent_id: str = None):
 
 
 def _write_skills_config(config: dict, skills_dir: str) -> None:
-    """Persist skills_config.json through a sibling file, then swap it in.
+    _ensure_project_on_path()
+    from common.atomic_write import write_json_atomic
 
-    Serialising straight into skills_config.json truncates it before the new
-    bytes are there, so anything that fails halfway through -- a full disk, the
-    process being killed -- leaves a half-written document behind.
-    ``load_skills_config`` cannot parse that and answers ``{}``, so the next
-    install, enable or uninstall writes its own baseline back over the file and
-    every other skill's entry is gone with it. Building the result beside the
-    file and renaming it into place means a failed save leaves the previous
-    document untouched.
-
-    The sibling is named the way ``agent/skills/manager.py`` names its own --
-    dot-prefixed and unique per writer -- because that is the shape
-    ``agent/evolution/executor.py`` lists in ``_WATCH_IGNORE_PREFIXES``: the
-    workspace scanner must not read an in-flight replacement as a real change,
-    and two overlapping saves must not rename each other's partial file.
-    """
-    config_path = os.path.join(skills_dir, "skills_config.json")
     os.makedirs(skills_dir, exist_ok=True)
-    tmp_path = os.path.join(
-        skills_dir,
-        f".skills_config.json.{os.getpid()}.{threading.get_ident()}.tmp",
-    )
-    try:
-        with open(tmp_path, "w", encoding="utf-8") as f:
-            json.dump(config, f, indent=4, ensure_ascii=False)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp_path, config_path)
-    except BaseException:
-        try:
-            os.remove(tmp_path)
-        except OSError:
-            pass
-        raise
+    write_json_atomic(os.path.join(skills_dir, "skills_config.json"), config)
 
 
 def _register_installed_skill(name: str, source: str = "cowhub", display_name: str = "", agent_id: str = None):

common/atomic_write.py

@@ -0,0 +1,50 @@
+"""Replace a file's contents without ever leaving it half-written.
+
+Writing straight into a file truncates it before the new bytes exist, so a
+crash, a kill or a full disk in between leaves a partial document that most
+stores here cannot parse back. These helpers write a sibling first and rename
+it over the target, so readers see either the old file or the new one.
+
+The sibling is dot-prefixed (workspace watchers skip it) and unique per call,
+so overlapping writers never rename each other's partial file.
+"""
+
+import json
+import os
+import stat
+import uuid
+
+
+def write_text_atomic(path, text: str, encoding: str = "utf-8") -> None:
+    _replace(path, lambda f: f.write(text), encoding)
+
+
+def write_json_atomic(path, data, indent: int = 4, ensure_ascii: bool = False) -> None:
+    _replace(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii))
+
+
+def _replace(path, write, encoding: str = "utf-8") -> None:
+    path = os.fspath(path)
+    directory, name = os.path.split(os.path.abspath(path))
+    tmp_path = os.path.join(directory, f".{name}.{uuid.uuid4().hex[:12]}.tmp")
+    try:
+        with open(tmp_path, "w", encoding=encoding) as f:
+            write(f)
+            f.flush()
+            os.fsync(f.fileno())
+        _copy_mode(path, tmp_path)
+        os.replace(tmp_path, path)
+    except BaseException:
+        try:
+            os.remove(tmp_path)
+        except OSError:
+            pass
+        raise
+
+
+def _copy_mode(src: str, dst: str) -> None:
+    """Keep the target's permissions, e.g. 0600 on a credentials file."""
+    try:
+        os.chmod(dst, stat.S_IMODE(os.stat(src).st_mode))
+    except OSError:
+        pass

models/model_catalog.py

@@ -38,6 +38,7 @@ simply its whole model list and ``hidden`` stays empty.
 import json
 import os
 
+from common.atomic_write import write_json_atomic
 from common.log import logger
 
 # Legacy: catalogs used to live under this key in config.json (pre-overlay),
@@ -182,7 +183,7 @@ def _strip_legacy_config_key() -> None:
         return
     if LEGACY_CATALOG_KEY in data:
         data.pop(LEGACY_CATALOG_KEY, None)
-        _write_json_atomic(cfg_path, data)
+        write_json_atomic(cfg_path, data)
 
 
 def _read_store() -> dict:
@@ -215,40 +216,10 @@ def _read_store() -> dict:
     return {"providers": providers}
 
 
-def _write_json_atomic(path: str, data: dict) -> None:
-    """Write ``data`` to ``path`` through a sibling file, then swap it in.
-
-    Both files written from this module are read back with a bare json.load, and
-    a partial one is worse than a missing one. Writing straight into the target
-    truncates it before the new bytes exist, so anything that fails while
-    serialising -- a full disk, an interrupted update -- leaves a document that
-    no longer parses. For the overlay store that reads as "the user has no
-    overrides at all" (``_read_store`` deliberately swallows the decode error so
-    one bad file cannot take the whole models view down), and the next save
-    writes that empty baseline back, which turns a recoverable file into a
-    permanent loss. The config.json side is the file the console also writes: a
-    truncated user config is what ``load_config`` treats as corruption.
-    """
-    temporary = f"{path}.tmp"
-    try:
-        with open(temporary, "w", encoding="utf-8") as f:
-            json.dump(data, f, indent=4, ensure_ascii=False)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(temporary, path)
-    except Exception:
-        try:
-            if os.path.exists(temporary):
-                os.remove(temporary)
-        except OSError:
-            pass
-        raise
-
-
 def _write_store(store: dict) -> None:
     path = _store_path()
     os.makedirs(os.path.dirname(path), exist_ok=True)
-    _write_json_atomic(path, store)
+    write_json_atomic(path, store)
 
 
 # In-memory cache so the budget/request paths (called every LLM turn) don't hit

plugins/banwords/banwords.py

@@ -1,11 +1,11 @@
 # encoding:utf-8
 
-import json
 import os
 
 import plugins
 from bridge.context import ContextType
 from bridge.reply import Reply, ReplyType
+from common.atomic_write import write_json_atomic
 from common.log import logger
 from plugins import *
 
@@ -16,33 +16,6 @@ from .lib.WordsSearch import WordsSearch
 DEFAULT_CONFIG = {"action": "ignore"}
 
 
-def _write_config_atomically(config_path: str, conf: dict) -> None:
-    """Write *conf* beside the file, then rename it into place.
-
-    The repair that calls this runs precisely because the file on disk is
-    already unreadable, so truncating it first can turn "empty" into
-    "unparseable" -- and ``activate_plugins`` answers a plugin that fails to
-    initialise by persisting ``enabled=false``, the outcome this fallback
-    exists to avoid. A write that dies halfway now leaves whatever the user
-    had, so the plugin still starts from the defaults held in memory.
-    """
-    tmp_path = f"{config_path}.tmp"
-    try:
-        with open(tmp_path, "w", encoding="utf-8") as f:
-            json.dump(conf, f, indent=4)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp_path, config_path)
-    except BaseException:
-        try:
-            os.remove(tmp_path)
-        except OSError:
-            pass
-        raise
-
-
-
-
 @plugins.register(
     name="Banwords",
     desire_priority=100,
@@ -68,7 +41,7 @@ class Banwords(Plugin):
                 conf = {**DEFAULT_CONFIG, **(conf if isinstance(conf, dict) else {})}
                 config_path = os.path.join(curdir, "config.json")
                 try:
-                    _write_config_atomically(config_path, conf)
+                    write_json_atomic(config_path, conf)
                 except OSError as e:
                     # Repairing the file on disk is a convenience; the defaults
                     # above are enough to run. Raising here would reach

plugins/cow_cli/cow_cli.py

@@ -15,7 +15,6 @@ Does NOT match:
   /开头但不是已知命令
 """
 
-import json
 import os
 import threading
 
@@ -23,6 +22,7 @@ import plugins
 from plugins import Plugin, Event, EventContext, EventAction
 from bridge.context import ContextType
 from bridge.reply import Reply, ReplyType
+from common.atomic_write import write_json_atomic
 from common.log import logger
 from common.i18n import t as _t
 from config import conf
@@ -85,40 +85,6 @@ _HUB_OFF_MSG = (
 _HUB_HELP_PREFIXES = ("/skill list --remote", "/skill search", "/skill install")
 
 
-def _write_json_atomically(config_path: str, data) -> None:
-    """Serialise *data* beside *config_path*, then rename it into place.
-
-    ``open(path, "w")`` truncates before it writes, so a crash, a ``kill``, or
-    a full disk between the truncate and the final flush leaves a half-written
-    file behind. For ``config.json`` that is not recoverable: ``load_config``
-    deliberately re-raises on a parse error outside the packaged desktop build
-    (``config.py``), so the next start dies on "Initialization Failed" with no
-    way out for the user.
-
-    The sibling carries a leading dot and a per-writer suffix, matching the
-    other stores here (``agent/skills/manager.py``, ``channel/web/core/_common.py``):
-    the workspace scanner skips dot-prefixed files, so an in-flight replacement
-    is never mistaken for a real change, and two overlapping saves cannot
-    rename each other's half-written file.
-    """
-    tmp_path = os.path.join(
-        os.path.dirname(config_path),
-        f".{os.path.basename(config_path)}.{os.getpid()}.{threading.get_ident()}.tmp",
-    )
-    try:
-        with open(tmp_path, "w", encoding="utf-8") as f:
-            json.dump(data, f, indent=4, ensure_ascii=False)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp_path, config_path)
-    except BaseException:
-        try:
-            os.remove(tmp_path)
-        except OSError:
-            pass
-        raise
-
-
 @plugins.register(
     name="cow_cli",
     desc="Handle cow/slash commands in chat messages",
@@ -882,7 +848,7 @@ class CowCliPlugin(Plugin):
             with open(config_path, "r", encoding="utf-8-sig") as f:
                 file_config = _json.load(f)
             file_config.update(updates)
-            _write_json_atomically(config_path, file_config)
+            write_json_atomic(config_path, file_config)
         except Exception as e:
             return _t(f"写入 config.json 失败: {e}", f"Failed to write config.json: {e}")
 
@@ -1332,7 +1298,7 @@ class CowCliPlugin(Plugin):
                 with open(config_path, "r", encoding="utf-8") as f:
                     config = json.load(f)
                 config.pop(name, None)
-                _write_json_atomically(config_path, config)
+                write_json_atomic(config_path, config)
             except Exception:
                 pass
 
@@ -1453,7 +1419,7 @@ class CowCliPlugin(Plugin):
             return _t(f"技能 '{name}' 未在配置中找到", f"Skill '{name}' not found in config")
 
         config[name]["enabled"] = enabled
-        _write_json_atomically(config_path, config)
+        write_json_atomic(config_path, config)
 
         icon = "✅" if enabled else "⬚"
         if enabled:
@@ -1891,7 +1857,7 @@ class CowCliPlugin(Plugin):
             with open(config_path, "r", encoding="utf-8-sig") as f:
                 file_config = _json.load(f)
             file_config["knowledge"] = enabled
-            _write_json_atomically(config_path, file_config)
+            write_json_atomic(config_path, file_config)
         except Exception as e:
             return _t(f"⚠️ 内存中已切换，但写入 config.json 失败: {e}", f"⚠️ Switched in memory, but failed to write config.json: {e}")
 

plugins/godcmd/godcmd.py

@@ -1,6 +1,5 @@
 # encoding:utf-8
 
-import json
 import os
 import random
 import string
@@ -12,6 +11,7 @@ from bridge.bridge import Bridge
 from bridge.context import ContextType
 from bridge.reply import Reply, ReplyType
 from common import const
+from common.atomic_write import write_json_atomic
 from config import conf, load_config, global_config
 from plugins import *
 
@@ -176,33 +176,6 @@ def get_help_text(isadmin, isgroup):
 DEFAULT_CONFIG = {"password": "", "admin_users": []}
 
 
-def _write_config_atomically(config_path: str, conf: dict) -> None:
-    """Write *conf* beside the file, then rename it into place.
-
-    The repair that calls this runs precisely because the file on disk is
-    already unreadable, so truncating it first can turn "empty" into
-    "unparseable" -- and ``activate_plugins`` answers a plugin that fails to
-    initialise by persisting ``enabled=false``, the outcome this fallback
-    exists to avoid. A write that dies halfway now leaves whatever the user
-    had, so the plugin still starts from the defaults held in memory.
-    """
-    tmp_path = f"{config_path}.tmp"
-    try:
-        with open(tmp_path, "w", encoding="utf-8") as f:
-            json.dump(conf, f, indent=4)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(tmp_path, config_path)
-    except BaseException:
-        try:
-            os.remove(tmp_path)
-        except OSError:
-            pass
-        raise
-
-
-
-
 @plugins.register(
     name="Godcmd",
     desire_priority=999,
@@ -226,7 +199,7 @@ class Godcmd(Plugin):
         if not isinstance(gconf, dict) or not all(k in gconf for k in DEFAULT_CONFIG):
             gconf = {**DEFAULT_CONFIG, **(gconf if isinstance(gconf, dict) else {})}
             try:
-                _write_config_atomically(config_path, gconf)
+                write_json_atomic(config_path, gconf)
             except OSError as e:
                 # Repairing the file on disk is a convenience; the defaults above
                 # are enough to run. Raising here would reach activate_plugins,

plugins/plugin.py

@@ -1,37 +1,10 @@
 import os
 import json
-import tempfile
 from config import pconf, plugin_config, conf, write_plugin_config  # noqa: F401  # re-exported via `from .plugin import *`
+from common.atomic_write import write_json_atomic
 from common.log import logger
 
 
-def _write_json_atomically(path, config):
-    """Store ``config`` at ``path`` through a sibling file, then swap it in.
-
-    Writing straight into the file truncates it before the new bytes are there,
-    so anything that fails while serialising — a full disk, an interrupted
-    update — leaves a half-written store behind. ``PluginManager._load_all_config``
-    reads ``plugins/config.json`` with a bare ``json.load`` whose failure is only
-    logged, so one damaged write takes every plugin's configuration with it
-    instead of reporting anything to the user.
-    """
-    fd, temporary = tempfile.mkstemp(prefix=".config.json.", suffix=".tmp",
-                                     dir=os.path.dirname(os.path.abspath(path)))
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            json.dump(config, f, indent=4, ensure_ascii=False)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(temporary, path)
-    except Exception:
-        try:
-            if os.path.exists(temporary):
-                os.remove(temporary)
-        except OSError:
-            pass
-        raise
-
-
 class Plugin:
     def __init__(self):
         self.handlers = {}
@@ -60,11 +33,11 @@ class Plugin:
             # 写入全局配置
             global_config_path = "./plugins/config.json"
             if os.path.exists(global_config_path):
-                _write_json_atomically(global_config_path, plugin_config)
+                write_json_atomic(global_config_path, plugin_config)
             # 写入插件配置
             plugin_config_path = os.path.join(self.path, "config.json")
             if os.path.exists(plugin_config_path):
-                _write_json_atomically(plugin_config_path, config)
+                write_json_atomic(plugin_config_path, config)
 
         except Exception as e:
             logger.warn("save plugin config failed: {}".format(e))

plugins/plugin_manager.py

@@ -6,6 +6,7 @@ import json
 import os
 import sys
 
+from common.atomic_write import write_json_atomic
 from common.log import logger
 from common.singleton import singleton
 from common.sorted_dict import SortedDict
@@ -73,21 +74,7 @@ class PluginManager:
         and load_config has no guard around that. See load_config for why a
         damaged store must not be allowed to abort the load.
         """
-        cfg_path = os.path.join(_plugins_data_dir(), "plugins.json")
-        temporary = f"{cfg_path}.tmp"
-        try:
-            with open(temporary, "w", encoding="utf-8") as f:
-                json.dump(self.pconf, f, indent=4, ensure_ascii=False)
-                f.flush()
-                os.fsync(f.fileno())
-            os.replace(temporary, cfg_path)
-        except Exception:
-            try:
-                if os.path.exists(temporary):
-                    os.remove(temporary)
-            except OSError:
-                pass
-            raise
+        write_json_atomic(os.path.join(_plugins_data_dir(), "plugins.json"), self.pconf)
 
     @staticmethod
     def _read_plugin_store(path: str):

tests/test_atomic_write.py

@@ -0,0 +1,65 @@
+import json
+import os
+import stat
+
+import pytest
+
+from common import atomic_write
+from common.atomic_write import write_json_atomic, write_text_atomic
+
+
+def test_text_write_replaces_the_file(tmp_path):
+    path = tmp_path / "MEMORY.md"
+    path.write_text("old", encoding="utf-8")
+
+    write_text_atomic(path, "新内容\n")
+
+    assert path.read_text(encoding="utf-8") == "新内容\n"
+    assert [p.name for p in tmp_path.iterdir()] == ["MEMORY.md"]
+
+
+def test_json_write_creates_the_file(tmp_path):
+    path = tmp_path / "config.json"
+
+    write_json_atomic(str(path), {"name": "小牛"})
+
+    assert path.read_text(encoding="utf-8") == '{\n    "name": "小牛"\n}'
+
+
+def test_a_failed_write_keeps_the_old_file_and_no_temp(tmp_path):
+    path = tmp_path / "config.json"
+    path.write_text('{"a": 1}', encoding="utf-8")
+
+    with pytest.raises(TypeError):
+        write_json_atomic(path, {"bad": object()})
+
+    assert json.loads(path.read_text(encoding="utf-8")) == {"a": 1}
+    assert [p.name for p in tmp_path.iterdir()] == ["config.json"]
+
+
+def test_temp_file_is_a_dot_prefixed_sibling(tmp_path, monkeypatch):
+    path = tmp_path / "skills_config.json"
+    seen = []
+    real_replace = os.replace
+
+    def spy(src, dst):
+        seen.append(src)
+        real_replace(src, dst)
+
+    monkeypatch.setattr(atomic_write.os, "replace", spy)
+    write_json_atomic(path, {})
+
+    tmp_name = os.path.basename(seen[0])
+    assert os.path.dirname(seen[0]) == str(tmp_path)
+    assert tmp_name.startswith(".skills_config.json.") and tmp_name.endswith(".tmp")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits")
+def test_existing_permissions_are_kept(tmp_path):
+    path = tmp_path / "credentials.json"
+    path.write_text("{}", encoding="utf-8")
+    os.chmod(path, 0o600)
+
+    write_json_atomic(path, {"token": "x"})
+
+    assert stat.S_IMODE(os.stat(path).st_mode) == 0o600

tests/test_model_catalog_atomic_write.py

@@ -69,7 +69,7 @@ def test_a_failed_store_write_leaves_no_temp_file(store_path, monkeypatch):
     with pytest.raises(OSError):
         model_catalog._write_store({"providers": {}})
 
-    assert not (store_path.parent / "models.json.tmp").exists()
+    assert list(store_path.parent.glob("*.tmp")) == []
 
 
 def test_a_store_write_lands_in_one_piece(store_path):
@@ -78,7 +78,7 @@ def test_a_store_write_lands_in_one_piece(store_path):
     model_catalog._write_store(doc)
 
     assert json.loads(store_path.read_text(encoding="utf-8")) == doc
-    assert not (store_path.parent / "models.json.tmp").exists()
+    assert list(store_path.parent.glob("*.tmp")) == []
 
 
 def test_a_failed_legacy_cleanup_keeps_config_json(tmp_path, monkeypatch):

tests/test_plugin_save_config_atomic_write.py

@@ -14,6 +14,7 @@ store and the same shape that ``PluginManager.save_config`` was given a sibling
 file and ``os.replace`` for.
 """
 
+import builtins
 import errno
 import io
 import json
@@ -48,7 +49,7 @@ def _an_installed_plugin(tmp_path, monkeypatch):
 def _a_disk_that_fills_up(monkeypatch):
     """Every write-mode open gets half the document out, then reports ENOSPC.
 
-    ``io.open`` covers the plain ``open(..., "w")`` shape and ``os.fdopen`` the
+    ``open``/``io.open`` cover the plain ``open(..., "w")`` shape and ``os.fdopen`` the
     descriptor one, so the injection follows whichever the writer uses.
     """
 
@@ -73,6 +74,7 @@ def _a_disk_that_fills_up(monkeypatch):
         return _sabotaged(handle) if "w" in mode else handle
 
     monkeypatch.setattr(io, "open", open_for_write)
+    monkeypatch.setattr(builtins, "open", open_for_write)
     monkeypatch.setattr(os, "fdopen", fdopen)
 
 

tests/test_plugin_store_atomic_write.py

@@ -155,4 +155,4 @@ def test_a_failed_save_leaves_no_temp_file_behind(store, monkeypatch):
     with pytest.raises(OSError):
         manager.save_config()
 
-    assert not (data_cfg.parent / "plugins.json.tmp").exists()
+    assert list(data_cfg.parent.glob("*.tmp")) == []

tests/test_summarizer_atomic_memory_write.py

@@ -82,7 +82,7 @@ def _break_writes(monkeypatch, root):
         if "w" in mode:
             path = Path(file)
             name = path.name
-            if name.endswith(".md") or name.endswith(".md.tmp"):
+            if name.endswith(".md") or (name.endswith(".tmp") and ".md." in name):
                 if str(path.resolve()).startswith(root_prefix):
                     return _TruncatingHandle(handle)
         return handle
@@ -98,7 +98,7 @@ def test_an_interrupted_distillation_keeps_the_existing_memory(workspace, monkey
     assert manager.deep_dream(force=True) is False
 
     assert (workspace / "MEMORY.md").read_text(encoding="utf-8") == OLD_MEMORY
-    assert not (workspace / "MEMORY.md.tmp").exists()
+    assert list(workspace.glob("*.tmp")) == []
 
 
 def test_an_interrupted_diary_write_keeps_the_previous_diary(workspace, monkeypatch):

tests/test_team_legacy_retire_atomic_write.py

@@ -10,6 +10,7 @@ deployment raises and never starts. ``team.write``, twenty lines above and
 writing the same kind of file, already saves through a sibling and swaps it in.
 """
 
+import builtins
 import errno
 import io
 import json
@@ -36,7 +37,7 @@ def _an_existing_config(tmp_path):
 def _a_disk_that_fills_up(monkeypatch):
     """Every write-mode open gets half the document out, then reports ENOSPC.
 
-    ``io.open`` covers the ``Path.write_text`` shape and ``os.fdopen`` the
+    ``open``/``io.open`` cover the ``Path.write_text`` shape and ``os.fdopen`` the
     descriptor one, so the injection follows whichever the writer uses.
     """
 
@@ -61,6 +62,7 @@ def _a_disk_that_fills_up(monkeypatch):
         return _sabotaged(handle) if "w" in mode else handle
 
     monkeypatch.setattr(io, "open", open_for_write)
+    monkeypatch.setattr(builtins, "open", open_for_write)
     monkeypatch.setattr(os, "fdopen", fdopen)
 
 

voice/edge/edge_voice.py

@@ -48,12 +48,8 @@ class EdgeVoice(Voice):
         try:
             asyncio.run(self.gen_voice(text, fileName))
         except Exception as e:
-            # Every other provider in this package answers a failure with an
-            # ERROR reply. Letting the raise travel instead reaches
-            # ChatChannel._fail_callback, which only logs
-            # (channel/chat_channel.py:453), and because this runs while the
-            # text reply is being converted to voice the user loses the text
-            # answer as well.
+            # Raising here would drop the text answer too, since this runs while
+            # the text reply is being converted to voice.
             logger.error("[EdgeTTS] textToVoice failed: {}".format(e))
             return Reply(ReplyType.ERROR, "抱歉，语音合成失败")
 

voice/elevent/elevent_voice.py

@@ -32,12 +32,8 @@ class ElevenLabsVoice(Voice):
             )
             save(audio, fileName)
         except Exception as e:
-            # Every other provider in this package answers a failure with an
-            # ERROR reply. Letting the raise travel instead reaches
-            # ChatChannel._fail_callback, which only logs
-            # (channel/chat_channel.py:453), and because this runs while the
-            # text reply is being converted to voice the user loses the text
-            # answer as well.
+            # Raising here would drop the text answer too, since this runs while
+            # the text reply is being converted to voice.
             logger.error("[ElevenLabs] textToVoice failed: {}".format(e))
             return Reply(ReplyType.ERROR, "抱歉，语音合成失败")