fix(channels): keep staging cleanup from replacing the save error (#7106)

* fix(channels): keep staging cleanup from replacing the save error

The channel stores publish state through a staged temporary file and, if the
rename fails, close and unlink that file inside the handler before re-raising.
Neither step was guarded, so a failing close skipped the unlink and a failing
unlink replaced the error that actually caused the save to fail -- leaving the
staged copy, which for the runtime-config and WeChat stores holds an 0600 IM
token, behind with nothing reporting it.

Make both steps independently best-effort, name the leaked file in a WARNING,
and suppress the diagnostic itself so a broken log handler cannot become the
reported failure. This is the same contract PR #6871 establishes for browser
captures, applied to the three JSON publishers in app/channels.

* fix(channels): address staging cleanup review follow-ups

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -1017,6 +1017,15 @@ DeerFlow supports receiving tasks from messaging apps. Channels auto-start when
 
 Cancelling a channel restart discards its pending configuration reload, preserving newer runtime settings applied afterward.
 
+Conversation-binding, runtime channel-credential and WeChat auth-state saves preserve the original
+persistence failure when staged-file cleanup also fails. Remaining cleanup still runs, and warnings
+identify copies that could not be removed. Buzz keeps failed snapshots available for retry, even if
+cleanup or logging fails.
+
+On startup, runtime credential and WeChat auth stores remove their own `.<state-file>.deerflow-*.tmp`
+staging files older than 24 hours. Fresh files, symlinks and unrelated files are preserved. Older
+anonymous `tmp*.tmp` files require manual inspection because their writer cannot be identified safely.
+
 DeerFlow can also expose user-owned IM channel connections in the workspace UI. When `channel_connections` is enabled, logged-in users can bind Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, WeCom, QQ, or Buzz from the sidebar / Settings > Channels. It reuses the existing outbound `channels.*` transports, so no public IP or provider callback URL is required. Incoming IM messages then run under the connected DeerFlow user account. See [IM Channel Connections](backend/docs/IM_CHANNEL_CONNECTIONS.md) for setup and security notes. Buzz backlog paging stops with a warning if timestamp-based queries cannot advance, and channel removal closes both live and backlog subscriptions.
 
 | Channel | Transport | Difficulty |

backend/AGENTS.md

@@ -174,6 +174,8 @@ The backend is split into two layers with a strict dependency direction:
 - **Harness** (`packages/harness/deerflow/`): Publishable agent framework package (`deerflow-harness`). Import prefix: `deerflow.*`. Contains agent orchestration, tools, sandbox, models, MCP, skills, config — everything needed to build and run agents.
 - **App** (`app/`): Unpublished application code. Import prefix: `app.*`. Contains the FastAPI Gateway API and IM channel integrations (Feishu, Slack, Telegram, DingTalk). Feishu option clarifications can render as interactive button cards behind `channels.feishu.card_interaction_enabled` (default off); see `app/channels/AGENTS.md`.
 
+Channel credential staging (`app/channels/credential_staging.py`) is swept on worker-thread state loads, including SQL legacy import. Only owned regular files older than 24h are removed; leave anonymous temporary files alone and keep WeChat constructors IO-free.
+
 **Dependency rule**: App imports deerflow, but deerflow never imports app. This boundary is enforced by `tests/test_harness_boundary.py` which runs in CI.
 
 **Import conventions**:

backend/app/channels/buzz_seen_events.py

@@ -56,6 +56,7 @@ import logging
 import tempfile
 import threading
 from collections import OrderedDict, deque
+from contextlib import suppress
 from pathlib import Path
 
 logger = logging.getLogger(__name__)
@@ -136,7 +137,7 @@ class BuzzSeenEventStore:
     def _write_snapshot(self, payload: dict[str, list[str]]) -> bool:
         if self._path is None:
             return True
-        tmp_name: str | None = None
+        fh = None
         try:
             path = self._path
             path.parent.mkdir(parents=True, exist_ok=True)
@@ -144,17 +145,29 @@ class BuzzSeenEventStore:
             # crash mid-write must never truncate the store (a truncated store
             # would fail open into replay on the next start, which is
             # recoverable — but there is no reason to accept even that).
-            with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, suffix=".tmp", delete=False, encoding="utf-8") as fh:
-                tmp_name = fh.name
-                json.dump(payload, fh)
-            Path(tmp_name).replace(path)
+            fh = tempfile.NamedTemporaryFile(mode="w", dir=path.parent, suffix=".tmp", delete=False, encoding="utf-8")
+            json.dump(payload, fh)
+            fh.close()
+            Path(fh.name).replace(path)
             return True
