diff --git a/README.md b/README.md index d5d85934f4..2048978d4d 100644 --- a/README.md +++ b/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 `..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 | diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 7c48fdf46f..8fb288abca 100644 --- a/backend/AGENTS.md +++ b/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**: diff --git a/backend/app/channels/buzz_seen_events.py b/backend/app/channels/buzz_seen_events.py index 855320e2c1..d2a8a83462 100644 --- a/backend/app/channels/buzz_seen_events.py +++ b/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: diff --git a/backend/app/channels/credential_staging.py b/backend/app/channels/credential_staging.py new file mode 100644 index 0000000000..50f1f2e164 --- /dev/null +++ b/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) diff --git a/backend/app/channels/runtime_config_store.py b/backend/app/channels/runtime_config_store.py index 715170c110..1dac6f8a21 100644 --- a/backend/app/channels/runtime_config_store.py +++ b/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: diff --git a/backend/app/channels/store.py b/backend/app/channels/store.py index 53f0cf75a5..8a7a854289 100644 --- a/backend/app/channels/store.py +++ b/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 ------------------------------------------------------- diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index 37512acaba..0a90230e12 100644 --- a/backend/app/channels/wechat.py +++ b/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 diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index 9a7d218266..e334b05fd7 100644 --- a/backend/tests/AGENTS.md +++ b/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` diff --git a/backend/tests/blocking_io/test_wechat_channel_state.py b/backend/tests/blocking_io/test_wechat_channel_state.py index af76dec8be..150bb06848 100644 --- a/backend/tests/blocking_io/test_wechat_channel_state.py +++ b/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" diff --git a/backend/tests/test_channel_credential_staging.py b/backend/tests/test_channel_credential_staging.py new file mode 100644 index 0000000000..98ca9183e7 --- /dev/null +++ b/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() diff --git a/backend/tests/test_channel_staging_cleanup.py b/backend/tests/test_channel_staging_cleanup.py new file mode 100644 index 0000000000..da29d2d160 --- /dev/null +++ b/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()) == []