fix(channels): publish Discord and WeChat JSON state atomically (#7595)

* fix(channels): publish Discord and WeChat JSON state atomically

* fix(channels): adopt the shared atomic publisher for both state writers

* test(channels): pin the cleanup-failure direction too

* test(channels): keep a platform-independent guard on the publish call shape

* style(tests): apply ruff format to the channel encoding test

backend/app/channels/discord.py

@@ -28,6 +28,7 @@ from app.channels.message_bus import (
     OutboundMessage,
     ResolvedAttachment,
 )
+from deerflow.utils.atomic_write import publish_bytes_atomically
 from deerflow.utils.file_io import await_drained
 
 logger = logging.getLogger(__name__)
@@ -275,7 +276,10 @@ class DiscordChannel(Channel):
             try:
                 snapshot = dict(self._active_threads)
                 self._thread_store_path.parent.mkdir(parents=True, exist_ok=True)
-                self._thread_store_path.write_text(json.dumps(snapshot, indent=2), encoding="utf-8", newline="")
+                # Publish through the shared helper: it fsyncs the staged bytes
+                # before the rename (so a machine crash cannot leave an empty
+                # mapping behind) and re-applies an existing file's mode.
+                publish_bytes_atomically(self._thread_store_path, json.dumps(snapshot, indent=2).encode("utf-8"))
             except Exception:
                 logger.exception("[Discord] failed to persist thread mappings")
 

backend/app/channels/wechat.py

@@ -32,6 +32,7 @@ from app.channels.commands import is_known_channel_command
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.credential_staging import cleanup_stale_credential_staging, credential_staging_prefix
 from app.channels.message_bus import INBOUND_FILE_CLEANUP_KEY, InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment, cleanup_inbound_files
+from deerflow.utils.atomic_write import publish_bytes_atomically
 from deerflow.utils.file_io import await_drained
 
 logger = logging.getLogger(__name__)
@@ -1707,7 +1708,10 @@ class WechatChannel(Channel):
             return
         try:
             self._cursor_path.parent.mkdir(parents=True, exist_ok=True)
-            self._cursor_path.write_text(json.dumps({"get_updates_buf": self._get_updates_buf}, ensure_ascii=False, indent=2), encoding="utf-8", newline="")
+            # Publish through the shared helper: it fsyncs the staged bytes
+            # before the rename (so a machine crash cannot leave a truncated
+            # cursor behind) and re-applies an existing file's mode.
+            publish_bytes_atomically(self._cursor_path, json.dumps({"get_updates_buf": self._get_updates_buf}, ensure_ascii=False, indent=2).encode("utf-8"))
         except OSError:
             logger.warning("[WeChat] failed to persist cursor state to %s", self._cursor_path)
 

backend/tests/test_channel_state_atomic_publish.py

@@ -0,0 +1,148 @@
+"""Channel JSON state must be published whole, never truncated in place.
+
+``DiscordChannel._persist_thread_mappings`` and ``WechatChannel._save_state``
+write state a crash could otherwise truncate: the Discord mapping would fail to
+load and the WeChat cursor buffer would be lost, re-reading from the start. Both
+now stage a sibling and replace it, and a stage that cannot be discarded is
+reported instead of replacing the outcome.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from pathlib import Path
+
+from app.channels.discord import DiscordChannel
+from app.channels.wechat import WechatChannel
+
+
+def _discord(tmp_path):
+    channel = DiscordChannel.__new__(DiscordChannel)
+    channel._thread_store_lock = threading.Lock()
+    channel._active_threads = {"a": {"thread_id": "t1"}}
+    channel._thread_store_path = tmp_path / "state" / "discord-threads.json"
+    return channel
+
+
+def _wechat(tmp_path):
+    channel = WechatChannel.__new__(WechatChannel)
+    channel._cursor_path = tmp_path / "state" / "wechat-cursor.json"
+    channel._get_updates_buf = "cursor-1"
+    return channel
+
+
+def _fail_staged_write(monkeypatch):
+    """Fail the staged write the way a full disk would, before any rename.
+
+    With the shared helper this is the earliest injectable point: the bytes are
+    written to a sibling, fsync fails, and the rename never happens. The point of
+    the test is that the live file is untouched by that failure.
+    """
+
+    def failing_fsync(_fd):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("deerflow.utils.atomic_write.os.fsync", failing_fsync)
+
+
+def _no_sidecar(target: Path) -> bool:
+    prefix = f".{target.name}."
+    return [p.name for p in target.parent.iterdir() if p.name.startswith(prefix)] == []
+
+
+def test_discord_state_is_replaced_whole(tmp_path):
+    channel = _discord(tmp_path)
+    target = tmp_path / "state" / "discord-threads.json"
+
+    channel._persist_thread_mappings()
+
+    assert json.loads(target.read_text(encoding="utf-8")) == {"a": {"thread_id": "t1"}}
+    assert _no_sidecar(target)
+
+
+def test_discord_failed_staged_write_keeps_the_previous_mapping(tmp_path, monkeypatch):
+    channel = _discord(tmp_path)
+    target = tmp_path / "state" / "discord-threads.json"
+    target.parent.mkdir(parents=True)
+    target.write_text('{"kept": true}', encoding="utf-8")
+    _fail_staged_write(monkeypatch)
+
+    channel._persist_thread_mappings()
+
+    assert target.read_text(encoding="utf-8") == '{"kept": true}'
+    assert json.loads(target.read_text(encoding="utf-8")) == {"kept": True}
+    assert _no_sidecar(target)
+
+
+def test_wechat_state_is_replaced_whole(tmp_path):
+    channel = _wechat(tmp_path)
+    target = tmp_path / "state" / "wechat-cursor.json"
+
+    channel._save_state()
+
+    assert json.loads(target.read_text(encoding="utf-8")) == {"get_updates_buf": "cursor-1"}
+    assert _no_sidecar(target)
+
+
+def test_wechat_failed_staged_write_keeps_the_previous_cursor(tmp_path, monkeypatch):
+    channel = _wechat(tmp_path)
+    target = tmp_path / "state" / "wechat-cursor.json"
+    target.parent.mkdir(parents=True)
+    target.write_text('{"get_updates_buf": "kept"}', encoding="utf-8")
+    _fail_staged_write(monkeypatch)
+
+    channel._save_state()
+
+    assert json.loads(target.read_text(encoding="utf-8")) == {"get_updates_buf": "kept"}
+    assert _no_sidecar(target)
+
+
+def _fail_staged_write_and_cleanup(monkeypatch):
+    """Fail the staged write, and fail discarding the staged sibling too."""
+
+    def failing_fsync(_fd):
+        raise OSError("disk full")
+
+    real_unlink = Path.unlink
+
+    def failing_unlink(self, *args, **kwargs):
+        if self.name.startswith(".") and self.name.endswith(".tmp"):
+            raise PermissionError("staging is still open")
+        return real_unlink(self, *args, **kwargs)
+
+    monkeypatch.setattr("deerflow.utils.atomic_write.os.fsync", failing_fsync)
+    monkeypatch.setattr(Path, "unlink", failing_unlink)
+
+
+def test_discord_failed_cleanup_still_reports_and_keeps_the_outcome(tmp_path, monkeypatch, caplog):
+    """A discard that fails must not escape, and both failures stay visible."""
+    channel = _discord(tmp_path)
+    target = tmp_path / "state" / "discord-threads.json"
+    target.parent.mkdir(parents=True)
+    target.write_text('{"kept": true}', encoding="utf-8")
+    _fail_staged_write_and_cleanup(monkeypatch)
+    caplog.clear()
+
+    with caplog.at_level("WARNING"):
+        channel._persist_thread_mappings()
+
+    assert target.read_text(encoding="utf-8") == '{"kept": true}'
+    assert "failed to persist thread mappings" in caplog.text
+    assert "Failed to remove temporary output file" in caplog.text
+
+
+def test_wechat_failed_cleanup_still_reports_and_keeps_the_outcome(tmp_path, monkeypatch, caplog):
+    channel = _wechat(tmp_path)
+    target = tmp_path / "state" / "wechat-cursor.json"
+    target.parent.mkdir(parents=True)
+    target.write_text('{"get_updates_buf": "kept"}', encoding="utf-8")
+    _fail_staged_write_and_cleanup(monkeypatch)
+    caplog.clear()
+
+    with caplog.at_level("WARNING"):
+        channel._save_state()
+
+    assert target.read_text(encoding="utf-8") == '{"get_updates_buf": "kept"}'
+    assert "failed to persist cursor state" in caplog.text
+    assert "Failed to remove temporary output file" in caplog.text

backend/tests/test_channel_state_encoding.py

@@ -22,10 +22,8 @@ def _state_io_calls(filename: str, class_name: str, method_name: str, call_name:
     ("filename", "class_name", "method_name", "call_name"),
     [
         ("store.py", "JsonChannelStore", "_save", "NamedTemporaryFile"),
-        ("discord.py", "DiscordChannel", "_persist_thread_mappings", "write_text"),
         ("runtime_config_store.py", "ChannelRuntimeConfigStore", "_save", "NamedTemporaryFile"),
         ("buzz_seen_events.py", "BuzzSeenEventStore", "_write_snapshot", "NamedTemporaryFile"),
-        ("wechat.py", "WechatChannel", "_save_state", "write_text"),
         ("wechat.py", "WechatChannel", "_save_auth_state", "NamedTemporaryFile"),
     ],
 )
@@ -57,3 +55,67 @@ def test_discord_state_read_is_explicitly_utf8() -> None:
         keywords = {keyword.arg: keyword.value for keyword in call.keywords}
         encoding = keywords.get("encoding")
         assert isinstance(encoding, ast.Constant) and encoding.value == "utf-8"
+
+
+def test_discord_thread_mappings_publish_utf8_lf_bytes(tmp_path: Path) -> None:
+    import threading
+
+    from app.channels.discord import DiscordChannel
+
+    channel = DiscordChannel.__new__(DiscordChannel)
+    channel._thread_store_lock = threading.Lock()
+    channel._active_threads = {"微信:资源": {"thread_id": "thread-讨论"}}
+    channel._thread_store_path = tmp_path / "discord-threads.json"
+
+    channel._persist_thread_mappings()
+
+    text = channel._thread_store_path.read_bytes().decode("utf-8")
+    assert "\r" not in text
+    assert json.loads(text)["微信:资源"]["thread_id"] == "thread-讨论"
+
+
+def test_wechat_cursor_state_publishes_utf8_lf_bytes(tmp_path: Path) -> None:
+    from app.channels.wechat import WechatChannel
+
+    channel = WechatChannel.__new__(WechatChannel)
+    channel._cursor_path = tmp_path / "wechat-cursor.json"
+    channel._get_updates_buf = "游标-1"
+
+    channel._save_state()
+
+    text = channel._cursor_path.read_bytes().decode("utf-8")
+    assert "\r" not in text
+    assert json.loads(text) == {"get_updates_buf": "游标-1"}
+
+
+def _method_calls(filename: str, class_name: str, method_name: str) -> list[ast.Call]:
+    module = ast.parse((CHANNELS / filename).read_text(encoding="utf-8"))
+    owner = next(node for node in module.body if isinstance(node, ast.ClassDef) and node.name == class_name)
+    method = next(node for node in owner.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == method_name)
+    return [node for node in ast.walk(method) if isinstance(node, ast.Call)]
+
+
+@pytest.mark.parametrize(
+    ("filename", "class_name", "method_name"),
+    [
+        ("discord.py", "DiscordChannel", "_persist_thread_mappings"),
+        ("wechat.py", "WechatChannel", "_save_state"),
+    ],
+)
+def test_json_channel_state_writers_publish_through_the_helper(filename: str, class_name: str, method_name: str) -> None:
+    """Platform-independent: the call shape, not just the bytes it happens to emit.
+
+    CI runs with ``PYTHONUTF8=1``, so a regression to ``write_text(...)`` without
+    ``encoding=`` would still produce valid UTF-8 there; only a source-level check
+    catches it on every platform.
+    """
+    calls = _method_calls(filename, class_name, method_name)
+    names = [call.func.attr if isinstance(call.func, ast.Attribute) else call.func.id for call in calls if isinstance(call.func, (ast.Attribute, ast.Name))]
+    assert "publish_bytes_atomically" in names, f"{filename}:{class_name}.{method_name} must publish through the helper"
+
+    for call in calls:
+        if not isinstance(call.func, ast.Attribute) or call.func.attr not in {"write_text", "write_bytes"}:
+            continue
+        keywords = {keyword.arg: keyword.value for keyword in call.keywords}
+        encoding = keywords.get("encoding")
+        assert isinstance(encoding, ast.Constant) and encoding.value == "utf-8", f"{filename}:{call.lineno} writes text without an explicit utf-8 encoding"