From 81194b6456b6c42ace8ccd5d4074184a08170072 Mon Sep 17 00:00:00 2001 From: Reinhold <310554180+reinhold-ph@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:30:00 +0000 Subject: [PATCH 1/3] fix(buzz): discover newly joined channels dynamically --- plugins/platforms/buzz/adapter.py | 45 ++++++++-- tests/gateway/test_buzz_adapter.py | 128 +++++++++++++++++++++++++++++ 2 files changed, 168 insertions(+), 5 deletions(-) diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index 8b77ae334120..74517fb5675a 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -510,8 +510,14 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: except ImportError: self._lock_key = None # status module not available (e.g. tests) + # Start the membership cursor before taking the joined-channel + # snapshot. A join racing with startup is then present either in the + # snapshot or in the membership subscription's inclusive overlap. + if self.transport in ("auto", "websocket"): + self._membership_since = int(time.time()) + # Map channel ids to names and pick the watch set. - code, out, err = await self._run_cli(["channels", "list"]) + code, out, err = await self._run_cli(["channels", "list", "--member"]) if code != 0: message = _cli_error_message(err, code) logger.error("Buzz: failed to list channels — %s", message) @@ -743,7 +749,8 @@ async def _start_websocket(self) -> bool: logger.info("Buzz: WebSocket transport unavailable (%s); falling back to polling", e) return False self._ws_ready = asyncio.Event() - self._membership_since = int(time.time()) + if not self._membership_since: + self._membership_since = int(time.time()) self._ws_task = asyncio.create_task(self._websocket_loop()) try: await asyncio.wait_for(self._ws_ready.wait(), timeout=_WS_AUTH_TIMEOUT + 5) @@ -823,10 +830,12 @@ async def _subscribe_websocket(self, websocket) -> Dict[str, Optional[str]]: return subscriptions async def _handle_membership_event(self, websocket, subscriptions: Dict[str, Optional[str]], event: dict) -> None: - """A membership event p-tagged to us: rediscover conversations and - subscribe to any new ones (fresh DMs dispatch from their beginning).""" - self._membership_since = max(self._membership_since, int(event.get("created_at") or 0)) + """Rediscover and subscribe after a membership event p-tagged to us.""" + event_since = max(int(event.get("created_at") or 0), 0) before = set(self._channel_state) + if not await self._discover_joined_channels(since=event_since): + raise ConnectionError("Buzz joined-channel discovery failed") + self._membership_since = max(self._membership_since, event_since) await self._discover_dms(seed=False) for channel_id in self._channel_state: if channel_id in before: @@ -945,6 +954,32 @@ async def _seed_channel(self, channel_id: str, chat_type: str) -> None: self._maybe_latch_dm(channel_id, state, event) self._trim_seen(state) + async def _discover_joined_channels(self, *, since: int) -> bool: + """Watch newly joined channels when no explicit allowlist is set.""" + if self.channels: + return True + code, out, err = await self._run_cli(["channels", "list", "--member"]) + if code != 0: + logger.warning( + "Buzz: failed to rediscover joined channels — %s", + _cli_error_message(err, code), + ) + return False + for channel in _parse_json_list(out): + channel_id = str(channel.get("channel_id") or "") + if not channel_id: + continue + self._channel_meta[channel_id] = channel + self._channel_names[channel_id] = str(channel.get("name") or channel_id) + if channel_id in self._channel_state: + continue + self._channel_state[channel_id] = { + "chat_type": "group", + "last_ts": max(int(since), 0), + "seen": OrderedDict(), + } + return True + async def _discover_dms(self, *, seed: bool) -> None: """Watch DM conversations. New ones found mid-run dispatch from their beginning (a fresh conversation has no history worth suppressing); diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index d612021bfe39..4eab2638f3da 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -381,6 +381,134 @@ async def test_dm_shaped_channel_discovered_when_dms_list_empty(self): assert a._may_reclassify_as_dm(CHANNEL) is False +# ── Dynamic joined-channel discovery ───────────────────────────────────── + + +class TestChannelDiscovery: + + @pytest.mark.asyncio + async def test_connect_lists_only_joined_channels(self, monkeypatch): + import gateway.status as gateway_status + + monkeypatch.setattr( + gateway_status, "acquire_scoped_lock", lambda platform, key: True + ) + monkeypatch.setattr(_buzz_mod, "_resolve_private_key", lambda extra=None: "nsec1test") + monkeypatch.setattr(_buzz_mod.time, "time", lambda: 1000) + adapter = _make_adapter() + adapter.cli_path = "/fake/buzz" + adapter._start_websocket = AsyncMock(return_value=False) + unjoined_channel = "12c81eb7-3a12-47c1-b8af-c66f1c74ca8b" + cli = _ScriptedCli() + cli.script( + "users", "get", + [{"pubkey": SELF_PUBKEY, "display_name": "Chip"}], + ) + cli.script("messages", "get", []) + cli.script("dms", "list", []) + membership_cursors = [] + + async def run_cli(args, *, input_text=None): + if args == ["channels", "list", "--member"]: + membership_cursors.append(adapter._membership_since) + return 0, json.dumps([ + {"channel_id": CHANNEL, "name": "general", "description": "General"}, + ]), "" + if args == ["channels", "list"]: + return 0, json.dumps([ + {"channel_id": CHANNEL, "name": "general", "description": "General"}, + {"channel_id": unjoined_channel, "name": "other", "description": "Other"}, + ]), "" + return await cli(args, input_text=input_text) + + adapter._run_cli = run_cli + + try: + assert await adapter.connect() is True + finally: + await adapter.disconnect() + + assert membership_cursors == [1000] + assert unjoined_channel not in adapter._channel_state + + @pytest.mark.asyncio + async def test_membership_event_subscribes_to_new_joined_channel(self): + adapter = _make_adapter() + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": CHANNEL, "name": "general", "description": "General"}, + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("dms", "list", []) + adapter._run_cli = cli + websocket = AsyncMock() + subscriptions = {"hermes-buzz-0": CHANNEL} + + await adapter._handle_membership_event( + websocket, + subscriptions, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + assert new_channel in adapter._channel_state + assert adapter._channel_state[new_channel]["chat_type"] == "group" + assert adapter._channel_state[new_channel]["last_ts"] == 1234 + assert new_channel in subscriptions.values() + assert (["channels", "list", "--member"], None) in cli.calls + request = json.loads(websocket.send.await_args.args[0]) + assert request[2]["#h"] == [new_channel] + assert request[2]["since"] == 1233 + + @pytest.mark.asyncio + async def test_membership_event_respects_explicit_channel_allowlist(self): + adapter = _make_adapter({"channels": [CHANNEL]}) + adapter._channel_state[CHANNEL] = { + "chat_type": "group", "last_ts": 100, "seen": {}, + } + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("dms", "list", []) + adapter._run_cli = cli + websocket = AsyncMock() + subscriptions = {"hermes-buzz-0": CHANNEL} + + await adapter._handle_membership_event( + websocket, + subscriptions, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + assert new_channel not in adapter._channel_state + assert new_channel not in subscriptions.values() + websocket.send.assert_not_awaited() + + @pytest.mark.asyncio + async def test_membership_event_retries_after_joined_channel_discovery_failure(self): + adapter = _make_adapter() + adapter._membership_since = 100 + cli = _ScriptedCli() + cli.script("channels", "list", [], code=2, stderr="temporary failure") + adapter._run_cli = cli + websocket = AsyncMock() + + with pytest.raises(ConnectionError, match="joined-channel discovery failed"): + await adapter._handle_membership_event( + websocket, + {"hermes-buzz-0": CHANNEL}, + {"created_at": 1234, "kind": _buzz_mod._WS_MEMBERSHIP_KIND}, + ) + + assert adapter._membership_since == 100 + websocket.send.assert_not_awaited() + + # ── Sending ─────────────────────────────────────────────────────────────── From 3f6fb1cf7b37a66472cc5362e0fa98838f4528ec Mon Sep 17 00:00:00 2001 From: "Stingel (AgentOS)" <9309469+coldcrippy@users.noreply.github.com> Date: Thu, 6 Aug 2026 05:25:43 +0000 Subject: [PATCH 2/3] fix(buzz): add channel-scoped free response policy --- plugins/platforms/buzz/adapter.py | 29 +++++++++++++++--- tests/gateway/test_buzz_adapter.py | 36 +++++++++++++++++++++++ website/docs/user-guide/messaging/buzz.md | 2 ++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index 74517fb5675a..ea633440efb0 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -25,6 +25,7 @@ cli_path: "" # path to the buzz binary (default: PATH, then ~/bin/buzz) credentials_file: "" # JSON file holding the nsec (fallback for BUZZ_PRIVATE_KEY) allowed_users: [] # empty = allow all; entries are hex pubkeys or npubs + free_response_channels: [] # channel UUIDs that do not require an @mention Or via environment variables (overrides config.yaml): BUZZ_RELAY_URL, BUZZ_CHANNELS, BUZZ_HOME_CHANNEL, BUZZ_POLL_INTERVAL, @@ -392,6 +393,21 @@ def __init__(self, config, **kwargs): _rm_cfg = _rm_raw self.require_mention = str(_rm_cfg).strip().lower() not in ("false", "0", "no", "off") + # Profile-scoped channel exemptions from the global mention gate. This + # mirrors the cross-platform ``free_response_channels`` vocabulary used + # by Discord and Slack. It changes relevance only: channel discovery, + # membership, allowed-user, and self-echo gates remain independent. + raw_free_response = extra.get("free_response_channels", []) + if isinstance(raw_free_response, (list, tuple, set)): + free_response_values = raw_free_response + else: + free_response_values = str(raw_free_response).split(",") + self.free_response_channels: set[str] = { + str(channel_id).strip() + for channel_id in free_response_values + if str(channel_id).strip() + } + # Inbound transport: "auto" (WebSocket with poll fallback, default), # "websocket" (require WS; fail connect when it can't authenticate), # or "poll" (CLI polling only). Env (BUZZ_TRANSPORT) overrides @@ -1065,10 +1081,15 @@ async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None self._maybe_latch_dm(channel_id, state, event) is_dm = state["chat_type"] == "dm" - # In shared channels, respond only when addressed — unless - # require_mention is disabled, in which case respond to every message. - # DMs always dispatch. - if not is_dm and self.require_mention and not self._is_mentioned(content): + # In shared channels, respond only when addressed unless mention gating + # is globally disabled or this exact channel is explicitly exempted. + # DMs always dispatch. Relevance never bypasses author authorization. + if ( + not is_dm + and self.require_mention + and channel_id not in self.free_response_channels + and not self._is_mentioned(content) + ): return # Adapter-level allow-list (the gateway applies BUZZ_ALLOWED_USERS / diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index 4eab2638f3da..4068c9817965 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -243,6 +243,42 @@ async def test_name_mention_dispatched(self, adapter): await self._poll_with(adapter, _event("e1", content="hey @Chip can you help?", created_at=10)) assert len(adapter._dispatched) == 1 + @pytest.mark.asyncio + async def test_free_response_channel_dispatches_without_mention(self): + adapter = _make_adapter({"free_response_channels": [CHANNEL]}) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + await self._poll_with(adapter, _event("e1", content="ordinary owner message", created_at=10)) + assert [d["message_id"] for d in adapter._dispatched] == ["e1"] + + @pytest.mark.asyncio + async def test_dm_dispatches_without_mention(self, adapter): + adapter._channel_state[CHANNEL]["chat_type"] = "dm" + await self._poll_with(adapter, _event("e1", content="ordinary DM", created_at=10)) + assert [d["message_id"] for d in adapter._dispatched] == ["e1"] + + @pytest.mark.asyncio + async def test_self_echo_stays_suppressed_in_free_response_channel(self): + adapter = _make_adapter({"free_response_channels": [CHANNEL]}) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + adapter._channel_state[CHANNEL] = {"chat_type": "group", "last_ts": 0, "seen": {}} + await self._poll_with( + adapter, + _event("e1", pubkey=SELF_PUBKEY, content="agent echo", created_at=10), + ) + assert adapter._dispatched == [] @pytest.mark.asyncio async def test_allowlist_blocks_unauthorized(self, adapter): diff --git a/website/docs/user-guide/messaging/buzz.md b/website/docs/user-guide/messaging/buzz.md index 848e21c0532e..96217ef0a287 100644 --- a/website/docs/user-guide/messaging/buzz.md +++ b/website/docs/user-guide/messaging/buzz.md @@ -34,6 +34,7 @@ gateway: cli_path: "" # buzz binary (default: PATH, then ~/bin/buzz) credentials_file: "" # JSON file with the nsec (BUZZ_PRIVATE_KEY fallback) allowed_users: [] # empty = allow all; hex pubkeys or npubs + free_response_channels: [] # channel UUIDs where ordinary messages do not require @mention ``` Plus, in `~/.hermes/.env`: @@ -80,6 +81,7 @@ gateway: credentials_file: "" # JSON file with the nsec (BUZZ_PRIVATE_KEY fallback) allowed_users: [] # empty = allow all if allow_all_users is true; otherwise restrict to listed npubs/hex pubkeys require_mention: true # in channels: only respond when addressed (@name, npub, or hex pubkey); DMs always dispatch regardless + free_response_channels: [] # exact channel UUIDs that bypass require_mention; authorization still applies allow_all_users: false # set true for community mode (everyone can chat, only owner is admin); false for private mode (only allowed_users) ``` From a5851b1c2c0f3c09706a19ca279da93967f6355e Mon Sep 17 00:00:00 2001 From: "Stingel (AgentOS)" <9309469+coldcrippy@users.noreply.github.com> Date: Thu, 6 Aug 2026 05:34:53 +0000 Subject: [PATCH 3/3] fix(buzz): seed live discovery before dispatch --- plugins/platforms/buzz/adapter.py | 63 ++++++++++------- tests/gateway/test_buzz_adapter.py | 108 +++++++++++++++++++++++++++++ 2 files changed, 147 insertions(+), 24 deletions(-) diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index ea633440efb0..b4a12cb439b8 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -848,14 +848,9 @@ async def _subscribe_websocket(self, websocket) -> Dict[str, Optional[str]]: async def _handle_membership_event(self, websocket, subscriptions: Dict[str, Optional[str]], event: dict) -> None: """Rediscover and subscribe after a membership event p-tagged to us.""" event_since = max(int(event.get("created_at") or 0), 0) - before = set(self._channel_state) - if not await self._discover_joined_channels(since=event_since): - raise ConnectionError("Buzz joined-channel discovery failed") + discovered = await self._discover_conversations(since=event_since) self._membership_since = max(self._membership_since, event_since) - await self._discover_dms(seed=False) - for channel_id in self._channel_state: - if channel_id in before: - continue + for channel_id in discovered: subscription_id = f"hermes-buzz-dm-{len(subscriptions)}" subscriptions[subscription_id] = channel_id await self._send_channel_subscription(websocket, subscription_id, channel_id) @@ -933,7 +928,7 @@ async def _poll_loop(self) -> None: self._poll_count += 1 try: if self._poll_count % _DM_DISCOVERY_EVERY == 0: - await self._discover_dms(seed=False) + await self._discover_conversations(since=int(time.time())) for channel_id in list(self._channel_state): await self._poll_channel(channel_id) except asyncio.CancelledError: @@ -943,9 +938,15 @@ async def _poll_loop(self) -> None: except asyncio.CancelledError: raise - async def _seed_channel(self, channel_id: str, chat_type: str) -> None: + async def _seed_channel(self, channel_id: str, chat_type: str, *, floor_ts: int = 0) -> None: """Initialize a channel's high-water mark from its newest events.""" - state = {"chat_type": chat_type, "last_ts": 0, "seen": OrderedDict()} + floor_ts = max(int(floor_ts), 0) + state = { + "chat_type": chat_type, + "last_ts": floor_ts, + "not_before_ts": floor_ts, + "seen": OrderedDict(), + } self._channel_state[channel_id] = state code, out, err = await self._run_cli( ["messages", "get", "--channel", channel_id, "--limit", str(_FETCH_LIMIT)] @@ -956,7 +957,7 @@ async def _seed_channel(self, channel_id: str, chat_type: str) -> None: ) # Fall back to "now" so a transiently unreadable channel does not # replay its whole history once it becomes readable. - state["last_ts"] = int(time.time()) + state["last_ts"] = max(state["last_ts"], int(time.time())) return for event in _parse_json_list(out): event_id = event.get("id") @@ -970,6 +971,14 @@ async def _seed_channel(self, channel_id: str, chat_type: str) -> None: self._maybe_latch_dm(channel_id, state, event) self._trim_seen(state) + async def _discover_conversations(self, *, since: int) -> List[str]: + """Discover joined channels and DMs, seeding each before delivery.""" + before = set(self._channel_state) + if not await self._discover_joined_channels(since=since): + raise ConnectionError("Buzz joined-channel discovery failed") + await self._discover_dms(seed=True, floor_ts=since) + return [channel_id for channel_id in self._channel_state if channel_id not in before] + async def _discover_joined_channels(self, *, since: int) -> bool: """Watch newly joined channels when no explicit allowlist is set.""" if self.channels: @@ -989,17 +998,11 @@ async def _discover_joined_channels(self, *, since: int) -> bool: self._channel_names[channel_id] = str(channel.get("name") or channel_id) if channel_id in self._channel_state: continue - self._channel_state[channel_id] = { - "chat_type": "group", - "last_ts": max(int(since), 0), - "seen": OrderedDict(), - } + await self._seed_channel(channel_id, chat_type="group", floor_ts=since) return True - async def _discover_dms(self, *, seed: bool) -> None: - """Watch DM conversations. New ones found mid-run dispatch from their - beginning (a fresh conversation has no history worth suppressing); - ones present at startup are seeded like channels. + async def _discover_dms(self, *, seed: bool, floor_ts: int = 0) -> None: + """Watch DM conversations, optionally seeding history before delivery. ``dms list`` is only a best-effort source: on some hosted relays it returns ``[]`` even when DM conversations exist (#68871). Those DMs @@ -1016,9 +1019,14 @@ async def _discover_dms(self, *, seed: bool) -> None: if not dm_id or dm_id in self._channel_state: continue if seed: - await self._seed_channel(dm_id, chat_type="dm") + await self._seed_channel(dm_id, chat_type="dm", floor_ts=floor_ts) else: - self._channel_state[dm_id] = {"chat_type": "dm", "last_ts": 0, "seen": OrderedDict()} + self._channel_state[dm_id] = { + "chat_type": "dm", + "last_ts": floor_ts, + "not_before_ts": floor_ts, + "seen": OrderedDict(), + } self._channel_names.setdefault(dm_id, "DM") code, out, _err = await self._run_cli(["channels", "list"]) @@ -1033,9 +1041,14 @@ async def _discover_dms(self, *, seed: bool) -> None: if ch_id in self._channel_state or not self._may_reclassify_as_dm(ch_id): continue if seed: - await self._seed_channel(ch_id, chat_type="group") + await self._seed_channel(ch_id, chat_type="group", floor_ts=floor_ts) else: - self._channel_state[ch_id] = {"chat_type": "group", "last_ts": 0, "seen": OrderedDict()} + self._channel_state[ch_id] = { + "chat_type": "group", + "last_ts": floor_ts, + "not_before_ts": floor_ts, + "seen": OrderedDict(), + } async def _poll_channel(self, channel_id: str) -> None: state = self._channel_state.get(channel_id) @@ -1060,6 +1073,8 @@ async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None """De-dupe, filter, and dispatch a single ``messages get`` event.""" event_id = str(event.get("id") or "") created_at = int(event.get("created_at") or 0) + if created_at < int(state.get("not_before_ts") or 0): + return if not event_id or event_id in state["seen"]: return state["seen"][event_id] = None diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index 4068c9817965..7109eea6ce69 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -280,6 +280,48 @@ async def capture(**kwargs): ) assert adapter._dispatched == [] + @pytest.mark.asyncio + async def test_other_channel_still_requires_mention(self): + other_channel = "12c81eb7-3a12-47c1-b8af-c66f1c74ca8b" + adapter = _make_adapter({"free_response_channels": [CHANNEL]}) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + state = {"chat_type": "group", "last_ts": 0, "seen": {}} + adapter._channel_state[other_channel] = state + await adapter._handle_event( + other_channel, + state, + _tagged_event("e1", other_channel, content="ordinary shared-room message"), + ) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_free_response_channel_preserves_author_allowlist(self): + adapter = _make_adapter({ + "free_response_channels": [CHANNEL], + "allowed_users": ["b" * 64], + }) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + state = {"chat_type": "group", "last_ts": 0, "seen": {}} + adapter._channel_state[CHANNEL] = state + await adapter._handle_event( + CHANNEL, + state, + _event("e1", content="ordinary unauthorized message"), + ) + assert adapter._dispatched == [] + @pytest.mark.asyncio async def test_allowlist_blocks_unauthorized(self, adapter): adapter._allowed_pubkeys = {"b" * 64} @@ -422,6 +464,72 @@ async def test_dm_shaped_channel_discovered_when_dms_list_empty(self): class TestChannelDiscovery: + @pytest.mark.asyncio + async def test_poll_discovery_seeds_new_joined_channel_without_replay(self): + new_channel = "4764ae67-7cd8-4f3e-967d-7dd93986b11a" + adapter = _make_adapter({"free_response_channels": [new_channel]}) + adapter._dispatched = [] + + async def capture(**kwargs): + adapter._dispatched.append(kwargs) + + adapter._dispatch_message = capture + adapter._message_handler = AsyncMock() + cli = _ScriptedCli() + cli.script("channels", "list", [ + {"channel_id": new_channel, "name": "project", "description": "Project"}, + ]) + cli.script("messages", "get", [ + _tagged_event("old", new_channel, content="old history", created_at=1500), + ]) + cli.script("dms", "list", []) + adapter._run_cli = cli + + discovered = await adapter._discover_conversations(since=2000) + assert discovered == [new_channel] + assert adapter._channel_state[new_channel]["not_before_ts"] == 2000 + assert adapter._dispatched == [] + + await adapter._handle_event( + new_channel, + adapter._channel_state[new_channel], + _tagged_event("older-unseen", new_channel, content="older history", created_at=1600), + ) + assert adapter._dispatched == [] + + await adapter._handle_event( + new_channel, + adapter._channel_state[new_channel], + _tagged_event("new", new_channel, content="new live message", created_at=2001), + ) + assert [d["message_id"] for d in adapter._dispatched] == ["new"] + + @pytest.mark.asyncio + async def test_dynamic_dm_discovery_seeds_existing_history(self): + adapter = _make_adapter() + adapter._dispatched = [] + cli = _ScriptedCli() + cli.script("channels", "list", []) + cli.script("dms", "list", [{"dm_id": DM_CHANNEL}]) + cli.script("messages", "get", [ + _tagged_event( + "old-dm", + DM_CHANNEL, + content="old direct history", + created_at=1500, + p=SELF_PUBKEY, + ), + ]) + adapter._run_cli = cli + + discovered = await adapter._discover_conversations(since=2000) + assert discovered == [DM_CHANNEL] + state = adapter._channel_state[DM_CHANNEL] + assert state["chat_type"] == "dm" + assert state["not_before_ts"] == 2000 + assert "old-dm" in state["seen"] + assert adapter._dispatched == [] + @pytest.mark.asyncio async def test_connect_lists_only_joined_channels(self, monkeypatch): import gateway.status as gateway_status