-        except Exception:
-            # Mirror ChannelStore._save: never leave the temp file behind, or a
-            # persistently failing write accumulates one *.tmp per attempt.
-            if tmp_name is not None:
-                Path(tmp_name).unlink(missing_ok=True)
-            logger.warning("[buzz] failed to persist seen-event store (will retry on next flush)", exc_info=True)
+        except BaseException as exc:
+            # Each cleanup step and its diagnostic is independent. Keep the
+            # persistence failure attributable and the dirty snapshot retryable.
+            if fh is not None:
+                try:
+                    fh.close()
+                except OSError:
+                    with suppress(Exception):
+                        logger.warning("Failed to close temporary Buzz seen-event store file %s", fh.name, exc_info=True)
+                try:
+                    Path(fh.name).unlink(missing_ok=True)
+                except OSError:
+                    with suppress(Exception):
+                        logger.warning("Failed to remove temporary Buzz seen-event store file %s", fh.name, exc_info=True)
+            if not isinstance(exc, Exception):
+                raise
+            with suppress(Exception):
+                logger.warning("[buzz] failed to persist seen-event store (will retry on next flush)", exc_info=True)
             return False
 
     def _save(self) -> None:

backend/app/channels/credential_staging.py

@@ -0,0 +1,56 @@
+"""Best-effort cleanup for identifiable crash-orphaned credential copies.
+
+All calls touch disk and belong on the existing worker-thread state-load paths.
+Only the reserved per-target namespace is owned here: anonymous temporary files
+from older releases cannot safely be attributed to a credential writer.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import stat
+import time
+from contextlib import suppress
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+_STALE_AFTER_SECONDS = 24 * 60 * 60
+
+
+def credential_staging_prefix(target: Path) -> str:
+    """Reserve a namespace for this target without changing tempfile's 0600 mode."""
+    return f".{target.name}.deerflow-"
+
+
+def cleanup_stale_credential_staging(target: Path) -> None:
+    """Remove only this store's regular staging files older than a day.
+
+    Scan siblings only, never attachments or symlink targets. The age guard
+    leaves fresh files to concurrent writers. Failure to inspect or remove a
+    stale copy, including failure to report it, must not stop credential loading.
+    Missing files are expected when a peer completes its atomic publication.
+    """
+    prefix = credential_staging_prefix(target)
+    cutoff = time.time() - _STALE_AFTER_SECONDS
+    try:
+        with os.scandir(target.parent) as entries:
+            for entry in entries:
+                if not entry.name.startswith(prefix) or not entry.name.endswith(".tmp"):
+                    continue
+                try:
+                    metadata = entry.stat(follow_symlinks=False)
+                    if not stat.S_ISREG(metadata.st_mode) or metadata.st_mtime >= cutoff:
+                        continue
+                    Path(entry.path).unlink(missing_ok=True)
+                except FileNotFoundError:
+                    continue
+                except OSError:
+                    with suppress(Exception):
+                        logger.warning("Failed to clean stale channel credential staging file %s", entry.path, exc_info=True)
+    except FileNotFoundError:
+        return
+    except OSError:
+        with suppress(Exception):
+            logger.warning("Failed to scan channel credential staging directory %s", target.parent, exc_info=True)

backend/app/channels/runtime_config_store.py

@@ -31,10 +31,12 @@ import tempfile
 import threading
 import time
 from collections.abc import Iterator
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Protocol
 
+from app.channels.credential_staging import cleanup_stale_credential_staging, credential_staging_prefix
+
 if TYPE_CHECKING:
     from deerflow.config.credentials_key import CredentialsCipher
     from deerflow.persistence.channel_runtime_configs import SqlChannelRuntimeConfigRepository, StoredChannelRuntimeConfig
@@ -120,6 +122,7 @@ class ChannelRuntimeConfigStore:
     def __init__(self, path: str | Path | None = None) -> None:
         self._path = Path(path) if path is not None else default_runtime_config_path()
         self._path.parent.mkdir(parents=True, exist_ok=True)
+        cleanup_stale_credential_staging(self._path)
         self._data: dict[str, dict[str, Any]] = self._load()
         self._lock = threading.Lock()
 
@@ -160,6 +163,7 @@ class ChannelRuntimeConfigStore:
             mode="w",
             encoding="utf-8",
             dir=self._path.parent,
+            prefix=credential_staging_prefix(self._path),
             suffix=".tmp",
             delete=False,
         )
