fix(buzz): drop replayed events across reconnects with a persistent seen-id store (#4888)

* fix(buzz): drop replayed events across reconnects with a persistent seen-id store

The Buzz connector's resubscribe filter replays by design: 'since' is the
created_at of the last processed event and NIP-01 'since' is inclusive, so
every relay reconnect redelivers at least that event. The guard against
re-running the agent on it was the manager's inbound dedupe, whose default
store is in-process with a 10-minute TTL — so any reconnect more than ten
minutes after a channel's last message (or any gateway restart) re-answered
that message. Users saw the agent respond to an old question after every
relay restart.

Fix: persist the ids of fully processed events per channel
(BuzzSeenEventStore, JSON under {base_dir}/channels/, atomic writes) and
drop redelivered ids in _handle_chat_event before the /connect branch —
a replayed /connect would otherwise be re-answered with a spurious
'code invalid or expired'. Matching is by exact event id only, never
timestamp, so a genuinely new event (same-second or clock-skewed author)
can never be skipped, preserving the connector's fail-toward-replay
invariant. Only fully processed events are recorded, mirroring the
watermark rule: a gated drop or failed publish stays replayable.

Fail-open in both directions: an unreadable store loads empty (costs one
replayed reply, the previous behavior) and a failed write is logged and
retried on the next record. Id lists and the channel map are bounded like
the connector's other remote-fed maps. The persistent path is wired in
ChannelService (like channel_store); directly constructed channels get a
memory-only store so tests and tooling stay free of filesystem side
effects.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(buzz): coalesce seen-store writes, clean up temp files, harden docs and coverage

Address review on the seen-event store:
- record() now marks the store dirty and coalesces persistence to one
  write per FLUSH_DELAY_SECONDS on the event loop, so a reconnect
  backlog burst pays one O(store) file write instead of one per event;
  sync callers (no running loop) keep immediate writes, and
  BuzzChannel.stop() flushes so a clean shutdown loses nothing. A crash
  inside the window only costs replay, never a skip.
- _save() unlinks its temp file on failure (ChannelStore parity), so a
  persistently unwritable path no longer accumulates *.tmp litter.
- Module docstring now documents that restart protection is bounded to
  the newest MAX_IDS_PER_CHANNEL ids per channel (and to raise it if a
  relay ever serves a deeper default backlog), and pins the
  single-event-loop assumption that makes the class safe without a lock.
- New tests: MAX_CHANNELS LRU eviction, coalescing behavior, flush
  idempotence, temp-file cleanup, stop() flushing, and the
  ChannelService wiring that injects seen_event_store_path (the line
  that makes real deployments durable).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix(buzz): reschedule the coalesced flush when the pending timer's loop is gone

A pending flush handle pinned to a since-closed event loop kept
_flush_handle non-None forever, so later record() calls on a new loop
never scheduled a timer and the store silently stopped persisting until
an explicit flush(). Track the scheduling loop (TimerHandle has no
public get_loop()) and reschedule when it differs from the running one.
Unreachable in production (one loop per process, stop() flushes), but
now hardened and tested.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

backend/app/channels/buzz.py

@@ -50,6 +50,7 @@ from urllib.parse import urlparse
 
 from app.channels import buzz_nostr
 from app.channels.base import Channel
+from app.channels.buzz_seen_events import BuzzSeenEventStore
 from app.channels.commands import is_known_channel_command
 from app.channels.connection_identity import attach_connection_identity
 from app.channels.message_bus import InboundMessage, InboundMessageType, InboundQueueFullError, MessageBus, OutboundMessage
@@ -273,6 +274,14 @@ class BuzzChannel(Channel):
         self._last_requester: dict[tuple[str, str | None], str] = {}
         self._pending_auth_challenge: str | None = None  # set from an AUTH relay frame; consumed by the NIP-42 flow in _session
         self._seen_created_at: dict[str, int] = {}  # channel id -> high-water mark of PROCESSED created_at (see _advance_watermark)
+        # Persistent processed-event-id record. The resubscribe filter replays
+        # by design (inclusive ``since``), and the manager's inbound dedupe has
+        # a 10-minute in-process TTL — so without durable ids, every relay
+        # reconnect (or gateway restart) re-answers the last message in each
+        # channel. Dedupe here is by exact event id only, never timestamp, so
+        # it cannot skip a genuinely new event. ``seen_event_store`` is a test
+        # seam; the default persists under ``{base_dir}/channels/``.
+        self._seen_events: BuzzSeenEventStore = config.get("seen_event_store") or BuzzSeenEventStore(config.get("seen_event_store_path"))
         self._chat_subscriptions: set[str] = set()  # channel ids with a live per-channel REQ on the CURRENT connection
         self._resubscribe_attempts: dict[str, int] = {}  # sub id -> CLOSED-recovery attempts spent on the current connection/auth epoch
         self._auth_completed = False  # has THIS socket's NIP-42 handshake been ACKNOWLEDGED by the relay? (see _handle_auth_ok, _AUTH_REQUIRED_CLOSE_PREFIX)
@@ -369,6 +378,9 @@ class BuzzChannel(Channel):
         self._pending_auth_event_id = None
         self._session_started_at = None
         self._pending_auth_challenge = None
+        # Seen-id persistence is coalesced (FLUSH_DELAY_SECONDS); a clean stop
+        # must not lose records still inside that window to a replay on restart.
+        self._seen_events.flush()
         logger.info("[buzz] channel stopped")
 
     # -- subscriptions ------------------------------------------------------
@@ -1231,6 +1243,21 @@ class BuzzChannel(Channel):
         channel_id = channel_id_values[0]
         created_at = int(ev.get("created_at", 0))
         text = str(ev.get("content", ""))
+        event_id = str(ev.get("id", ""))
+
+        # Drop redeliveries of events this connector already fully processed.
+        # The resubscribe filter replays the watermark event on every reconnect
+        # (inclusive ``since``), and after a long disconnect or cursor eviction
+        # the relay's default backlog comes back too. The manager's inbound
+        # dedupe only covers a 10-minute in-process window, so without this a
+        # relay reconnect re-answers the last message in each channel. Matching
+        # is by exact event id, so a new event — whatever its author-chosen
+        # created_at — is never affected. Sits before the /connect branch
+        # deliberately: a replayed /connect would otherwise be re-answered with
+        # a spurious "code invalid or expired" reply on every reconnect.
+        if self._seen_events.seen(channel_id, event_id):
+            logger.debug("[buzz] dropped replayed event id=%s in channel %s", event_id, channel_id)
+            return
 
         # /connect <code> must be consulted before the allowlist gate (framework
         # ordering rule — see Channel._pending_connect_code) so a not-yet-bound
@@ -1252,6 +1279,7 @@ class BuzzChannel(Channel):
             # cursor: leaving it behind would replay this /connect on every reconnect
             # and answer each replay with a spurious "code invalid or expired" reply.
             self._advance_watermark(channel_id, created_at)
+            self._seen_events.record(channel_id, event_id)
             return
         if author not in self._allowed_users:
             # Deny-by-default is intentional (see start()'s empty-allowlist warning),
@@ -1283,8 +1311,11 @@ class BuzzChannel(Channel):
         await self._publish(inbound)
         # Only a fully accepted-and-published event advances the cursor, and only
         # after the publish actually succeeded. A dropped event must never move it
-        # (that was the DoS), and a failed publish must leave it replayable.
+        # (that was the DoS), and a failed publish must leave it replayable. The
+        # same rule applies to the persistent seen-id record: recording before a
+        # failed publish would turn "replayable" into "silently skipped".
         self._advance_watermark(channel_id, created_at)
+        self._seen_events.record(channel_id, event_id)
 
     # -- outbound --------------------------------------------------------------
 

backend/app/channels/buzz_seen_events.py

@@ -0,0 +1,207 @@
+"""Persistent record of Buzz chat events that were fully processed.
+
+Why this exists
+---------------
+The Buzz connector's resubscribe filter deliberately replays rather than skips:
+``since`` is the created_at of the last processed event and NIP-01 ``since`` is
+inclusive, so every reconnect redelivers at least that event (see
+``BuzzChannel._chat_filter``). The manager's inbound dedupe absorbs those
+redeliveries — but its default store is in-process with a 10-minute TTL, so a
+reconnect more than 10 minutes after the last message (or any gateway restart)
+re-runs the agent on an already-answered message.
+
+This store closes that gap at the connector: the ids of fully processed events
+are persisted per channel, and a redelivered id is dropped before it reaches the
+bus. Dedupe is by exact event id only — never by timestamp — so a genuinely new
+event (which always has a fresh id, whatever its author-chosen created_at) can
+never be skipped, preserving the connector's fail-toward-replay invariant.
+
+Failure policy is fail-open in both directions: an unreadable file loads as
+empty (costing at most one replayed answer, the pre-existing behavior) and a
+failed write is logged and retried on the next flush (costing replay, never a
+skip). The id lists are bounded per channel and the channel map is bounded like
+the connector's other remote-fed maps. That per-channel bound also bounds the
+restart protection itself: after a gateway restart ``_seen_created_at`` is
+empty, the resubscribe REQ carries no ``since``, and the relay's default
+backlog replays — only the newest ``MAX_IDS_PER_CHANNEL`` processed ids per
+channel are dropped, so a relay backlog deeper than that would re-answer the
+tail. If a relay ever serves a deeper default backlog, raise
+``MAX_IDS_PER_CHANNEL`` here.
+
+Writes are coalesced: ``record()`` marks the store dirty and schedules one
+flush per ``FLUSH_DELAY_SECONDS`` on the running event loop, so a reconnect
+backlog burst pays one O(store) file write instead of one per event. With no
+running loop (tests, tooling) ``record()`` flushes synchronously, and
+``BuzzChannel.stop()`` flushes pending state on shutdown. This class is not
+thread-safe by design: everything runs on the single channel event loop
+(mutation in ``_handle_chat_event``, the coalesced flush via ``call_later`` on
+that same loop). Anyone adding an off-loop user — e.g. a threaded flusher —
+must add a lock around ``_ids``/``_sets`` first, as ChannelStore does.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import tempfile
+from collections import OrderedDict, deque
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+# Coalescing window for persisting the store. Losing this window's records in a
+# crash only costs replay (fail-open), never a skip.
+FLUSH_DELAY_SECONDS = 1.0
+
+# Ids retained per channel. Reconnect replay is normally the single watermark
+# event; the deep case is a channel whose cursor was evicted, which replays the
+# relay's default backlog window. Both are far below this bound.
+MAX_IDS_PER_CHANNEL = 512
+# Channel-map cap, mirroring the connector's other remote-fed maps
+# (channel ids arrive in remote ``h`` tags).
+MAX_CHANNELS = 512
+
+
+class BuzzSeenEventStore:
+    """Bounded, JSON-persisted map of channel id -> recently processed event ids."""
+
+    def __init__(self, path: str | Path | None = None) -> None:
+        # ``path=None`` means memory-only: no file is read or written, which is
+        # exactly the pre-existing (non-durable) behavior. The channel service
+        # wires the persistent path for real deployments; constructing a
+        # channel directly (tests, tooling) must not create directories or
+        # files as a side effect.
+        self._path = Path(path) if path is not None else None
+        self._ids: OrderedDict[str, deque[str]] = OrderedDict()
+        self._sets: dict[str, set[str]] = {}
+        self._loaded = False
+        self._dirty = False
+        self._flush_handle: asyncio.TimerHandle | None = None
+        # The loop the pending handle was scheduled on. TimerHandle has no
+        # public get_loop(), so it is tracked here to detect a stale handle.
+        self._flush_loop: asyncio.AbstractEventLoop | None = None
+
+    # -- persistence ---------------------------------------------------------
+
+    def _ensure_loaded(self) -> None:
+        if self._loaded:
+            return
+        self._loaded = True
+        if self._path is None:
+            return
+        try:
+            if not self._path.exists():
+                return
+            raw = json.loads(self._path.read_text(encoding="utf-8"))
+        except Exception:
+            logger.warning("[buzz] unreadable seen-event store, starting fresh (costs at most one replayed reply)", exc_info=True)
+            return
+        if not isinstance(raw, dict):
+            return
+        for channel_id, ids in raw.items():
+            if not isinstance(ids, list):
+                continue
+            clean = deque((str(i) for i in ids if i), maxlen=MAX_IDS_PER_CHANNEL)
+            self._ids[str(channel_id)] = clean
+            self._sets[str(channel_id)] = set(clean)
+        self._enforce_channel_cap()
+
+    def _save(self) -> None:
+        if self._path is None:
+            return
+        tmp_name: str | None = None
+        try:
+            path = self._path
+            path.parent.mkdir(parents=True, exist_ok=True)
+            payload = {channel: list(ids) for channel, ids in self._ids.items()}
+            # Atomic same-directory replace, matching ChannelStore._save: a
+            # 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)
+            self._dirty = False
+        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)
+
+    def _request_flush(self) -> None:
+        """Coalesce persistence: at most one write per FLUSH_DELAY_SECONDS.
+
+        With no running event loop (tests, tooling) the write happens
+        synchronously, preserving the immediate-durability semantics direct
+        callers had before coalescing existed.
+        """
+        self._dirty = True
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self.flush()
+            return
+        # A pending handle pinned to a since-closed loop would otherwise block
+        # scheduling forever, silently stopping persistence until an explicit
+        # flush() (only reachable when callers span loops, e.g. tests).
+        if self._flush_handle is None or self._flush_loop is not loop:
+            if self._flush_handle is not None:
+                self._flush_handle.cancel()
+            self._flush_loop = loop
+            self._flush_handle = loop.call_later(FLUSH_DELAY_SECONDS, self._flush_scheduled)
+
+    def _flush_scheduled(self) -> None:
+        self._flush_handle = None
+        if self._dirty:
+            self._save()
+
+    def flush(self) -> None:
+        """Persist pending records now (no-op when nothing is dirty).
+
+        Called on channel stop so a clean shutdown never loses records to the
+        coalescing window; a crash inside the window only costs replay.
+        """
+        if self._flush_handle is not None:
+            self._flush_handle.cancel()
+            self._flush_handle = None
+        if self._dirty:
+            self._save()
+
+    def _enforce_channel_cap(self) -> None:
+        while len(self._ids) > MAX_CHANNELS:
+            evicted, _ = self._ids.popitem(last=False)
+            self._sets.pop(evicted, None)
+
+    # -- api ----------------------------------------------------------------
+
+    def seen(self, channel_id: str, event_id: str) -> bool:
+        """True if *event_id* was already fully processed in *channel_id*."""
+        if not event_id:
+            return False
+        self._ensure_loaded()
+        return event_id in self._sets.get(channel_id, ())
+
+    def record(self, channel_id: str, event_id: str) -> None:
+        """Record a fully processed event and schedule a coalesced persist."""
+        if not channel_id or not event_id:
+            return
+        self._ensure_loaded()
+        ids = self._ids.get(channel_id)
+        if ids is None:
+            ids = deque(maxlen=MAX_IDS_PER_CHANNEL)
+            self._ids[channel_id] = ids
+            self._sets[channel_id] = set()
+        id_set = self._sets[channel_id]
+        if event_id in id_set:
+            return
+        if len(ids) == ids.maxlen:
+            id_set.discard(ids[0])
+        ids.append(event_id)
+        id_set.add(event_id)
+        # Move the channel to the back so the channel-cap eviction is LRU-ish.
+        self._ids.move_to_end(channel_id)
+        self._enforce_channel_cap()
+        self._request_flush()

backend/app/channels/service.py

@@ -7,6 +7,7 @@ import logging
 import math
 import os
 from collections.abc import Callable
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from app.channels.base import Channel
@@ -392,6 +393,14 @@ class ChannelService:
         try:
             config = dict(config)
             config["channel_store"] = self.store
+            if name == "buzz" and "seen_event_store_path" not in config:
+                # Durable processed-event ids for the Buzz connector's replay
+                # guard. Wired here (like channel_store) rather than defaulted
+                # inside the connector so that directly constructed channels
+                # (tests, tooling) stay free of filesystem side effects.
+                from deerflow.config.paths import get_paths
+
+                config["seen_event_store_path"] = str(Path(get_paths().base_dir) / "channels" / "buzz_seen_events.json")
             if self._connection_repo is not None:
                 config["connection_repo"] = self._connection_repo
             channel = channel_cls(bus=self.bus, config=config)

backend/tests/test_buzz_seen_events.py

@@ -0,0 +1,347 @@
+"""Tests for the Buzz connector's persistent replay guard (seen-event store)."""
+
+import asyncio
+import json
+
+import pytest
+
+pytest.importorskip("coincurve")
+
+from app.channels import buzz_nostr
+from app.channels.buzz import BuzzChannel
+from app.channels.buzz_seen_events import MAX_IDS_PER_CHANNEL, BuzzSeenEventStore
+from app.channels.message_bus import MessageBus
+
+# Keypairs mirror tests/test_buzz_channel.py: inbound events are
+# signature-verified, so fixture authors must be real keypairs.
+SK3_HEX = "0000000000000000000000000000000000000000000000000000000000000003"
+PK3_HEX = "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9"
+SK_OWNER = "0000000000000000000000000000000000000000000000000000000000000005"
+OWNER = buzz_nostr.parse_private_key(SK_OWNER).pubkey_hex
+CHANNEL = "136852ee-63e1-49c2-8927-413b5ee8e5f7"
+
+
+def _event(*, sk=SK_OWNER, kind=9, content="@DeerFlow hello", channel=CHANNEL, mentions=(PK3_HEX,), reply_to=None, created_at=1700000100):
+    tags = [["h", channel]]
+    if reply_to:
+        tags.append(["e", reply_to])
+    tags.extend(["p", m] for m in mentions)
+    return buzz_nostr.sign_event(buzz_nostr.parse_private_key(sk), kind, tags, content, created_at)
+
+
+# -- store ------------------------------------------------------------------
+
+
+def test_memory_only_store_records_without_touching_disk(tmp_path):
+    store = BuzzSeenEventStore(None)
+    store.record(CHANNEL, "e1")
+    assert store.seen(CHANNEL, "e1")
+    assert not store.seen(CHANNEL, "e2")
+    assert list(tmp_path.iterdir()) == []  # nothing written anywhere we can observe
+
+
+def test_persistent_store_round_trips_across_instances(tmp_path):
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+    store.record(CHANNEL, "e1")
+    store.record(CHANNEL, "e2")
+    reloaded = BuzzSeenEventStore(path)
+    assert reloaded.seen(CHANNEL, "e1") and reloaded.seen(CHANNEL, "e2")
+    assert not reloaded.seen(CHANNEL, "e3")
+    assert not reloaded.seen("other-channel", "e1")
+
+
+def test_per_channel_id_cap_evicts_oldest(tmp_path):
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+    for i in range(MAX_IDS_PER_CHANNEL + 5):
+        store.record(CHANNEL, f"id-{i}")
+    assert not store.seen(CHANNEL, "id-0")
+    assert store.seen(CHANNEL, f"id-{MAX_IDS_PER_CHANNEL + 4}")
+    # The persisted form respects the cap too.
+    persisted = json.loads(path.read_text())
+    assert len(persisted[CHANNEL]) == MAX_IDS_PER_CHANNEL
+
+
+def test_corrupt_store_file_fails_open_to_empty(tmp_path):
+    path = tmp_path / "seen.json"
+    path.write_text("{not json", encoding="utf-8")
+    store = BuzzSeenEventStore(path)
+    assert not store.seen(CHANNEL, "e1")
+    store.record(CHANNEL, "e1")  # recovers by overwriting on next record
+    assert BuzzSeenEventStore(path).seen(CHANNEL, "e1")
+
+
+def test_empty_event_id_is_never_seen_or_recorded(tmp_path):
+    store = BuzzSeenEventStore(tmp_path / "seen.json")
+    store.record(CHANNEL, "")
+    assert not store.seen(CHANNEL, "")
+
+
+def test_unwritable_path_fails_open(tmp_path):
+    blocker = tmp_path / "blocker"
+    blocker.write_text("")  # a file where the store expects a parent directory
+    store = BuzzSeenEventStore(blocker / "seen.json")
+    store.record(CHANNEL, "e1")  # write fails, logged, no raise
+    assert store.seen(CHANNEL, "e1")  # in-memory state still works
+
+
+# -- connector integration ---------------------------------------------------
+
+
+def _started_with_store(store: BuzzSeenEventStore):
+    ch = BuzzChannel(
+        bus=MessageBus(),
+        config={
+            "relay_url": "wss://buzz.example.com",
+            "private_key": SK3_HEX,
+            "allowed_users": [OWNER],
+            "seen_event_store": store,
+        },
+    )
+    ch._keys = buzz_nostr.parse_private_key(SK3_HEX)
+    captured = []
+
+    async def publish(msg):
+        captured.append(msg)
+
+    ch._publish = publish
+    return ch, captured
+
+
+def _dispatch(ch, ev):
+    asyncio.run(ch.handle_relay_frame(json.dumps(["EVENT", "sub1", ev])))
+
+
+def test_redelivered_event_is_dropped_within_one_process(tmp_path):
+    ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json"))
+    ev = _event()
+    _dispatch(ch, ev)
+    assert len(captured) == 1
+    _dispatch(ch, ev)  # relay replay of the same event (inclusive `since`)
+    assert len(captured) == 1
+
+
+def test_redelivered_event_is_dropped_across_restart(tmp_path):
+    """The bug: a relay reconnect after a gateway restart re-answered the last message."""
+    path = tmp_path / "seen.json"
+    ev = _event()
+    store1 = BuzzSeenEventStore(path)
+    ch1, captured1 = _started_with_store(store1)
+    _dispatch(ch1, ev)
+    assert len(captured1) == 1
+    store1.flush()  # what BuzzChannel.stop() does on a clean shutdown
+
+    # Simulated restart: a fresh channel instance, fresh store object, same file.
+    ch2, captured2 = _started_with_store(BuzzSeenEventStore(path))
+    _dispatch(ch2, ev)
+    assert captured2 == []
+
+
+def test_new_event_at_same_created_at_is_still_processed(tmp_path):
+    """Dedupe is by id only: a same-second new event must never be skipped."""
+    ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json"))
+    _dispatch(ch, _event(content="@DeerFlow first", created_at=1700000100))
+    _dispatch(ch, _event(content="@DeerFlow second", created_at=1700000100))
+    assert len(captured) == 2
+
+
+def test_older_created_at_new_event_is_still_processed(tmp_path):
+    """A clock-skewed author's new event (older timestamp) must never be skipped."""
+    ch, captured = _started_with_store(BuzzSeenEventStore(tmp_path / "seen.json"))
+    _dispatch(ch, _event(content="@DeerFlow newer", created_at=1700000200))
+    _dispatch(ch, _event(content="@DeerFlow older-clock", created_at=1700000100))
+    assert len(captured) == 2
+
+
+def test_dropped_event_is_not_recorded_and_stays_replayable(tmp_path):
+    """Only fully processed events are recorded — a gated drop must not poison the id."""
+    store = BuzzSeenEventStore(tmp_path / "seen.json")
+    ch, captured = _started_with_store(store)
+    ev = _event(content="no mention here", mentions=())  # dropped by the mention gate
+    _dispatch(ch, ev)
+    assert captured == []
+    assert not store.seen(CHANNEL, str(ev["id"]))
+
+
+def test_replayed_connect_is_not_reprocessed(tmp_path):
+    """A replayed /connect must not be re-answered with 'code invalid or expired'."""
+    store = BuzzSeenEventStore(tmp_path / "seen.json")
+    ch, captured = _started_with_store(store)
+    replies = []
+
+    async def fake_bind(code, author, channel_id):
+        replies.append(code)
+
+    ch._bind_connection = fake_bind
+    ch._connection_repo = object()  # /connect is only consulted when connections are configured
+    ev = _event(sk=SK_OWNER, content="/connect abc123", mentions=(PK3_HEX,))
+    _dispatch(ch, ev)
+    assert replies == ["abc123"]
+    _dispatch(ch, ev)  # replay
+    assert replies == ["abc123"]
+
+
+def test_default_config_uses_memory_only_store():
+    """Direct construction (no service wiring) must not persist anywhere."""
+    ch = BuzzChannel(
+        bus=MessageBus(),
+        config={"relay_url": "wss://buzz.example.com", "private_key": SK3_HEX, "allowed_users": [OWNER]},
+    )
+    assert isinstance(ch._seen_events, BuzzSeenEventStore)
+    assert ch._seen_events._path is None
+
+
+def test_channel_cap_evicts_least_recently_recorded(tmp_path):
+    from app.channels.buzz_seen_events import MAX_CHANNELS
+
+    store = BuzzSeenEventStore(tmp_path / "seen.json")
+    for i in range(MAX_CHANNELS):
+        store.record(f"chan-{i}", f"id-{i}")
+    # Touch the oldest channel so LRU ordering (move_to_end) protects it.
+    store.record("chan-0", "id-0b")
+    store.record("chan-new", "id-new")  # overflows the map by one
+    assert store.seen("chan-0", "id-0")  # refreshed -> survived
+    assert store.seen("chan-new", "id-new")
+    assert not store.seen("chan-1", "id-1")  # now the least recently used -> evicted
+    # The persisted form respects the cap too.
+    persisted = json.loads((tmp_path / "seen.json").read_text())
+    assert len(persisted) == MAX_CHANNELS
+    assert "chan-1" not in persisted
+
+
+def test_saves_are_coalesced_under_a_running_loop(tmp_path):
+    """One timer (and at most one file write) per burst, not one write per record."""
+    from app.channels import buzz_seen_events
+
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+    saves = []
+    original_save = store._save
+
+    def counting_save():
+        saves.append(1)
+        original_save()
+
+    store._save = counting_save
+
+    async def burst():
+        for i in range(50):
+            store.record(CHANNEL, f"id-{i}")
+        assert saves == []  # nothing written yet: flush is pending on the loop
+        assert store._flush_handle is not None
+        await asyncio.sleep(buzz_seen_events.FLUSH_DELAY_SECONDS + 0.1)
+
+    asyncio.run(burst())
+    assert len(saves) == 1
+    persisted = json.loads(path.read_text())
+    assert len(persisted[CHANNEL]) == 50
+
+
+def test_flush_persists_pending_records_and_is_idempotent(tmp_path):
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+
+    async def record_only():
+        store.record(CHANNEL, "e1")
+        assert not path.exists()  # still inside the coalescing window
+
+    asyncio.run(record_only())
+    store.flush()
+    assert BuzzSeenEventStore(path).seen(CHANNEL, "e1")
+    mtime = path.stat().st_mtime_ns
+    store.flush()  # nothing dirty -> no rewrite
+    assert path.stat().st_mtime_ns == mtime
+
+
+def test_failed_save_leaves_no_tmp_files(tmp_path, monkeypatch):
+    """A persistent write failure must not accumulate *.tmp files (ChannelStore parity)."""
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+
+    def boom(*args, **kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("app.channels.buzz_seen_events.json.dump", boom)
+    for i in range(3):
+        store.record(CHANNEL, f"id-{i}")  # sync context -> each record attempts a save
+    assert not path.exists()
+    assert [p.name for p in tmp_path.iterdir()] == []  # no *.tmp litter
+    monkeypatch.undo()
+    store.record(CHANNEL, "id-3")  # recovery: next flush persists everything
+    assert json.loads(path.read_text())[CHANNEL] == ["id-0", "id-1", "id-2", "id-3"]
+
+
+def test_channel_stop_flushes_pending_records(tmp_path):
+    """A clean stop must not lose records still inside the coalescing window."""
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+    ch, captured = _started_with_store(store)
+    ev = _event()
+
+    async def dispatch_then_stop():
+        await ch.handle_relay_frame(json.dumps(["EVENT", "sub1", ev]))
+        assert not path.exists()  # flush still pending
+        ch._running = True  # stop() is a no-op on a never-started channel
+        ch.bus.subscribe_outbound(ch._on_outbound)
+        await ch.stop()
+
+    asyncio.run(dispatch_then_stop())
+    assert len(captured) == 1
+    assert BuzzSeenEventStore(path).seen(CHANNEL, str(ev["id"]))
+
+
+def test_service_wiring_injects_persistent_store_path(tmp_path, monkeypatch):
+    """The _start_channel wiring is what makes real deployments durable — a
+    silent regression there would revert to memory-only with all unit tests green."""
+    from app.channels.service import ChannelService
+
+    captured_config = {}
+
+    class StubChannel:
+        def __init__(self, bus, config):
+            captured_config.update(config)
+            self.is_running = True
+
+        async def start(self):
+            pass
+
+    class StubPaths:
+        base_dir = str(tmp_path)
+
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: StubPaths())
+    monkeypatch.setattr("deerflow.reflection.resolve_class", lambda path, base_class=None: StubChannel)
+
+    service = ChannelService(channels_config={})
+    started = asyncio.run(service._start_channel("buzz", {"relay_url": "wss://x", "private_key": SK3_HEX}))
+    assert started
+    expected = str(tmp_path / "channels" / "buzz_seen_events.json")
+    assert captured_config["seen_event_store_path"] == expected
+    # An explicitly configured path must win over the default wiring.
+    captured_config.clear()
+    asyncio.run(service._start_channel("buzz", {"relay_url": "wss://x", "private_key": SK3_HEX, "seen_event_store_path": "/custom/seen.json"}))
+    assert captured_config["seen_event_store_path"] == "/custom/seen.json"
+
+
+def test_stale_timer_from_closed_loop_does_not_block_rescheduling(tmp_path):
+    """A pending flush timer pinned to a since-closed loop must not stop the
+    store from scheduling on a new loop (it would silently stop persisting)."""
+    from app.channels import buzz_seen_events
+
+    path = tmp_path / "seen.json"
+    store = BuzzSeenEventStore(path)
+
+    async def record_and_abandon():
+        store.record(CHANNEL, "e1")  # schedules a timer on THIS loop...
+
+    asyncio.run(record_and_abandon())  # ...which closes before the timer fires
+    assert store._flush_handle is not None  # the stale handle survives the loop
+
+    async def record_on_new_loop():
+        store.record(CHANNEL, "e2")
+        assert store._flush_loop is asyncio.get_running_loop()
+        await asyncio.sleep(buzz_seen_events.FLUSH_DELAY_SECONDS + 0.1)
+
+    asyncio.run(record_on_new_loop())
+    persisted = json.loads(path.read_text())
+    assert persisted[CHANNEL] == ["e1", "e2"]  # both records made it to disk