diff --git a/plugins/platforms/buzz/adapter.py b/plugins/platforms/buzz/adapter.py index b07830ae9b6e4..b118bed40b7d0 100644 --- a/plugins/platforms/buzz/adapter.py +++ b/plugins/platforms/buzz/adapter.py @@ -320,6 +320,38 @@ def _parse_json_list(stdout: str) -> List[dict]: return [item for item in data if isinstance(item, dict)] +def _event_reply_parent_id(event: dict) -> Optional[str]: + """Resolve a chat event's direct parent event id (NIP-10 ``e`` tags). + + Prefer a ``reply``-marked tag, then a ``root``-marked tag, else the last + positional ``e`` tag. Buzz Desktop thread replies typically carry both + root and reply markers; the reply marker is the direct parent. + """ + tags = event.get("tags") + if not isinstance(tags, list): + return None + reply_id: Optional[str] = None + root_id: Optional[str] = None + last_e: Optional[str] = None + for tag in tags: + if not isinstance(tag, (list, tuple)) or len(tag) < 2 or tag[0] != "e": + continue + target = str(tag[1] or "").strip() + if not target: + continue + marker = str(tag[3] or "") if len(tag) > 3 else "" + last_e = target + if marker == "reply": + reply_id = target + elif marker == "root": + root_id = target + return reply_id or root_id or last_e + + +# Cap stored parent content snippets (gateway reply injection also clips). +_EVENT_META_CONTENT_CAP = 500 + + # --------------------------------------------------------------------------- # Buzz Adapter # --------------------------------------------------------------------------- @@ -404,7 +436,13 @@ def __init__(self, config, **kwargs): self._ws_active = False # True while the WS loop owns inbound delivery self._membership_since = 0 self._lock_key: Optional[str] = None - # channel_id -> {"chat_type", "last_ts", "seen": OrderedDict[event_id, None]} + # channel_id -> { + # "chat_type", "last_ts", + # "seen": OrderedDict[event_id, None], + # "event_meta": OrderedDict[event_id, (author_pubkey, content_snippet)], + # } + # event_meta backs NIP-10 reply-parent resolution for require_mention + # (thread replies to our own messages count as addressed — #75826). self._channel_state: Dict[str, dict] = {} self._channel_names: Dict[str, str] = {} # channel_id -> raw ``channels list`` entry; drives DM-vs-channel @@ -603,7 +641,12 @@ async def send( if event_id: # Belt-and-braces echo suppression: the poll loop already skips # our own pubkey, but marking the id seen makes de-dupe explicit. + # Also record event_meta so a thread reply to this send matches + # even if the WS/poll echo never arrives (#75826). self._mark_seen(str(chat_id), str(event_id)) + self._remember_event_meta( + str(chat_id), str(event_id), self._self_pubkey, content + ) return SendResult( success=bool(data.get("accepted", True)), message_id=str(event_id) if event_id else None, @@ -669,6 +712,12 @@ async def send_image( event_id = data.get("event_id") if event_id: self._mark_seen(str(chat_id), str(event_id)) + self._remember_event_meta( + str(chat_id), + str(event_id), + self._self_pubkey, + caption or "", + ) return SendResult( success=bool(data.get("accepted", True)), message_id=str(event_id) if event_id else None, @@ -894,9 +943,17 @@ async def _poll_loop(self) -> None: except asyncio.CancelledError: raise + def _new_channel_state(self, chat_type: str) -> dict: + return { + "chat_type": chat_type, + "last_ts": 0, + "seen": OrderedDict(), + "event_meta": OrderedDict(), + } + async def _seed_channel(self, channel_id: str, chat_type: str) -> None: """Initialize a channel's high-water mark from its newest events.""" - state = {"chat_type": chat_type, "last_ts": 0, "seen": OrderedDict()} + state = self._new_channel_state(chat_type) self._channel_state[channel_id] = state code, out, err = await self._run_cli( ["messages", "get", "--channel", channel_id, "--limit", str(_FETCH_LIMIT)] @@ -915,6 +972,10 @@ async def _seed_channel(self, channel_id: str, chat_type: str) -> None: if event_id: state["seen"][str(event_id)] = None state["last_ts"] = max(state["last_ts"], created_at) + # History is never dispatched, but it still classifies and feeds + # the event_meta cache so post-restart thread replies to messages + # we sent before the gateway came up still match (#75826). + self._remember_event(state, event) # History is never dispatched, but it still classifies: a DM that # leaked in via ``channels list`` latches to chat_type="dm" here, # so it bypasses the mention gate from the very first poll. @@ -943,7 +1004,7 @@ async def _discover_dms(self, *, seed: bool) -> None: if seed: await self._seed_channel(dm_id, chat_type="dm") else: - self._channel_state[dm_id] = {"chat_type": "dm", "last_ts": 0, "seen": OrderedDict()} + self._channel_state[dm_id] = self._new_channel_state("dm") self._channel_names.setdefault(dm_id, "DM") code, out, _err = await self._run_cli(["channels", "list"]) @@ -960,7 +1021,7 @@ async def _discover_dms(self, *, seed: bool) -> None: if seed: await self._seed_channel(ch_id, chat_type="group") else: - self._channel_state[ch_id] = {"chat_type": "group", "last_ts": 0, "seen": OrderedDict()} + self._channel_state[ch_id] = self._new_channel_state("group") async def _poll_channel(self, channel_id: str) -> None: state = self._channel_state.get(channel_id) @@ -997,6 +1058,10 @@ async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None if not pubkey or not isinstance(content, str) or not content.strip(): return + # Feed the per-channel event cache before any early return so self-echo + # and concurrent-author traffic can still be reply parents (#75826). + self._remember_event(state, event) + # Suppress self-echo: never dispatch our own messages back to the agent. if pubkey == self._self_pubkey: return @@ -1006,10 +1071,19 @@ 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" + reply_parent_id = _event_reply_parent_id(event) + reply_meta = self._lookup_event_meta(state, reply_parent_id) if reply_parent_id else None + reply_to_is_own = bool( + reply_meta is not None and reply_meta[0] == self._self_pubkey + ) # 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): + # A NIP-10 thread reply whose direct parent is one of our messages is + # treated as addressed (parity with Signal/WhatsApp; fixes #75826 — + # e.g. Desktop "/approve session" replies that never type @name). + # DMs always dispatch. p-tag semantics stay untouched (#68871). + mentioned = self._is_mentioned(content) + if not is_dm and self.require_mention and not mentioned and not reply_to_is_own: return # Adapter-level allow-list (the gateway applies BUZZ_ALLOWED_USERS / @@ -1032,6 +1106,10 @@ async def _handle_event(self, channel_id: str, state: dict, event: dict) -> None user_name=await self._resolve_user_name(pubkey), message_id=event_id, created_at=created_at, + reply_to_message_id=reply_parent_id, + reply_to_text=reply_meta[1] if reply_meta else None, + reply_to_author_id=reply_meta[0] if reply_meta else None, + reply_to_is_own_message=reply_to_is_own, ) # ── DM classification (issue #68871) ────────────────────────────────── @@ -1179,6 +1257,10 @@ def _trim_seen(state: dict) -> None: seen = state["seen"] while len(seen) > _SEEN_CAP: seen.popitem(last=False) + meta = state.get("event_meta") + if isinstance(meta, OrderedDict): + while len(meta) > _SEEN_CAP: + meta.popitem(last=False) def _mark_seen(self, channel_id: str, event_id: str) -> None: state = self._channel_state.get(channel_id) @@ -1186,6 +1268,55 @@ def _mark_seen(self, channel_id: str, event_id: str) -> None: state["seen"][event_id] = None self._trim_seen(state) + def _remember_event(self, state: dict, event: dict) -> None: + """Record author + content snippet for later NIP-10 parent lookup.""" + event_id = str(event.get("id") or "") + if not event_id: + return + pubkey = str(event.get("pubkey") or "").lower() + content = event.get("content") + snippet = content[:_EVENT_META_CONTENT_CAP] if isinstance(content, str) else "" + self._store_event_meta(state, event_id, pubkey, snippet) + + def _remember_event_meta( + self, + channel_id: str, + event_id: str, + pubkey: str, + content: str, + ) -> None: + state = self._channel_state.get(channel_id) + if state is None or not event_id: + return + snippet = (content or "")[:_EVENT_META_CONTENT_CAP] + self._store_event_meta(state, event_id, (pubkey or "").lower(), snippet) + + @staticmethod + def _store_event_meta( + state: dict, + event_id: str, + pubkey: str, + snippet: str, + ) -> None: + cache = state.setdefault("event_meta", OrderedDict()) + if not isinstance(cache, OrderedDict): + cache = OrderedDict(cache) + state["event_meta"] = cache + cache[event_id] = (pubkey, snippet) + cache.move_to_end(event_id) + while len(cache) > _SEEN_CAP: + cache.popitem(last=False) + + @staticmethod + def _lookup_event_meta(state: dict, event_id: Optional[str]) -> Optional[Tuple[str, str]]: + if not event_id: + return None + cache = state.get("event_meta") or {} + entry = cache.get(event_id) + if not entry or not isinstance(entry, tuple) or len(entry) < 2: + return None + return str(entry[0] or ""), str(entry[1] or "") + async def _dispatch_message( self, text: str, @@ -1195,6 +1326,10 @@ async def _dispatch_message( user_name: str, message_id: str, created_at: int, + reply_to_message_id: Optional[str] = None, + reply_to_text: Optional[str] = None, + reply_to_author_id: Optional[str] = None, + reply_to_is_own_message: bool = False, ) -> None: """Build a MessageEvent and hand it to the base class handler.""" if not self._message_handler: @@ -1214,10 +1349,14 @@ async def _dispatch_message( source=source, message_id=message_id, timestamp=datetime.fromtimestamp(created_at) if created_at else datetime.now(), + reply_to_message_id=reply_to_message_id, + reply_to_text=reply_to_text, + reply_to_author_id=reply_to_author_id, + reply_to_is_own_message=reply_to_is_own_message, ) await self.handle_message(event) - + # Add a "seen" reaction after dispatching — signals to the user that # their message was received and is being processed. try: diff --git a/tests/gateway/test_buzz_adapter.py b/tests/gateway/test_buzz_adapter.py index d612021bfe39f..7a4dbf5ccdf6d 100644 --- a/tests/gateway/test_buzz_adapter.py +++ b/tests/gateway/test_buzz_adapter.py @@ -19,6 +19,7 @@ _normalize_user_ref = _buzz_mod._normalize_user_ref _cli_error_message = _buzz_mod._cli_error_message _resolve_private_key = _buzz_mod._resolve_private_key +_event_reply_parent_id = _buzz_mod._event_reply_parent_id check_requirements = _buzz_mod.check_requirements validate_config = _buzz_mod.validate_config register = _buzz_mod.register @@ -251,19 +252,20 @@ async def test_allowlist_blocks_unauthorized(self, adapter): assert adapter._dispatched == [] -# ── DM classification via p-tags (issue #68871) ────────────────────────── +# ── NIP-10 thread replies as addressed (issue #75826) ──────────────────── # -# `buzz dms list` returns [] on some hosted relays, so DM conversations leak -# in via `channels list` and get seeded chat_type="group". The adapter must -# reclassify them from the Nostr tags of real traffic: DM messages are -# p-tagged to our own pubkey WITHOUT the text mentioning us, while channel -# messages only ever p-tag us when the text visibly @mentions us. +# With require_mention (default), channel replies whose direct parent is the +# agent's own message must dispatch even when the text has no @name — Buzz +# Desktop's natural reply affordance for /approve never types a mention. def _tagged_event(event_id, channel, *, content, pubkey=OTHER_PUBKEY, - created_at=1000, kind=9, p=None, reply_to=None): + created_at=1000, kind=9, p=None, reply_to=None, root=None): """Event with the tag shapes observed on a live relay (h/p/e tags).""" tags = [["h", channel]] + # NIP-10 order as Desktop emits: root first, then reply (when both set). + if root: + tags.append(["e", root, "", "root"]) if reply_to: tags.append(["e", reply_to, "", "reply"]) if p: @@ -278,6 +280,245 @@ def _tagged_event(event_id, channel, *, content, pubkey=OTHER_PUBKEY, } +class TestNip10ThreadReplyMentionGate: + """require_mention + NIP-10 reply-to-own-message (#75826).""" + + @pytest.fixture + def adapter(self): + a = _make_adapter() + a._dispatched = [] + + async def capture(**kwargs): + a._dispatched.append(kwargs) + + a._dispatch_message = capture + a._message_handler = AsyncMock() + a._channel_state[CHANNEL] = a._new_channel_state("group") + return a + + async def _poll_with(self, adapter, *events): + cli = _ScriptedCli() + cli.script("messages", "get", list(events)) + adapter._run_cli = cli + await adapter._poll_channel(CHANNEL) + + def test_event_reply_parent_prefers_reply_marker(self): + ev = _tagged_event( + "child", CHANNEL, content="ok", root="root-id", reply_to="parent-id" + ) + assert _event_reply_parent_id(ev) == "parent-id" + assert _event_reply_parent_id( + _tagged_event("c2", CHANNEL, content="ok", root="only-root") + ) == "only-root" + + @pytest.mark.asyncio + async def test_thread_reply_to_own_message_dispatches_without_mention(self, adapter): + # Live agent prompt lands first (self-echo is cached, not dispatched). + await self._poll_with( + adapter, + _tagged_event( + "agent-prompt", + CHANNEL, + content="⚠️ Dangerous command requires approval", + pubkey=SELF_PUBKEY, + created_at=10, + ), + _tagged_event( + "user-reply", + CHANNEL, + content="sure go ahead", + root="agent-prompt", + reply_to="agent-prompt", + created_at=11, + ), + ) + assert [d["message_id"] for d in adapter._dispatched] == ["user-reply"] + assert adapter._dispatched[0]["text"] == "sure go ahead" + assert adapter._dispatched[0]["reply_to_message_id"] == "agent-prompt" + assert adapter._dispatched[0]["reply_to_is_own_message"] is True + assert "approval" in (adapter._dispatched[0]["reply_to_text"] or "") + + @pytest.mark.asyncio + async def test_approve_thread_reply_dispatches(self, adapter): + await self._poll_with( + adapter, + _tagged_event( + "agent-approve-prompt", + CHANNEL, + content="⚠️ Dangerous command requires approval", + pubkey=SELF_PUBKEY, + created_at=20, + ), + _tagged_event( + "approve-msg", + CHANNEL, + content="/approve session", + root="agent-approve-prompt", + reply_to="agent-approve-prompt", + created_at=21, + ), + ) + assert [d["message_id"] for d in adapter._dispatched] == ["approve-msg"] + assert adapter._dispatched[0]["text"] == "/approve session" + assert adapter._dispatched[0]["reply_to_is_own_message"] is True + + @pytest.mark.asyncio + async def test_reply_to_other_user_stays_gated(self, adapter): + third = "c" * 64 + await self._poll_with( + adapter, + _tagged_event( + "other-msg", + CHANNEL, + content="anyone around?", + pubkey=third, + created_at=30, + ), + _tagged_event( + "reply-other", + CHANNEL, + content="yeah I'm here", + root="other-msg", + reply_to="other-msg", + created_at=31, + ), + ) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_reply_to_unknown_parent_stays_gated(self, adapter): + await self._poll_with( + adapter, + _tagged_event( + "orphan-reply", + CHANNEL, + content="/approve session", + root="never-seen", + reply_to="never-seen", + created_at=40, + ), + ) + assert adapter._dispatched == [] + + @pytest.mark.asyncio + async def test_seeded_own_history_matches_thread_reply(self, adapter): + """Replies to agent messages sent before a gateway restart still match.""" + cli = _ScriptedCli() + cli.script( + "messages", + "get", + [ + _tagged_event( + "pre-restart-agent", + CHANNEL, + content="⚠️ Dangerous command requires approval", + pubkey=SELF_PUBKEY, + created_at=50, + ), + ], + ) + adapter._run_cli = cli + await adapter._seed_channel(CHANNEL, chat_type="group") + assert "pre-restart-agent" in adapter._channel_state[CHANNEL]["event_meta"] + assert adapter._dispatched == [] + + cli.responses.clear() + cli.script( + "messages", + "get", + [ + _tagged_event( + "post-restart-approve", + CHANNEL, + content="/approve always", + root="pre-restart-agent", + reply_to="pre-restart-agent", + created_at=51, + ), + ], + ) + await adapter._poll_channel(CHANNEL) + assert [d["message_id"] for d in adapter._dispatched] == ["post-restart-approve"] + assert adapter._dispatched[0]["reply_to_is_own_message"] is True + + @pytest.mark.asyncio + async def test_send_recorded_id_matches_thread_reply(self, adapter): + """send()'s returned event_id is cached even without a WS/poll echo.""" + cli = _ScriptedCli() + cli.script( + "messages", + "send", + {"accepted": True, "event_id": "sent-prompt", "message": ""}, + ) + adapter._run_cli = cli + result = await adapter.send( + CHANNEL, "⚠️ Dangerous command requires approval" + ) + assert result.success is True + assert "sent-prompt" in adapter._channel_state[CHANNEL]["event_meta"] + meta = adapter._channel_state[CHANNEL]["event_meta"]["sent-prompt"] + assert meta[0] == SELF_PUBKEY + + cli.responses.clear() + cli.script( + "messages", + "get", + [ + _tagged_event( + "reply-to-send", + CHANNEL, + content="/approve session", + root="sent-prompt", + reply_to="sent-prompt", + created_at=61, + ), + ], + ) + await adapter._poll_channel(CHANNEL) + assert [d["message_id"] for d in adapter._dispatched] == ["reply-to-send"] + assert adapter._dispatched[0]["reply_to_is_own_message"] is True + assert adapter._dispatched[0]["reply_to_message_id"] == "sent-prompt" + + @pytest.mark.asyncio + async def test_mention_path_still_populates_reply_context(self, adapter): + """Visible @mention + thread reply still fills reply_to_* on dispatch.""" + await self._poll_with( + adapter, + _tagged_event( + "agent-prior", + CHANNEL, + content="previous answer", + pubkey=SELF_PUBKEY, + created_at=70, + ), + _tagged_event( + "mentioned-reply", + CHANNEL, + content="@Chip follow up please", + root="agent-prior", + reply_to="agent-prior", + created_at=71, + ), + ) + assert len(adapter._dispatched) == 1 + d = adapter._dispatched[0] + assert d["message_id"] == "mentioned-reply" + assert d["text"] == "follow up please" # leading @Chip stripped + assert d["reply_to_message_id"] == "agent-prior" + assert d["reply_to_author_id"] == SELF_PUBKEY + assert d["reply_to_is_own_message"] is True + assert d["reply_to_text"] == "previous answer" + + +# ── DM classification via p-tags (issue #68871) ────────────────────────── +# +# `buzz dms list` returns [] on some hosted relays, so DM conversations leak +# in via `channels list` and get seeded chat_type="group". The adapter must +# reclassify them from the Nostr tags of real traffic: DM messages are +# p-tagged to our own pubkey WITHOUT the text mentioning us, while channel +# messages only ever p-tag us when the text visibly @mentions us. + + class TestDmClassification: @pytest.fixture