@@ -176,8 +180,19 @@ class ChannelRuntimeConfigStore:
             except OSError:
                 logger.debug("Unable to chmod channel runtime config store at %s", self._path, exc_info=True)
         except BaseException:
-            fd.close()
-            Path(fd.name).unlink(missing_ok=True)
+            # Cleanup is best-effort: a failing close or unlink must not take the
+            # original save error with it, and the unlink is still attempted after
+            # a failed close.
+            try:
+                fd.close()
+            except OSError:
+                with suppress(Exception):
+                    logger.warning("Failed to close temporary channel runtime config file %s", fd.name, exc_info=True)
+            try:
+                Path(fd.name).unlink(missing_ok=True)
+            except OSError:
+                with suppress(Exception):
+                    logger.warning("Failed to remove temporary channel runtime config file %s", fd.name, exc_info=True)
             raise
 
     def load_all(self) -> dict[str, dict[str, Any]]:
@@ -346,6 +361,9 @@ class SqlChannelRuntimeConfigStore:
 
 def _read_legacy_file(path: Path) -> dict[str, Any] | None:
     """Parse ``runtime-config.json``; ``None`` when it is missing, unreadable or not an object (kept in place)."""
+    # Also runs after migration removed the legacy file: a crash-orphaned
+    # credential copy must not survive merely because the store now uses SQL.
+    cleanup_stale_credential_staging(path)
     try:
         raw = path.read_text(encoding="utf-8")
     except FileNotFoundError:

backend/app/channels/store.py

@@ -42,6 +42,7 @@ import logging
 import tempfile
 import threading
 import time
+from contextlib import suppress
 from pathlib import Path
 from typing import Any, Protocol
 
@@ -140,8 +141,19 @@ class JsonChannelStore:
             fd.close()
             Path(fd.name).replace(self._path)
         except BaseException:
-            fd.close()
-            Path(fd.name).unlink(missing_ok=True)
+            # Cleanup is best-effort: a failing close or unlink must not take the
+            # original save error with it, and the unlink is still attempted after
+            # a failed close.
+            try:
+                fd.close()
+            except OSError:
+                with suppress(Exception):
+                    logger.warning("Failed to close temporary channel store file %s", fd.name, exc_info=True)
+            try:
+                Path(fd.name).unlink(missing_ok=True)
+            except OSError:
+                with suppress(Exception):
+                    logger.warning("Failed to remove temporary channel store file %s", fd.name, exc_info=True)
             raise
 
     # -- sync bodies -------------------------------------------------------

backend/app/channels/wechat.py

@@ -15,6 +15,7 @@ import shutil
 import tempfile
 import time
 from collections.abc import Mapping
+from contextlib import suppress
 from enum import IntEnum
 from functools import partial
 from pathlib import Path
@@ -29,6 +30,7 @@ from app.channels.allowed_users import parse_allowed_users
 from app.channels.base import Channel, ChannelUnavailable
 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.file_io import await_drained
 
@@ -1710,7 +1712,10 @@ class WechatChannel(Channel):
             logger.warning("[WeChat] failed to persist cursor state to %s", self._cursor_path)
 
     def _load_auth_state(self) -> None:
-        if not self._auth_path or not self._auth_path.exists():
+        if not self._auth_path:
+            return
+        cleanup_stale_credential_staging(self._auth_path)
+        if not self._auth_path.exists():
             return
         try:
             data = json.loads(self._auth_path.read_text(encoding="utf-8"))
@@ -1769,17 +1774,28 @@ class WechatChannel(Channel):
                 # iLink bot_token is never briefly readable at umask defaults
                 # (mirrors ChannelRuntimeConfigStore._save). NamedTemporaryFile
                 # uses mkstemp, which creates the file at 0o600 from the start.
-                fd = tempfile.NamedTemporaryFile(mode="w", dir=self._auth_path.parent, suffix=".tmp", delete=False, encoding="utf-8")
+                fd = tempfile.NamedTemporaryFile(mode="w", dir=self._auth_path.parent, prefix=credential_staging_prefix(self._auth_path), suffix=".tmp", delete=False, encoding="utf-8")
                 try:
                     json.dump(data, fd, ensure_ascii=False, indent=2)
                     fd.close()
                     Path(fd.name).replace(self._auth_path)
                 except BaseException:
-                    fd.close()
-                    Path(fd.name).unlink(missing_ok=True)
+                    # Cleanup is best-effort: a failing close or unlink must not take
+                    # the original persist error with it, and the unlink is still
+                    # attempted after a failed close.
+                    try:
+                        fd.close()
+                    except OSError:
+                        with suppress(Exception):
+                            logger.warning("Failed to close temporary WeChat auth state file %s", fd.name, exc_info=True)
+                    try:
+                        Path(fd.name).unlink(missing_ok=True)
+                    except OSError:
+                        with suppress(Exception):
+                            logger.warning("Failed to remove temporary WeChat auth state file %s", fd.name, exc_info=True)
                     raise
             except OSError:
-                logger.warning("[WeChat] failed to persist auth state to %s", self._auth_path)
+                logger.warning("[WeChat] failed to persist auth state to %s", self._auth_path, exc_info=True)
             else:
                 # Hardening only; the destination already inherits 0o600 from the
                 # temp file. A chmod failure on filesystems without POSIX perms

backend/tests/AGENTS.md

@@ -173,6 +173,25 @@ tree left behind. Do not replace these behavioral checks with assertions about
 source formatting or duplicate the runtime publication control in
 `test_lark_cli_atomic_publish.py`.
 
+## Channel staging cleanup
+
+`test_channel_staging_cleanup.py` blocks the rename destination (a non-empty
+directory, because an empty one would just swallow the staged file) and fails
+`Path.unlink`, the descriptor `close()`, and the log handler in turn. Assert the
+save error is what propagates, that the unlink is still attempted after a failed
+close, and that a staged copy the cleanup could not remove is named in a warning.
+Keep these as injected-failure behavior checks: an assertion that the handler
+source merely contains `suppress` would pass while the close error still skipped
+the unlink.
+
+`test_channel_credential_staging.py` leaves copies through the real credential writers, then ages
+one for restart cleanup. Preserve current credentials, recent peer staging, anonymous files,
+symlinks, directories and nested attachments. Exercise SQL startup after the legacy file is gone,
+plus failed unlink/listing and a real delayed FileHandler whose parent is missing. WeChat startup
+cleanup must also pass the strict blocking-I/O gate. Buzz failure tests keep the original write
+exception attributable through close/unlink/diagnostic errors, retain pending events for a later
+flush, and propagate interruptions after cleanup.
+
 ## Lark CLI blocking-I/O fixtures
 
 `blocking_io/test_integrations_router.py` uses a real local CLI stub: a `.cmd`

backend/tests/blocking_io/test_wechat_channel_state.py

@@ -29,6 +29,8 @@ from __future__ import annotations
 
 import asyncio
 import json
+import os
+import time
 from pathlib import Path
 
 import pytest
@@ -114,3 +116,23 @@ async def test_wechat_auth_state_load_does_not_block_event_loop(tmp_path: Path)
 
     assert result is True
     assert channel._bot_token == "loaded-token"
+
+
+async def test_wechat_startup_sweeps_stale_auth_staging_off_loop(tmp_path: Path) -> None:
+    orphan = tmp_path / ".wechat-auth.json.deerflow-orphan.tmp"
+
+    def seed():
+        orphan.write_text("stale-credential", encoding="utf-8")
+        old = time.time() - 2 * 24 * 60 * 60
+        os.utime(orphan, (old, old))
+        (tmp_path / "wechat-getupdates.json").write_text(json.dumps({"get_updates_buf": "saved-cursor"}), encoding="utf-8")
+
+    await asyncio.to_thread(seed)
+    # No token and no QR login: start loads state, then exits before any network
+    # setup. Construction itself must remain free of filesystem work.
+    channel = WechatChannel(bus=MessageBus(), config={"state_dir": str(tmp_path), "qrcode_login_enabled": False})
+
+    await channel.start()
+
+    assert not await asyncio.to_thread(orphan.exists)
+    assert channel._get_updates_buf == "saved-cursor"

backend/tests/test_channel_credential_staging.py

@@ -0,0 +1,187 @@
+"""Restart cleanup only owns stale, explicitly named credential staging files."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from app.channels.message_bus import MessageBus
+from app.channels.runtime_config_store import ChannelRuntimeConfigStore, SqlChannelRuntimeConfigStore
+from app.channels.wechat import WechatChannel
+
+
+def _age(path: Path) -> None:
+    old = time.time() - 2 * 24 * 60 * 60
+    os.utime(path, (old, old), follow_symlinks=False)
+
+
+def _load(kind: str, target: Path):
+    if kind == "runtime":
+        return ChannelRuntimeConfigStore(target).get_provider_config("slack")
+    channel = WechatChannel(bus=MessageBus(), config={"state_dir": str(target.parent)})
+    channel._load_state()
+    return channel._auth_state
+
+
+@pytest.mark.parametrize("kind", ["runtime", "wechat"])
+@pytest.mark.parametrize("destination_exists", [True, False])
+def test_restart_sweeps_crash_orphans_without_touching_other_state(tmp_path, monkeypatch, kind, destination_exists):
+    target = tmp_path / ("runtime-config.json" if kind == "runtime" else "wechat-auth.json")
+    if kind == "runtime":
+        store = ChannelRuntimeConfigStore(target)
+
+        def save(token):
+            store._save({"slack": {"bot_token": token}})
+    else:
+        channel = WechatChannel(bus=MessageBus(), config={"state_dir": str(tmp_path)})
+
+        def save(token):
+            channel._save_auth_state(status="confirmed", bot_token=token)
+
+    save("current-token")
+    current_bytes = target.read_bytes()
+    real_replace = Path.replace
+    real_unlink = Path.unlink
+
+    def fail_publish(path, destination):
+        if destination == target:
+            raise OSError("publication interrupted")
+        return real_replace(path, destination)
+
+    def fail_cleanup(path, *args, **kwargs):
+        if path.suffix == ".tmp":
+            raise PermissionError("cleanup interrupted")
+        return real_unlink(path, *args, **kwargs)
+
+    # Leave the same files a hard crash would leave, using the actual writers.
+    with monkeypatch.context() as patch:
+        patch.setattr(Path, "replace", fail_publish)
+        patch.setattr(Path, "unlink", fail_cleanup)
+        for token in ("orphan-token", "inflight-token"):
+            if kind == "runtime":
+                with pytest.raises(OSError, match="publication interrupted"):
+                    save(token)
+            else:
+                save(token)
+    staged = list(tmp_path.glob("*.tmp"))
+    assert len(staged) == 2
+    orphan, fresh = staged
+    assert all(path.name.startswith(f".{target.name}.deerflow-") for path in staged)
+    if os.name != "nt":
+        assert all(path.stat().st_mode & 0o777 == 0o600 for path in staged)
+    _age(orphan)
+    recent = time.time() - 23 * 60 * 60
+    os.utime(fresh, (recent, recent))
+
+    # Even old generic tempfile names cannot be attributed to this store.
+    retained = [tmp_path / name for name in ("backup.tmp", "tmpabcdefgh.tmp", ".other.json.deerflow-old.tmp")]
+    for path in retained:
+        path.write_text("unrelated", encoding="utf-8")
+        _age(path)
+    directory = tmp_path / f".{target.name}.deerflow-directory.tmp"
+    directory.mkdir()
+    _age(directory)
+    downloads = tmp_path / "downloads"
+    downloads.mkdir()
+    nested = downloads / orphan.name
+    nested.write_text("attachment", encoding="utf-8")
+    _age(nested)
+    if os.name != "nt":
+        link = tmp_path / f".{target.name}.deerflow-symlink.tmp"
+        link.symlink_to(retained[0])
+        _age(link)
+        retained.append(link)
+    if not destination_exists:
+        target.unlink()
+
+    loaded = _load(kind, target)
+
+    assert not orphan.exists()
+    assert fresh.exists(), "a peer may still be writing a fresh staging file"
+    assert all(path.exists() for path in retained)
+    assert directory.is_dir()
+    assert nested.read_text(encoding="utf-8") == "attachment"
+    if destination_exists:
+        assert target.read_bytes() == current_bytes
+        config = loaded if kind == "wechat" else loaded or {}
+        assert config["bot_token"] == "current-token"
+    else:
+        assert not target.exists()
+        assert not loaded
+
+
+@pytest.mark.parametrize("kind", ["runtime", "wechat"])
+@pytest.mark.parametrize("broken_handler", [False, True])
+def test_orphan_cleanup_failure_does_not_skip_other_files_or_loading(tmp_path, monkeypatch, caplog, kind, broken_handler):
+    target = tmp_path / ("runtime-config.json" if kind == "runtime" else "wechat-auth.json")
+    config = {"bot_token": "current-token"}
+    target.write_text(json.dumps({"slack": config} if kind == "runtime" else config), encoding="utf-8")
+    failed = tmp_path / f".{target.name}.deerflow-failed.tmp"
+    other = tmp_path / f".{target.name}.deerflow-other.tmp"
+    for path in (failed, other):
+        path.write_text("stale-credential", encoding="utf-8")
+        _age(path)
+    real_unlink = Path.unlink
+    failure = PermissionError("stale copy is locked")
+
+    def fail_one_unlink(path, *args, **kwargs):
+        if path == failed:
+            raise failure
+        return real_unlink(path, *args, **kwargs)
+
+    logger = logging.getLogger("app.channels.credential_staging")
+    handler = logging.FileHandler(tmp_path / "missing-log-directory" / "output.log", delay=True) if broken_handler else None
+    if handler is not None:
+        logger.addHandler(handler)
+    try:
+        with monkeypatch.context() as patch, caplog.at_level(logging.WARNING):
+            patch.setattr(Path, "unlink", fail_one_unlink)
+            assert _load(kind, target)["bot_token"] == "current-token"
+    finally:
+        if handler is not None:
+            logger.removeHandler(handler)
+            handler.close()
+
+    assert failed.exists()
+    assert not other.exists()
+    if not broken_handler:
+        records = [record for record in caplog.records if str(failed) in record.getMessage()]
+        assert len(records) == 1
+        assert records[0].exc_info[1] is failure
+
+
+@pytest.mark.parametrize("kind", ["runtime", "wechat"])
+def test_unreadable_staging_directory_does_not_prevent_loading_current_credentials(tmp_path, monkeypatch, kind):
+    target = tmp_path / ("runtime-config.json" if kind == "runtime" else "wechat-auth.json")
+    config = {"bot_token": "current-token"}
+    target.write_text(json.dumps({"slack": config} if kind == "runtime" else config), encoding="utf-8")
+    real_scandir = os.scandir
+
+    def unreadable(path):
+        if Path(path) == tmp_path:
+            raise PermissionError("directory listing denied")
+        return real_scandir(path)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(os, "scandir", unreadable)
+        assert _load(kind, target)["bot_token"] == "current-token"
+
+
+def test_database_startup_sweeps_staging_even_after_the_legacy_file_was_removed(tmp_path):
+    target = tmp_path / "runtime-config.json"
+    orphan = tmp_path / f".{target.name}.deerflow-orphan.tmp"
+    orphan.write_text("plaintext-credential", encoding="utf-8")
+    _age(orphan)
+    repo = Mock()
+    store = SqlChannelRuntimeConfigStore(repo, cipher=None, legacy_path=target)
+
+    assert store.import_legacy_json() == 0
+
+    assert not orphan.exists()
+    repo.insert_missing.assert_not_called()

backend/tests/test_channel_staging_cleanup.py

@@ -0,0 +1,271 @@
+"""Staged-publication cleanup in the channel stores must keep the original error.
+
+``JsonChannelStore``, ``ChannelRuntimeConfigStore`` and the WeChat auth-state file all
+publish by writing a staged temporary file and renaming it over the destination. Their
+cleanup path used to run ``close()`` and ``unlink()`` unguarded inside
+``except BaseException: raise``, so a failing cleanup replaced the error that actually
+made the save fail — and the staged file, which for the two credential stores holds an
+``0600`` copy of IM tokens, stayed on disk with nothing reporting it.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import pytest
+
+from app.channels.message_bus import MessageBus
+from app.channels.runtime_config_store import ChannelRuntimeConfigStore
+from app.channels.store import JsonChannelStore
+from app.channels.wechat import WechatChannel
+
+CLEANUP_ERRNO = 28
+
+
+def _block_destination(destination: Path) -> None:
+    """Make the atomic rename fail with a deterministic errno-21 error.
+
+    A non-empty directory in the destination's place cannot be replaced and cannot be
+    absorbed by the rename (an empty directory would just swallow the staged file).
+    """
+    destination.mkdir()
+    (destination / "occupant").write_text("x")
+
+
+def _staged_files(directory: Path) -> list[str]:
+    return sorted(p.name for p in directory.iterdir() if p.suffix == ".tmp")
+
+
+@pytest.fixture
+def cleanup_fails(monkeypatch):
+    def exploding_unlink(self, *args, **kwargs):
+        raise OSError(CLEANUP_ERRNO, "Simulated cleanup failure")
+
+    monkeypatch.setattr(Path, "unlink", exploding_unlink)
+
+
+class _CloseFails:
+    """A staged file whose ``close()`` always fails, so cleanup has two broken steps."""
+
+    def __init__(self, path: Path) -> None:
+        self.name = str(path)
+        self._path = path
+        self._written: list[str] = []
+
+    def write(self, text: str) -> None:
+        self._written.append(text)
+
+    def close(self) -> None:
+        self._path.write_text("".join(self._written), encoding="utf-8")
+        raise OSError(9, "Simulated close failure")
+
+
+def test_json_channel_store_keeps_the_save_error(tmp_path, cleanup_fails, caplog):
+    path = tmp_path / "store.json"
+    _block_destination(path)
+    store = JsonChannelStore(path)
+    store._data = {"feishu:chat": {"thread_id": "t"}}
+
+    with caplog.at_level(logging.WARNING), pytest.raises(IsADirectoryError) as raised:
+        store._save()
+
+    assert raised.value.errno == 21
+    assert "Failed to remove temporary channel store file" in caplog.text
+
+
+def test_runtime_config_store_keeps_the_save_error(tmp_path, cleanup_fails, caplog):
+    store = ChannelRuntimeConfigStore(tmp_path / "channels.json")
+    _block_destination(store._path)
+
+    with caplog.at_level(logging.WARNING), pytest.raises(IsADirectoryError) as raised:
+        store._save({"feishu": {"app_id": "id"}})
+
+    assert raised.value.errno == 21
+    assert "Failed to remove temporary channel runtime config file" in caplog.text
+
+
+def test_wechat_auth_state_persist_failure_stays_attributable(tmp_path, cleanup_fails, caplog):
+    state = tmp_path / "wechat"
+    state.mkdir()
+    channel = WechatChannel(bus=MessageBus(), config={"bot_token": "token", "state_dir": str(state)})
+    _block_destination(channel._auth_path)
+
+    # A persist failure is reported, not raised: the surviving signal is the log line,
+    # so it has to carry the error that actually happened.
+    with caplog.at_level(logging.WARNING):
+        channel._save_auth_state(status="confirmed")
+
+    records = [r for r in caplog.records if "failed to persist auth state" in r.getMessage()]
+    assert len(records) == 1
+    assert isinstance(records[0].exc_info[1], IsADirectoryError)
+    assert "Failed to remove temporary WeChat auth state file" in caplog.text
+
+
+def test_runtime_config_store_keeps_the_close_error_and_still_unlinks(tmp_path, monkeypatch, caplog):
+    path = tmp_path / "channels.json"
+    store = ChannelRuntimeConfigStore(path)
+    staged = path.parent / "staged.tmp"
+    monkeypatch.setattr("tempfile.NamedTemporaryFile", lambda *args, **kwargs: _CloseFails(staged))
+
+    with caplog.at_level(logging.WARNING), pytest.raises(OSError) as raised:
+        store._save({"feishu": {"app_id": "id"}})
+
+    assert raised.value.errno == 9
+    assert "Failed to close temporary channel runtime config file" in caplog.text
+    # The unlink is an independent step: it must still run after the close failed.
+    assert not staged.exists()
+
+
+def test_runtime_config_store_still_removes_the_staged_file(tmp_path):
+    store = ChannelRuntimeConfigStore(tmp_path / "channels.json")
+    _block_destination(store._path)
+
+    with pytest.raises(IsADirectoryError):
+        store._save({"feishu": {"app_id": "id"}})
+
+    assert _staged_files(tmp_path) == []
+
+
+def test_a_raising_log_handler_does_not_replace_the_save_error(tmp_path, cleanup_fails, caplog):
+    """Reporting the leaked file is itself best-effort.
+
+    A handler can raise out of ``emit`` (a delayed ``FileHandler`` whose directory
+    disappeared raises in the ``flush`` that ``StreamHandler.emit`` runs outside its
+    own try/except), so the diagnostic call sits inside ``suppress``: a broken log
+    must not become the error the caller sees, and must not skip remaining cleanup.
+    """
+    from app.channels import store as store_module
+
+    class _RaisingHandler(logging.Handler):
+        def emit(self, record):
+            raise RuntimeError("Simulated logging failure")
+
+    handler = _RaisingHandler(level=logging.WARNING)
+    store_module.logger.addHandler(handler)
+    try:
+        path = tmp_path / "store.json"
+        _block_destination(path)
+        store = JsonChannelStore(path)
+        store._data = {"feishu:chat": {"thread_id": "t"}}
+
+        with caplog.at_level(logging.WARNING), pytest.raises(IsADirectoryError) as raised:
+            store._save()
+
+        assert raised.value.errno == 21
+    finally:
+        store_module.logger.removeHandler(handler)
+
+
+@pytest.mark.parametrize("failed_step", ["write", "close", "replace"])
+@pytest.mark.parametrize("broken_handler", [False, True])
+def test_buzz_failed_save_keeps_retryable_events_when_cleanup_and_logging_fail(tmp_path, monkeypatch, caplog, failed_step, broken_handler):
+    from app.channels import buzz_seen_events as buzz_module
+
+    path = tmp_path / "seen.json"
+    store = buzz_module.BuzzSeenEventStore(path)
+    store.record("chat", "existing")
+    original_bytes = path.read_bytes()
+    failure = OSError(5, f"primary {failed_step} failure")
+    real_replace = Path.replace
+    staged = tmp_path / "staged.tmp"
+    attempted = []
+
+    class CloseFails(_CloseFails):
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_exc):
+            self.close()
+
+        def close(self):
+            self._path.write_text("".join(self._written), encoding="utf-8")
+            raise failure
+
+    def fail_write(*args, **kwargs):
+        raise failure
+
+    def fail_replace(source, destination):
+        if destination == path:
+            raise failure
+        return real_replace(source, destination)
+
+    def fail_unlink(source, *args, **kwargs):
+        attempted.append(source)
+        raise PermissionError("temporary file is locked")
+
+    handler = logging.FileHandler(tmp_path / "missing-log-directory" / "output.log", delay=True) if broken_handler else None
+    if handler is not None:
+        buzz_module.logger.addHandler(handler)
+    try:
+        with monkeypatch.context() as patch, caplog.at_level(logging.WARNING):
+            if failed_step == "write":
+                patch.setattr(buzz_module.json, "dump", fail_write)
+            elif failed_step == "close":
+                patch.setattr(buzz_module.tempfile, "NamedTemporaryFile", lambda *args, **kwargs: CloseFails(staged))
+            else:
+                patch.setattr(Path, "replace", fail_replace)
+            patch.setattr(Path, "unlink", fail_unlink)
+            store.record("chat", "pending")
+    finally:
+        if handler is not None:
+            buzz_module.logger.removeHandler(handler)
+            handler.close()
+
+    assert attempted, "unlink must still run after a failing close or diagnostic"
+    assert path.read_bytes() == original_bytes
+    assert store.seen("chat", "pending")
+    if not broken_handler:
+        records = [record for record in caplog.records if "failed to persist seen-event store" in record.getMessage()]
+        assert len(records) == 1
+        assert records[0].exc_info[1] is failure
+        assert any("temporary Buzz seen-event" in record.getMessage() for record in caplog.records)
+    # The failed snapshot must remain dirty, including the event recorded before
+    # the failure. A later flush retries without requiring a new incoming event.
+    store.flush()
+    assert buzz_module.BuzzSeenEventStore(path).seen("chat", "existing")
+    assert buzz_module.BuzzSeenEventStore(path).seen("chat", "pending")
+
+
+def test_buzz_write_and_close_failures_keep_the_original_persist_diagnostic(tmp_path, monkeypatch, caplog):
+    from app.channels import buzz_seen_events as buzz_module
+
+    primary = OSError("primary dump failure")
+    staged = tmp_path / "staged.tmp"
+
+    class CloseFails(_CloseFails):
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_exc):
+            self.close()
+
+    def fail_dump(*args, **kwargs):
+        raise primary
+
+    with monkeypatch.context() as patch, caplog.at_level(logging.WARNING):
+        patch.setattr(buzz_module.tempfile, "NamedTemporaryFile", lambda *args, **kwargs: CloseFails(staged))
+        patch.setattr(buzz_module.json, "dump", fail_dump)
+        assert buzz_module.BuzzSeenEventStore(tmp_path / "seen.json")._write_snapshot({"chat": ["pending"]}) is False
+
+    records = [record for record in caplog.records if "failed to persist seen-event store" in record.getMessage()]
+    assert len(records) == 1
+    assert records[0].exc_info[1] is primary
+    assert not staged.exists()
+
+
+@pytest.mark.parametrize("interruption", [KeyboardInterrupt, SystemExit])
+def test_buzz_interruption_cleans_staging_and_still_propagates(tmp_path, monkeypatch, interruption):
+    from app.channels import buzz_seen_events as buzz_module
+
+    primary = interruption("interrupted write")
+
+    def interrupt(*args, **kwargs):
+        raise primary
+
+    with monkeypatch.context() as patch, pytest.raises(interruption) as raised:
+        patch.setattr(buzz_module.json, "dump", interrupt)
+        buzz_module.BuzzSeenEventStore(tmp_path / "seen.json")._write_snapshot({"chat": ["pending"]})
+
+    assert raised.value is primary
+    assert list(tmp_path.iterdir()) == []