diff --git a/plans/2026-09-05-dm-address-resolution.md b/plans/2026-09-05-dm-address-resolution.md index 3871787a..2a64d586 100644 --- a/plans/2026-09-05-dm-address-resolution.md +++ b/plans/2026-09-05-dm-address-resolution.md @@ -23,13 +23,13 @@ 各 channel 的对应关系如下，均为调研结论： -- Telegram：按对方是不是 bot 分流。`sendMessage` 的 `chat_id` 文档写的是「target **bot**, supergroup or - channel 的 @username」，普通用户不在其列，因此**只有 bot 才用 `@username`，人一律用数字 id**； - 私聊的 chat_id 与对方的用户 id 相同。bot 没有 username 时退回数字 id，不构造飞书会拒绝的地址。 - Bot API 10.0（2026-05-08）起，两个机器人在双方于 BotFather 打开开关后可以按 `@username` 互发私聊； - 该开关无法探测（`getMe` 的 `User` 没有对应字段），因此这条是乐观的，成立与否由发送结果决定。 - 为承载 username，`TelegramThreadIdentity.chat_id` 由 `int` 放宽为 `int | str`：数字仍按原样解析， - 历史身份串的 uuid5 因此不变，已有会话不会被重新编号。 +- Telegram：取 `SenderIdentity.id`，即对方的用户 id，私聊的 chat_id 与之相同，人与 bot 同样对待。 + `@username` 只用于**打开一个尚不存在的私聊**：Bot API 10.0（2026-05-08）的 changelog 写的是 + 「Added the ability to send messages to other bots via username if both bots enabled bot-to-bot + communication」，文档中没有一处说 bot 可以用数字 id 作为尚未建立的私聊目标；而私聊一旦存在，数字 + chat_id 即可送达（本节点向 `telegram::7181589532:0` 发出的四条出站消息均为 `sent`）。该开关无法 + 探测（`getMe` 的 `User` 没有对应字段），因此首次发送是乐观的，成立与否由发送结果决定。 + `TelegramThreadIdentity.chat_id` 保持 `int | str`：历史身份串仍按原样解析，其 uuid5 不变。 - 飞书：取 `SenderIdentity.id`，即 `open_id`，而 `open_id` 是 `im/v1/messages` 的 `receive_id_type` 合法取值， 可直接作为 `receive_id`。`im.message.receive_v1` 的 `event_sender` 结构为 `sender_id{union_id, user_id, open_id}` 加 `sender_type`（取值 `user` 或 `bot`），两种发送者共用同一结构， @@ -49,11 +49,10 @@ 1. 按现有逻辑解析 `dm:@name`，命中即结束。 2. 未命中且目标形如 `dm:@name` 时，在当前 actor 可达范围内按 handle 查历史入站发送者， 取其 `sender_id` 与所属 channel。 -3. 把该身份交给 channel 换取私聊的 provider 地址。 -4. 换到地址则建立 `channel_sessions` 与 `threads` 两行并重新解析；换不到则抛 - `InboxTargetResolutionError`，`bcc` 调用方看到的仍是「找不到」。 - -发送失败不回滚已建立的映射。地址换不出来时没有别的补救手段，保留映射与删除映射对调用方没有区别。 +3. 把该身份交给 channel 换取私聊的 provider 地址；换不到则抛 `InboxTargetResolutionError`， + `bcc` 调用方看到的仍是「找不到」。 +4. 该地址对应的会话可能已经在库里、只是这个 token 找不到它；按 `(channel, provider_thread_id)` + 查到就按它继续，查不到才按「落库在投递之后」一节打开新的一段对话。 `dm:@name` 中的 `name` 就是 agent 在消息头 `@` 位置上看见的那一段。该位置由 `resources/command/sender.tpl` 渲染：有 handle 时是 handle，没有时是 `sender_id`；显示名只出现在括号里， @@ -80,6 +79,41 @@ 看到并且可以照抄的那一段（`resources/command/sender.tpl` 在没有 handle 时渲染 `@`）， 因此按 `messages.sender_id` 匹配这一级足以覆盖飞书，不需要为飞书补 handle。 +## 落库在投递之后 + +出站消息在拿到 provider 的答复之后才落库，一次写入即终态：投递之前不存在 `pending` 行。代价是进程在 +发出与写入之间挂掉会留下「对方收到了、本地没有」的窗口；换来的是本地不会记下一件没有发生的事，也不必 +把先落库的猜测再改正。 + +落库的条件只有一条，对所有会话一视同仁：**这条消息有一部分到了对方那里**——终态是 +`sent`/`queued`/`partial`，或者回执里带回了 provider 给的消息 id 或会话 id。没到就只记审计后返回： +消息历史是这段对话本身，一条没送出去的东西不属于它；失败的经过在审计里查得到。 + +由此一条命令只尝试一次：库里有它的出站消息，就说明它已经到过对方那里，不再重发；没到过的可以重试， +那正是重试的意义。在途的 command_id 记在内存里，写库后移除。 + +这条顺序对「打开一个尚不存在的私聊」尤其要紧。这样的会话没有 provider 身份可写——`@username` 是投递 +地址，不是身份；把它写进会话行会让「发出去的」和「回来的」落在两行上，同一段对话被劈成两半。因此： + +- `dm:@name` 解析不到且该发送者说过话时，取 channel 给出的 `DmAddress`（含 `provider_thread_id` 与 + `delivery_handle`），**先把消息发出去**。 +- `sendMessage` 成功时返回发出的那条 Message，其中必然带 `chat`，因此回执附带 provider 自己认定的会话 + id，与本次用什么形式寻址无关。`ChannelDeliveryReceipt` 与 `OutboundDeliveryResult` 携带 + `provider_thread_id`，由 channel 从回执中取出并按自己的身份串格式拼出。 +- 投递之后按该 id 建立 `channel_sessions` 与 `threads`，再把这条出站消息连同终态一并写入。回执没有给出 + id 时退回 `DmAddress` 给的那个。 +- **一部分都没到就什么都不写**：一段没被打开的对话不是对话，写下来只会让 `dm:@name` 解析到一个发不出去 + 的会话，之后每次重试都走普通路径、不带打开它所需的名字。 +- 找到的会话若没有 thread，说明上一次只写了一半，按尚未打开处理。 +- 这条路不经过草稿与新鲜度校验：一段尚不存在的对话没有未读可言。 + +会话行的 `created_at_ms` 不参与身份比较：同一段对话可能在投递在途时已被另一条路打开，此时该以先写下的 +那一刻为准，而不是把两次的时刻之差当成身份变化。 + +会话行上不保存任何投递用的名字，也就没有「用完要清」这回事。`delivery_handle` 只存在于 `DmAddress` +与 `ChannelSendRequest`，随这一次发送流转。telegram 出站在有该值时用 `@handle` 作为 `chat_id`，否则用 +身份中的数字 id；打字状态的路由仍按数字身份注册。 + ## Tasks ### Task 1：按 `@` 位置的取值查历史发送者 @@ -98,14 +132,15 @@ 两个 id 由各 channel 用**自己的身份串**做 uuid5 得出，core 无法从 `provider_thread_id` 推导。若由 core 自行编号，等对方之后真的发消息进来，channel 会算出另一组 id，同一个人会出现两条会话。 -`sender_kind` 供 Telegram 区分 bot 与人，其余 channel 忽略。该参数不进入 agent 的视野：agent 始终只写 -`dm:@name`，人与 bot 的差别只存在于内部解析。 +`sender_kind` 用于排除来路不明的发送者：以频道或匿名管理员身份发到群里的消息带的是该群的 id 而没有 +`from`，按它建立的私聊会把内容发回原群。它同时决定 `delivery_handle`——需不需要一个名字才能打开这段 +对话、什么样的名字算数，只有 channel 知道。该参数不进入 agent 的视野：agent 始终只写 `dm:@name`。 ### Task 3：未命中时建立映射 -在 `command.py` 的 `send` 解析点接入 Task 1 与 Task 2，按设计一节的流程建立 `channel_sessions` 与 -`threads` 两行后重新解析。新建的 dm 行写入 `target_handle` 与 `target_handle_key`，使后续解析直接命中 -第一步。 +在 `command.py` 的 `send` 解析点接入 Task 1 与 Task 2，按设计一节的流程发送并在其后建立 +`channel_sessions` 与 `threads` 两行。新建的 dm 行写入 `target_handle` 与 `target_handle_key`，使后续 +解析直接命中第一步。 只接 `send`，不接 `unfollow`：为一个尚不存在的会话建立映射只为了取消关注没有意义。 @@ -113,7 +148,21 @@ 冷启动私聊的目标会话必然零入站，与该校验直接冲突。该校验没有任何测试覆盖，也不来自任何既定要求， 因此删除；配套的 `ErrorKind.TARGET_NOT_REPLYABLE` 随之成为死代码，一并删除。 -### Task 4：端到端测试 +### Task 4：迁移既有的按 handle 命名的会话 + +按 handle 命名的 dm 行的正身可从入站消息推出：入站同时带着对方说话时用的 handle 与其 `sender_id`， +取该 handle 下 seq 最大的一条，与解析时的取法一致。据此： + +- 对方已经有按其 id 命名的会话时，把按 handle 命名的那行的 messages、thread、reminder 与 cursor + 并入该会话后删除该行。消息上的 `provider_thread_id` 不改写——它记录的是该条消息当时如何被寻址； + `target` 改写为留存会话的 `canonical_target`。 +- 对方还没有这样的会话时，就地把该行改名为按 id 命名。 + +cursor 合并取「两边未读中最小的 seq 减一」：`delivered_through_seq` 比较的是全局 seq，取两者较大值会把 +另一侧尚未读到的消息判为已送达。配对只在同一 channel 内进行，provider 身份的定义是 +`(agent_id, channel, provider_thread_id)`。 + +### Task 5：端到端测试 用 TestChannel 注入一条群消息，再以该发送者的 handle 执行 `bcc message send --target dm:@name`， 验证映射被建立且消息送达。TestChannel 需要具备解析能力以覆盖成功路径，并能返回 `None` 以覆盖「找不到」路径。 diff --git a/src/bazaar_compute_node/contrib/sqlite/codec.py b/src/bazaar_compute_node/contrib/sqlite/codec.py index e0d74287..d14cc6ce 100644 --- a/src/bazaar_compute_node/contrib/sqlite/codec.py +++ b/src/bazaar_compute_node/contrib/sqlite/codec.py @@ -294,9 +294,13 @@ def validate_outbound_message_input(message: object) -> None: def validate_outbound_insert(message: Message[OutboundAttachment]) -> None: - if message.delivery_state is not OutboundDeliveryState.PENDING: - raise ValueError("a new outbound message must start in pending state") - if any( + """Check a message written once the provider has already answered. + + An outbound is recorded after its attempt, so it arrives carrying whatever + became of it; only a message still waiting may claim to know nothing. + """ + + if message.delivery_state is OutboundDeliveryState.PENDING and any( value is not None for value in ( message.provider_message_id, @@ -442,7 +446,6 @@ def validate_channel_session_update( existing.channel != incoming.channel or existing.provider_thread_id != incoming.provider_thread_id or existing.target_kind is not incoming.target_kind - or existing.created_at_ms != incoming.created_at_ms ): raise ValueError("channel session identity cannot change") _validate_updated_at(existing.updated_at_ms, incoming.updated_at_ms) diff --git a/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py b/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py index e3004c14..4698ab77 100644 --- a/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py +++ b/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py @@ -37,6 +37,7 @@ from .v24_persist_sender_display_name import SENDER_DISPLAY_NAME_MIGRATION from .v25_rename_sessions_to_threads import THREAD_RENAME_MIGRATION from .v26_remove_handoff_messages import HANDOFF_MESSAGE_REMOVAL_MIGRATION +from .v27_name_dms_by_the_peer_id import NAME_DM_BY_PEER_ID_MIGRATION if TYPE_CHECKING: from ..executor import SqliteSession @@ -85,6 +86,7 @@ def _migration_ledger(*migrations: Migration) -> tuple[Migration, ...]: SENDER_DISPLAY_NAME_MIGRATION, THREAD_RENAME_MIGRATION, HANDOFF_MESSAGE_REMOVAL_MIGRATION, + NAME_DM_BY_PEER_ID_MIGRATION, ) diff --git a/src/bazaar_compute_node/contrib/sqlite/migrations/v27_name_dms_by_the_peer_id.py b/src/bazaar_compute_node/contrib/sqlite/migrations/v27_name_dms_by_the_peer_id.py new file mode 100644 index 00000000..e9eebc5f --- /dev/null +++ b/src/bazaar_compute_node/contrib/sqlite/migrations/v27_name_dms_by_the_peer_id.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +from .model import Migration + +# A DM opened by handle used to keep that handle as its name, while the peer +# speaks under its own id, so one conversation could end up stored as two. The +# peer's id is recoverable from what it said: an inbound message carries both +# the handle it spoke under and the id it spoke from. A handle can have been +# worn by more than one peer, and the one that opened this conversation is the +# one that spoke last under it. +_RENAMED = """ + SELECT + opened.agent_id AS agent_id, + opened.channel AS channel, + opened.id AS opened_id, + REPLACE( + opened.provider_thread_id, + ':@' || opened.target_handle || ':', + ':' || spoken.sender_id || ':' + ) AS peer_thread_id + FROM channel_sessions AS opened + JOIN messages AS spoken + ON spoken.agent_id = opened.agent_id + AND spoken.channel = opened.channel + AND spoken.direction = 'inbound' + AND spoken.sender_id IS NOT NULL + AND LOWER(spoken.sender) = opened.target_handle_key + AND spoken.seq = ( + SELECT MAX(latest.seq) FROM messages AS latest + WHERE latest.agent_id = opened.agent_id + AND latest.channel = opened.channel + AND latest.direction = 'inbound' + AND latest.sender_id IS NOT NULL + AND LOWER(latest.sender) = opened.target_handle_key + ) + WHERE opened.target_kind = 'dm' + AND opened.target_handle IS NOT NULL + AND INSTR(opened.provider_thread_id, ':@' || opened.target_handle || ':') > 0 +""" + +# Where the peer already has a conversation under its own id, that one is the +# conversation, and what was written under the handle joins it. +_MERGED = f""" + SELECT + renamed.agent_id AS agent_id, + renamed.opened_id AS opened_id, + opened_thread.id AS opened_thread_id, + peer.id AS peer_id, + peer_thread.id AS peer_thread_id, + 'dm:' || peer.id AS peer_target + FROM ({_RENAMED}) AS renamed + JOIN channel_sessions AS peer + ON peer.agent_id = renamed.agent_id + AND peer.channel = renamed.channel + AND peer.provider_thread_id = renamed.peer_thread_id + JOIN threads AS opened_thread + ON opened_thread.agent_id = renamed.agent_id + AND opened_thread.channel_session_id = renamed.opened_id + JOIN threads AS peer_thread + ON peer_thread.agent_id = renamed.agent_id + AND peer_thread.channel_session_id = peer.id +""" + +# A cursor says what has been delivered, so the merged one may not pass a +# message that either side still had waiting. +_UNREAD_FLOOR = """ + SELECT MIN(waiting.seq) FROM messages AS waiting + JOIN consumer_cursors AS reading + ON reading.thread_id = waiting.thread_id + WHERE waiting.thread_id IN (merged.opened_thread_id, merged.peer_thread_id) + AND waiting.direction = 'inbound' + AND waiting.seq > reading.delivered_through_seq +""" + +NAME_DM_BY_PEER_ID_MIGRATION = Migration( + version=27, + name="name_dms_by_the_peer_id", + statements=( + # the rows that identify this work are the rows it rewrites, so settle + # the set before touching any of them + f""" + CREATE TEMPORARY TABLE bcn_merged_dms AS {_MERGED} + """, + f""" + CREATE TEMPORARY TABLE bcn_renamed_dms AS + SELECT * FROM ({_RENAMED}) AS renamed + WHERE renamed.opened_id NOT IN (SELECT opened_id FROM bcn_merged_dms) + """, + f""" + UPDATE consumer_cursors SET delivered_through_seq = COALESCE( + ( + SELECT ({_UNREAD_FLOOR}) - 1 FROM bcn_merged_dms AS merged + WHERE merged.peer_thread_id = consumer_cursors.thread_id + ), + MAX( + delivered_through_seq, + COALESCE(( + SELECT waiting.delivered_through_seq + FROM consumer_cursors AS waiting + JOIN bcn_merged_dms AS merged + ON merged.opened_thread_id = waiting.thread_id + WHERE merged.peer_thread_id = consumer_cursors.thread_id + ), 0) + ) + ) + WHERE thread_id IN (SELECT peer_thread_id FROM bcn_merged_dms) + """, + """ + DELETE FROM consumer_cursors + WHERE thread_id IN (SELECT opened_thread_id FROM bcn_merged_dms) + """, + """ + UPDATE reminders SET owner_thread_id = ( + SELECT peer_thread_id FROM bcn_merged_dms + WHERE bcn_merged_dms.opened_thread_id = reminders.owner_thread_id + ) + WHERE owner_thread_id IN (SELECT opened_thread_id FROM bcn_merged_dms) + """, + # provider_thread_id stays as it was on a message: it records how that + # message was addressed, which no later merge changes + """ + UPDATE messages SET + thread_id = ( + SELECT peer_thread_id FROM bcn_merged_dms + WHERE bcn_merged_dms.opened_id = messages.channel_session_id + ), + target = ( + SELECT peer_target FROM bcn_merged_dms + WHERE bcn_merged_dms.opened_id = messages.channel_session_id + ), + channel_session_id = ( + SELECT peer_id FROM bcn_merged_dms + WHERE bcn_merged_dms.opened_id = messages.channel_session_id + ) + WHERE channel_session_id IN (SELECT opened_id FROM bcn_merged_dms) + """, + """ + DELETE FROM threads WHERE id IN (SELECT opened_thread_id FROM bcn_merged_dms) + """, + """ + DELETE FROM channel_sessions WHERE id IN (SELECT opened_id FROM bcn_merged_dms) + """, + # the rest have no second half yet, so they only need their own name + """ + UPDATE channel_sessions SET provider_thread_id = ( + SELECT peer_thread_id FROM bcn_renamed_dms + WHERE bcn_renamed_dms.opened_id = channel_sessions.id + ) + WHERE id IN (SELECT opened_id FROM bcn_renamed_dms) + """, + """ + DROP TABLE bcn_merged_dms + """, + """ + DROP TABLE bcn_renamed_dms + """, + ), +) + +__all__ = ["NAME_DM_BY_PEER_ID_MIGRATION"] diff --git a/src/bazaar_compute_node/contrib/sqlite/repository/messages.py b/src/bazaar_compute_node/contrib/sqlite/repository/messages.py index b14837df..8480d2ac 100644 --- a/src/bazaar_compute_node/contrib/sqlite/repository/messages.py +++ b/src/bazaar_compute_node/contrib/sqlite/repository/messages.py @@ -626,6 +626,16 @@ async def _available_message_id(self, message_id: str) -> str: raise ValueError("Agent-scoped message id is already in use") return message_id + async def has_outbound_for_command(self, command_id: str) -> bool: + return ( + await self.fetchone( + "SELECT 1 FROM messages WHERE agent_id = /*agent_id*/? " + "AND command_id = ? AND direction = 'outbound'", + (command_id,), + ) + is not None + ) + async def _resolve_reply( self, canonical: Message[InboundAttachment] ) -> Message[InboundAttachment]: @@ -949,7 +959,7 @@ async def _insert_outbound( canonical = replace( message, - message_id=str(uuid7()), + message_id=message.message_id or str(uuid7()), seq=await self._next_message_seq(), channel=channel_session.channel, provider_thread_id=channel_session.provider_thread_id, diff --git a/src/bazaar_compute_node/contrib/sqlite/storage.py b/src/bazaar_compute_node/contrib/sqlite/storage.py index 7e00eb46..b5c7b22a 100644 --- a/src/bazaar_compute_node/contrib/sqlite/storage.py +++ b/src/bazaar_compute_node/contrib/sqlite/storage.py @@ -14,6 +14,7 @@ "find_thread", "find_channel_session", "find_known_sender", + "has_outbound_for_command", "find_message", "get_thread", "get_channel_session", diff --git a/src/bazaar_compute_node/contrib/telegram/channel.py b/src/bazaar_compute_node/contrib/telegram/channel.py index 0b2a932c..6b562717 100644 --- a/src/bazaar_compute_node/contrib/telegram/channel.py +++ b/src/bazaar_compute_node/contrib/telegram/channel.py @@ -446,29 +446,26 @@ def dm_address( bot_id = self._bot_id if bot_id is None: return None - chat_id: int | str - # sendMessage accepts an @username only for a bot, supergroup or - # channel; a person is still addressable by numeric id alone. - if sender_kind is SenderKind.AGENT and sender.name is not None: - chat_id = f"@{sender.name}" # A message a channel or an anonymous admin posted to a group carries # that chat's id and no `from`, so its kind is unknown; addressing it # would publish the DM back into the group it came from. - elif ( - sender_kind in {SenderKind.HUMAN, SenderKind.AGENT} - and sender.id is not None - ): - try: - chat_id = int(sender.id) - except ValueError: - return None - else: + if sender_kind not in {SenderKind.HUMAN, SenderKind.AGENT} or sender.id is None: + return None + # The peer's own id is what its messages arrive under, so it is the one + # name this conversation can keep; a username only opens a chat that + # does not exist yet. + try: + chat_id = int(sender.id) + except ValueError: return None identity = TelegramThreadIdentity(bot_id=bot_id, chat_id=chat_id, topic_id=0) return DmAddress( channel_session_id=identity.channel_session_id, thread_id=identity.session_id, provider_thread_id=identity.provider_thread_id, + # `sendMessage` reaches a bot this node has never spoken to only by + # username; a person is not addressable that way at all + delivery_handle=sender.name if sender_kind is SenderKind.AGENT else None, ) async def send( diff --git a/src/bazaar_compute_node/contrib/telegram/outbound.py b/src/bazaar_compute_node/contrib/telegram/outbound.py index a9f02548..cab21edf 100644 --- a/src/bazaar_compute_node/contrib/telegram/outbound.py +++ b/src/bazaar_compute_node/contrib/telegram/outbound.py @@ -4,7 +4,7 @@ import math import re from collections.abc import Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from ...core.channel import ChannelContext, ChannelDeliveryReceipt, ChannelSendRequest from ...core.outcomes import ProviderCallResult, ProviderCallStatus @@ -39,12 +39,18 @@ def receipt(self) -> Mapping[str, object]: if receipt.get("state") == "confirmed" and isinstance(receipt.get("provider_message_id"), str) ) + thread_ids = tuple( + value + for receipt in self.receipts + if isinstance(value := receipt.get("provider_thread_id"), str) + ) return { "total_parts": self.total, "confirmed_parts": self.confirmed, "parts": tuple(self.receipts), "provider_message_id": confirmed_ids[0] if confirmed_ids else None, "provider_receipt_ref": confirmed_ids[-1] if confirmed_ids else None, + "provider_thread_id": thread_ids[0] if thread_ids else None, } @@ -109,7 +115,11 @@ def _outbound_route( "invalid_route", "Telegram outbound route belongs to another bot", ) + # typing follows the conversation, which its own id names; only the + # send itself may need the name that opens a chat self._stream_routes[request.session_id] = identity + if request.delivery_handle is not None: + identity = replace(identity, chat_id=f"@{request.delivery_handle}") reply_to_message_id: int | None = None if request.provider_reply_to_message_id is not None: @@ -325,6 +335,9 @@ async def _send_text_parts( "fallback_from": fallback_from, "state": "confirmed", "provider_message_id": provider_message_id, + "provider_thread_id": self._acknowledged_thread_id( + provider_message, identity + ), } ) delivery.confirmed += 1 @@ -423,7 +436,13 @@ async def _send_documents( delivery.receipts.append( receipt_base - | {"state": "confirmed", "provider_message_id": provider_message_id} + | { + "state": "confirmed", + "provider_message_id": provider_message_id, + "provider_thread_id": self._acknowledged_thread_id( + provider_message, identity + ), + } ) delivery.confirmed += 1 self._outbound_parts_confirmed += 1 @@ -660,6 +679,24 @@ def _outbound_provider_message_id(message: Mapping[str, object]) -> str | None: return None return str(provider_message_id) + @staticmethod + def _acknowledged_thread_id( + message: Mapping[str, object], identity: TelegramThreadIdentity + ) -> str | None: + """The conversation Telegram says this message landed in. + + A chat opened by `@username` answers under its own numeric id, and the + send acknowledgement is where that id first appears. + """ + + chat = message.get("chat") + if not isinstance(chat, Mapping): + return None + chat_id = chat.get("id") + if not isinstance(chat_id, int) or isinstance(chat_id, bool) or chat_id == 0: + return None + return replace(identity, chat_id=chat_id).provider_thread_id + @staticmethod def _channel_receipt(receipts: list[dict[str, object]]) -> ChannelDeliveryReceipt: confirmed_ids: list[str] = [] @@ -673,11 +710,17 @@ def _channel_receipt(receipts: list[dict[str, object]]) -> ChannelDeliveryReceip raise AssertionError( "confirmed Telegram delivery requires provider message id" ) + thread_ids = tuple( + value + for receipt in receipts + if isinstance(value := receipt.get("provider_thread_id"), str) + ) return ChannelDeliveryReceipt( provider_message_id=confirmed_ids[0], provider_receipt_ref=( confirmed_ids[-1] if len(confirmed_ids) > 1 else None ), + provider_thread_id=thread_ids[0] if thread_ids else None, ) diff --git a/src/bazaar_compute_node/core/channel.py b/src/bazaar_compute_node/core/channel.py index 82bd1f3c..a5f4bd37 100644 --- a/src/bazaar_compute_node/core/channel.py +++ b/src/bazaar_compute_node/core/channel.py @@ -51,6 +51,9 @@ class ChannelDeliveryReceipt: provider_message_id: str | None = None provider_receipt_ref: str | None = None + # the conversation the provider says this landed in, which is the only + # authority on a chat that had to be opened by name + provider_thread_id: str | None = None def __post_init__(self) -> None: if self.provider_message_id is None and self.provider_receipt_ref is None: @@ -69,6 +72,7 @@ class ChannelSendRequest: target_kind: ChannelTargetKind provider_thread_id: str provider_reply_to_message_id: str | None = None + delivery_handle: str | None = None @dataclass(frozen=True, slots=True) @@ -129,11 +133,16 @@ class DmAddress: Both ids come from the channel's own identity string rather than from the provider thread id, so a later inbound message from the same peer lands on this conversation instead of creating a second one. + + `delivery_handle` is what it takes to open a chat this node has never + spoken in, where the channel cannot reach it by id yet. Only the channel + knows whether such a name is needed and which one counts. """ channel_session_id: str thread_id: str provider_thread_id: str + delivery_handle: str | None = None class IChannel(IAsyncLifecycle, IApproval, Protocol): @@ -258,6 +267,7 @@ def dm_address( ), thread_id=thread_id, provider_thread_id=address.provider_thread_id, + delivery_handle=address.delivery_handle, ) async def send( diff --git a/src/bazaar_compute_node/core/orchestration/command.py b/src/bazaar_compute_node/core/orchestration/command.py index 86b1132f..e95218b5 100644 --- a/src/bazaar_compute_node/core/orchestration/command.py +++ b/src/bazaar_compute_node/core/orchestration/command.py @@ -6,13 +6,14 @@ import mimetypes import os import stat -from collections.abc import Callable, Mapping -from dataclasses import replace +from collections.abc import Callable +from dataclasses import dataclass, replace from pathlib import Path, PurePosixPath +from uuid import uuid7 from ..actor import Actor, Actors, Agent, Thread from ..audit import AuditRecorder, ErrorKind -from ..channel import ChannelSendRequest, IChannel +from ..channel import ChannelSendRequest, DmAddress, IChannel from ..command import ( ICommandService, InboxListResult, @@ -120,6 +121,57 @@ def __call__( return tuple(attachments) +_DELIVERED_STATES = frozenset( + { + OutboundDeliveryState.SENT, + OutboundDeliveryState.QUEUED, + OutboundDeliveryState.PARTIAL, + } +) + + +def _answered( + outbound: Message[OutboundAttachment], + delivery_result: OutboundDeliveryResult, + *, + at_ms: int, +) -> Message[OutboundAttachment]: + """Fold what the channel made of a message back into it.""" + + outbound = outbound.transition_to( + delivery_result.state, + at_ms=at_ms, + provider_message_id=delivery_result.provider_message_id, + provider_receipt_ref=delivery_result.provider_receipt_ref, + error_kind=delivery_result.error_kind, + error_message=delivery_result.error_message, + ) + if not delivery_result.receipt: + return outbound + return replace( + outbound, + metadata={ + **outbound.metadata, + "delivery_receipt": dict(delivery_result.receipt), + }, + ) + + +def _reached_the_peer(delivery_result: OutboundDeliveryResult) -> bool: + """Say whether any of this message is with the peer. + + A message that never left is not part of the conversation, and writing it + down would put words in a history that never carried them. An attempt whose + outcome is unknown still counts when the provider named a part it took. + """ + + return ( + delivery_result.state in _DELIVERED_STATES + or delivery_result.provider_message_id is not None + or delivery_result.provider_thread_id is not None + ) + + _DELIVERY_OUTCOMES: dict[ OutboundDeliveryState, tuple[ErrorKind | None, RuntimeEventState] ] = { @@ -133,6 +185,15 @@ def __call__( } +@dataclass(frozen=True, slots=True) +class _DmOpening: + """Where a DM that has never been held would be sent, and what to call it.""" + + address: DmAddress + channel: str + handle: str + + class CommandService(ICommandService): """Execute session-scoped check, read, and send commands.""" @@ -157,6 +218,9 @@ def __init__( self._attachment_resolver = OutboundAttachmentResolver(workspace) self._clock = clock self._drafts: dict[str, MessageDraft] = {} + # an outbound is written down once the provider answers, so until then + # only this holds a command to its one attempt + self._sending: set[str] = set() self._freshness_snapshots: dict[str, int] = {} self._logger = logging.getLogger("bazaar_compute_node.orchestration.command") @@ -232,76 +296,134 @@ async def read( ) return result - async def _resolve_or_mint( - self, actor: Actor, raw_target: str - ) -> ResolvedInboxTarget: - """Resolve a target, minting a DM conversation for a known sender. + async def _dm_opening(self, actor: Actor, raw_target: str) -> _DmOpening | None: + """Find where a DM this node has never held would be sent. - Only `dm:@` is minted, and only from a sender this Agent has already + Only `dm:@` opens, and only towards a sender this Agent has already heard from: the address comes from that past message, never from a directory lookup. A conversation-scoped actor never reaches a conversation it did not already own, so it is refused before anything - is written. Anything else stays a resolution failure. + is sent. `None` means the target stays unresolvable. """ - try: - return await self._storage.resolve_inbox_target(raw_target) - except AmbiguousInboxTargetError: - # Several conversations answer to this handle. Minting a new one - # would silently pick a peer for the caller. - raise - except InboxTargetResolutionError: - if ( - not isinstance(actor, Agent) - or not raw_target.startswith("dm:@") - or len(raw_target) == 4 - ): - raise - known = await self._storage.find_known_sender(raw_target[4:]) - if known is None: - raise - address = self._channel.dm_address( - known.sender, sender_kind=known.sender_kind + if ( + not isinstance(actor, Agent) + or not raw_target.startswith("dm:@") + or len(raw_target) == 4 + ): + return None + known = await self._storage.find_known_sender(raw_target[4:]) + if known is None: + return None + address = self._channel.dm_address(known.sender, sender_kind=known.sender_kind) + if address is None: + return None + # The name the provider calls the peer comes first, then the token that + # found it. + handle = known.sender.name or raw_target[4:] + return _DmOpening(address=address, channel=known.channel, handle=handle) + + async def _open_dm( + self, + *, + command_id: str, + raw_target: str, + opening: _DmOpening, + body: str, + attachments: tuple[OutboundAttachment, ...], + created_at_ms: int, + ) -> MessageSendSuccess: + """Send into a conversation that does not exist yet, then write it down. + + The provider names the conversation it delivered into, and a chat opened + by a name it answers to is not reachable by its id until then. Writing + afterwards is what keeps this message and every later one in the same + conversation. + """ + + address = opening.address + attempted_at_ms = self._clock() + delivery_result = await self._delivery.deliver( + ChannelSendRequest( + session_id=address.thread_id, + body=body, + attachments=attachments, + target_kind=ChannelTargetKind.DM, + provider_thread_id=address.provider_thread_id, + delivery_handle=address.delivery_handle, ) - if address is None: - raise + ) now = self._clock() - stored_session = await self._storage.get_channel_session( - address.channel_session_id + session = ChannelSession( + id=address.channel_session_id, + channel=opening.channel, + provider_thread_id=( + delivery_result.provider_thread_id or address.provider_thread_id + ), + created_at_ms=now, + updated_at_ms=now, + target_kind=ChannelTargetKind.DM, + target_handle=opening.handle, + target_handle_key=opening.handle.casefold(), ) - if stored_session is None: - stored_session = ChannelSession( - id=address.channel_session_id, - channel=known.channel, - provider_thread_id=address.provider_thread_id, - created_at_ms=now, - updated_at_ms=now, - target_kind=ChannelTargetKind.DM, - ) - # The name this conversation already answers to comes first, then what - # the provider calls the peer, and only then the token that found it. - handle = stored_session.target_handle or known.sender.name or raw_target[4:] - await self._storage.save_channel_session( - replace( - stored_session, - updated_at_ms=now, - target_handle=handle, - target_handle_key=handle.casefold(), - ) + outbound = Message[OutboundAttachment]( + direction=MessageDirection.OUTBOUND, + seq=0, + message_id=str(uuid7()), + command_id=command_id, + thread_id=address.thread_id, + channel_session_id=session.id, + target=session.canonical_target, + body=body, + attachments=attachments, + target_kind=ChannelTargetKind.DM, + delivery_state=OutboundDeliveryState.PENDING, + created_at_ms=created_at_ms, + provider_attempted_at_ms=attempted_at_ms, + ) + outbound = _answered(outbound, delivery_result, at_ms=self._clock()) + delivery_state = outbound.delivery_state + if delivery_state is None: + raise RuntimeError("outbound message has no delivery state") + audit_context = self._correlation( + thread_id=address.thread_id, + channel=session.channel, + channel_session_id=session.id, + command_id=command_id, + outbound_message_id=outbound.message_id, ) - stored_thread = await self._storage.get_thread(address.thread_id) - if stored_thread is None: - stored_thread = ConversationRow( + # a chat that was never opened is not a conversation, and writing one + # down would leave `dm:@name` resolving to something that cannot be + # spoken to; what became of the attempt is still worth recording + if not _reached_the_peer(delivery_result): + await self._record_delivery( + audit_context, + outbound, + delivery_result, + command_id=command_id, + canonical_target=raw_target, + ) + return MessageSendSuccess(message=outbound, target=raw_target) + await self._storage.save_channel_session(session) + await self._storage.save_thread( + ConversationRow( id=address.thread_id, - channel_session_id=address.channel_session_id, + channel_session_id=session.id, workspace_id=self._actors.agent_id, created_at_ms=now, updated_at_ms=now, ) - await self._storage.save_thread(replace(stored_thread, updated_at_ms=now)) - # Ask for the conversation we just wrote rather than for the token that - # found it: the handle it answers to is the provider's, not that token. - return await self._storage.resolve_inbox_target(stored_session.canonical_target) + ) + outbound = await self._storage.finalize_outbound_delivery(outbound) + await self._record_delivery( + audit_context, + outbound, + delivery_result, + command_id=command_id, + canonical_target=session.canonical_target, + ) + resolved = await self._storage.resolve_inbox_target(session.canonical_target) + return MessageSendSuccess(message=outbound, target=resolved.display_target) def _require_in_reach( self, @@ -405,42 +527,30 @@ async def _transmit( provider_reply_to_message_id=prepared.reply_to_provider_message_id, ) ) - outbound = replace( - outbound, - provider_attempted_at_ms=outbound.provider_attempted_at_ms or self._clock(), - ) - outbound = outbound.transition_to( - delivery_result.state, - at_ms=self._clock(), - provider_message_id=delivery_result.provider_message_id, - provider_receipt_ref=delivery_result.provider_receipt_ref, - error_kind=delivery_result.error_kind, - error_message=delivery_result.error_message, - ) - if delivery_result.receipt: - outbound = replace( - outbound, - metadata={ - **outbound.metadata, - "delivery_receipt": dict(delivery_result.receipt), - }, - ) - return outbound, delivery_result + return _answered( + outbound, delivery_result, at_ms=self._clock() + ), delivery_result async def _record_delivery( self, audit_context: CorrelationContext, + outbound: Message[OutboundAttachment], + delivery_result: OutboundDeliveryResult, *, command_id: str, canonical_target: str, - delivery_state: OutboundDeliveryState, - error_message: str | None, - receipt: Mapping[str, object] | None, - terminal_kind: ErrorKind | None, - terminal_state: RuntimeEventState, ) -> None: """Write down what the channel did with the message, twice over.""" + delivery_state = outbound.delivery_state + if delivery_state is None: + raise RuntimeError("outbound message has no delivery state") + error_message = outbound.error_message + receipt = delivery_result.receipt + terminal_kind, terminal_state = _DELIVERY_OUTCOMES.get( + delivery_result.state, + (ErrorKind.PROVIDER_UNKNOWN, RuntimeEventState.UNKNOWN), + ) await self._audit.append( event_name=f"channel.outbound.{delivery_state.value}", state=terminal_state, @@ -511,14 +621,11 @@ async def _deliver( correlation=audit_context, ) outbound, delivery_result = await self._transmit(outbound, prepared) - terminal_kind, terminal_state = _DELIVERY_OUTCOMES.get( - delivery_result.state, - (ErrorKind.PROVIDER_UNKNOWN, RuntimeEventState.UNKNOWN), - ) - outbound = await self._storage.finalize_outbound_delivery(outbound) delivery_state = outbound.delivery_state if delivery_state is None: raise RuntimeError("outbound message has no delivery state") + if _reached_the_peer(delivery_result): + outbound = await self._storage.finalize_outbound_delivery(outbound) if ( delivery_state in { @@ -530,13 +637,10 @@ async def _deliver( self._drafts.pop(target_id, None) await self._record_delivery( audit_context, + outbound, + delivery_result, command_id=command_id, canonical_target=canonical_target, - delivery_state=delivery_state, - error_message=outbound.error_message, - receipt=delivery_result.receipt, - terminal_kind=terminal_kind, - terminal_state=terminal_state, ) return outbound @@ -563,7 +667,67 @@ async def send( ) if not send_draft and not body.strip() and not attachments: raise ValueError("outbound message must not be empty") - target = await self._resolve_or_mint(actor, raw_target) + if command_id in self._sending: + raise ValueError(f"command was already sent: {command_id}") + self._sending.add(command_id) + try: + if await self._storage.has_outbound_for_command(command_id): + raise ValueError(f"command was already sent: {command_id}") + return await self._send( + actor=actor, + command_id=command_id, + raw_target=raw_target, + body=body, + created_at_ms=created_at_ms, + attachments=attachments, + reply_to_message_id=reply_to_message_id, + send_draft=send_draft, + ) + finally: + self._sending.discard(command_id) + + async def _send( + self, + *, + actor: Actor, + command_id: str, + raw_target: str, + body: str, + created_at_ms: int, + attachments: tuple[OutboundAttachment, ...], + reply_to_message_id: str | None, + send_draft: bool, + ) -> MessageSendResult: + try: + target = await self._storage.resolve_inbox_target(raw_target) + except AmbiguousInboxTargetError: + # Several conversations answer to this handle. Opening another one + # would silently pick a peer for the caller. + raise + except InboxTargetResolutionError: + opening = await self._dm_opening(actor, raw_target) + if opening is None or send_draft: + raise + # the conversation may be open already under a name this token does + # not answer to, and then there is nothing to open + held = await self._storage.find_channel_session( + channel=opening.channel, + provider_thread_id=opening.address.provider_thread_id, + ) + # a conversation whose thread never made it is not open yet, and + # resolving it would fail for good + if held is not None and await self._storage.find_thread(held.id) is None: + held = None + if held is None: + return await self._open_dm( + command_id=command_id, + raw_target=raw_target, + opening=opening, + body=body, + attachments=attachments, + created_at_ms=created_at_ms, + ) + target = await self._storage.resolve_inbox_target(held.canonical_target) self._require_in_reach(actor, target.thread.id, raw_target) staged = await self._stage_draft( diff --git a/src/bazaar_compute_node/core/orchestration/delivery.py b/src/bazaar_compute_node/core/orchestration/delivery.py index 716474c4..be4e54f7 100644 --- a/src/bazaar_compute_node/core/orchestration/delivery.py +++ b/src/bazaar_compute_node/core/orchestration/delivery.py @@ -58,6 +58,7 @@ def _map_provider_result( state=OutboundDeliveryState.SENT, provider_message_id=receipt.provider_message_id, provider_receipt_ref=receipt.provider_receipt_ref, + provider_thread_id=receipt.provider_thread_id, receipt=dict(provider_result.receipt), ) @@ -69,6 +70,7 @@ def _map_provider_result( state=OutboundDeliveryState.QUEUED, provider_message_id=receipt.provider_message_id, provider_receipt_ref=receipt.provider_receipt_ref, + provider_thread_id=receipt.provider_thread_id, receipt=dict(provider_result.receipt), ) @@ -80,6 +82,7 @@ def _map_provider_result( state=OutboundDeliveryState.PARTIAL, provider_message_id=receipt.provider_message_id, provider_receipt_ref=receipt.provider_receipt_ref, + provider_thread_id=receipt.provider_thread_id, error_kind=( provider_result.error_kind or ErrorKind.PROVIDER_PARTIAL.value ), @@ -91,10 +94,20 @@ def _map_provider_result( provider_receipt_ref = provider_result.receipt.get("provider_receipt_ref") if not isinstance(provider_receipt_ref, str) or not provider_receipt_ref: provider_receipt_ref = None + # an attempt that ended badly can still have left part of itself with + # the peer, and the receipt is where that shows + provider_message_id = provider_result.receipt.get("provider_message_id") + if not isinstance(provider_message_id, str) or not provider_message_id: + provider_message_id = None + provider_thread_id = provider_result.receipt.get("provider_thread_id") + if not isinstance(provider_thread_id, str) or not provider_thread_id: + provider_thread_id = None if provider_result.status is ProviderCallStatus.FAILED: return OutboundDeliveryResult( state=OutboundDeliveryState.FAILED, + provider_message_id=provider_message_id, provider_receipt_ref=provider_receipt_ref, + provider_thread_id=provider_thread_id, error_kind=( provider_result.error_kind or ErrorKind.PROVIDER_FAILED.value ), @@ -103,7 +116,9 @@ def _map_provider_result( ) return OutboundDeliveryResult( state=OutboundDeliveryState.UNKNOWN, + provider_message_id=provider_message_id, provider_receipt_ref=provider_receipt_ref, + provider_thread_id=provider_thread_id, error_kind=(provider_result.error_kind or ErrorKind.PROVIDER_UNKNOWN.value), error_message=provider_result.error_message, next_action="reconcile channel delivery before retrying", diff --git a/src/bazaar_compute_node/core/outcomes.py b/src/bazaar_compute_node/core/outcomes.py index e201900b..53cf226e 100644 --- a/src/bazaar_compute_node/core/outcomes.py +++ b/src/bazaar_compute_node/core/outcomes.py @@ -60,6 +60,7 @@ class OutboundDeliveryResult: state: OutboundDeliveryState provider_message_id: str | None = None provider_receipt_ref: str | None = None + provider_thread_id: str | None = None error_kind: str | None = None error_message: str | None = None next_action: str | None = None diff --git a/src/bazaar_compute_node/core/storage.py b/src/bazaar_compute_node/core/storage.py index 4c612d21..007c3480 100644 --- a/src/bazaar_compute_node/core/storage.py +++ b/src/bazaar_compute_node/core/storage.py @@ -3,6 +3,7 @@ from collections.abc import Sequence from dataclasses import dataclass, replace from typing import Any, Protocol +from uuid import uuid7 from .command import ( InboxListResult, @@ -376,23 +377,21 @@ async def materialize_outbound_if_fresh( if reply_message is not None and reply_message.target == payload.target: reply_to_message_id = payload.reply_to_message_id reply_to_provider_message_id = reply_message.provider_message_id - outcome = await self.save_message( - Message( - direction=MessageDirection.OUTBOUND, - seq=0, - message_id=f"outbound-{target_id}-{command_id}", - command_id=command_id, - thread_id=target_id, - channel_session_id=channel_session.id, - target=payload.target, - body=payload.body, - attachments=payload.attachments, - target_kind=channel_session.target_kind, - delivery_state=OutboundDeliveryState.PENDING, - created_at_ms=payload.created_at_ms, - provider_attempted_at_ms=attempted_at_ms, - reply_to_message_id=reply_to_message_id, - ) + outcome = Message( + direction=MessageDirection.OUTBOUND, + seq=0, + message_id=str(uuid7()), + command_id=command_id, + thread_id=target_id, + channel_session_id=channel_session.id, + target=payload.target, + body=payload.body, + attachments=payload.attachments, + target_kind=channel_session.target_kind, + delivery_state=OutboundDeliveryState.PENDING, + created_at_ms=payload.created_at_ms, + provider_attempted_at_ms=attempted_at_ms, + reply_to_message_id=reply_to_message_id, ) return MaterializeOutboundResult( channel_session=channel_session, @@ -633,6 +632,14 @@ async def resolve_message( delivery_states: frozenset[OutboundDeliveryState] | None = None, ) -> Message[InboundAttachment | OutboundAttachment] | None: ... + async def has_outbound_for_command(self, command_id: str) -> bool: + """Say whether this command already reached the peer. + + An attempt that never left may be made again; one that did would arrive + twice, and only an attempt that arrived is written down. + """ + ... + async def get_owned_message( self, agent_id: str, diff --git a/tests/contrib/test_orchestration.py b/tests/contrib/test_orchestration.py index 4e79d982..f0e7383a 100644 --- a/tests/contrib/test_orchestration.py +++ b/tests/contrib/test_orchestration.py @@ -4869,6 +4869,100 @@ async def test_a_sender_the_channel_cannot_address_stays_not_found() -> None: await orchestrator.stop(timeout=1) +@pytest.mark.asyncio +async def test_an_opened_dm_answers_to_the_conversation_it_was_delivered_into() -> None: + orchestrator, channel, _, storage, _ = await make_node( + mode=Mode.DANGEROUS_INDIVIDUAL + ) + try: + # the peer answers under an id of its own choosing, which is not the one + # this node would have guessed + channel.delivered_thread_ids["test:dm:peer-1"] = "test:dm:chat-9" + await channel.inject( + Message( + direction=MessageDirection.INBOUND, + seq=1, + message_id="message-group-1", + thread_id="bcn-group", + channel_session_id="channel-group", + channel="test", + provider_thread_id="thread-group", + provider_message_id="provider-group-1", + received_at_ms=1, + sender=SenderIdentity(id="peer-1", name="kana"), + message_type="text", + target="group:channel-group", + target_kind=ChannelTargetKind.GROUP, + body="hello everyone", + metadata={"sender_kind": SenderKind.AGENT.value}, + ) + ) + await wait_until( + lambda: ( + len( + _stored_messages( + storage, "bcn-group", direction=MessageDirection.INBOUND + ) + ) + == 1 + ) + ) + + # This peer has only spoken in a group, so its DM does not exist yet. + await orchestrator.command_service.send( + actor=Agent("workspace-1"), + command_id="command-dm", + raw_target="dm:@kana", + body="hello in private", + created_at_ms=2, + ) + + # the name that opens the chat went out with the message, and the + # conversation was written down under the id the delivery reported + assert any( + sent.delivery_handle == "kana" and sent.body == "hello in private" + for sent in channel.sent_messages + ) + opened = storage.channel_sessions["channel-dm-peer-1"] + assert opened.provider_thread_id == "test:dm:chat-9" + assert opened.target_handle == "kana" + + # so the peer speaking under that id reaches the same conversation + await channel.inject( + Message( + direction=MessageDirection.INBOUND, + seq=2, + message_id="message-dm-1", + thread_id="bcn-dm-inbound", + channel_session_id="channel-dm-inbound", + channel="test", + provider_thread_id="test:dm:chat-9", + provider_message_id="provider-dm-1", + received_at_ms=3, + sender=SenderIdentity(id="peer-1", name="kana"), + message_type="text", + target="dm:channel-dm-inbound", + target_kind=ChannelTargetKind.DM, + body="hello back", + metadata={"sender_kind": SenderKind.AGENT.value}, + ) + ) + await wait_until( + lambda: ( + len( + _stored_messages( + storage, "thread-dm-peer-1", direction=MessageDirection.INBOUND + ) + ) + == 1 + ) + ) + # the group and the one DM, not a second half of it + assert len(storage.channel_sessions) == 2 + finally: + await orchestrator.stop(timeout=1) + + @pytest.mark.asyncio async def test_a_handle_two_conversations_answer_to_stays_an_error() -> None: orchestrator, channel, _, storage, _ = await make_node( diff --git a/tests/contrib/test_sqlite_database.py b/tests/contrib/test_sqlite_database.py index 1b6cf46a..81315eb7 100644 --- a/tests/contrib/test_sqlite_database.py +++ b/tests/contrib/test_sqlite_database.py @@ -691,7 +691,7 @@ async def test_sqlite_bootstrap_binds_agent_scope_without_node_state() -> None: row["name"] for row in migration_columns } assert schema_version is not None - assert schema_version["version"] == 26 + assert schema_version["version"] == 27 assert {row["name"] for row in message_columns}.isdisjoint( {"snapshot_seq", "current_inbound_seq"} ) @@ -1217,7 +1217,162 @@ async def test_sqlite_v26_removes_handoff_messages_and_keeps_the_rest() -> None: "inbound-after-upgrade", ) assert schema_version is not None - assert schema_version["version"] == 26 + assert schema_version["version"] == 27 + finally: + await database.stop(timeout=2) + + +@pytest.mark.asyncio +async def test_sqlite_v27_names_a_dm_by_the_peer_it_belongs_to() -> None: + data_dir = resolve_data_dir() + data_dir.mkdir() + database_path = data_dir / "bcn.sqlite3" + + async with aiosqlite.connect(database_path) as connection: + connection.row_factory = aiosqlite.Row + await connection.create_function("bcn_agent_id", 0, lambda: "agent-1") + await connection.create_function("bcn_agent_name", 0, lambda: "Agent 1") + for migration in MIGRATIONS[:26]: + for statement in migration.statements: + await connection.execute(statement) + await connection.execute( + "INSERT INTO schema_migrations " + "(version, migration_name, checksum, applied_at_ms, duration_ms) " + "VALUES (?, ?, ?, ?, ?)", + (migration.version, migration.name, migration.checksum, 1, 0), + ) + # kana answered, so its DM is stored twice; mika has not answered yet, + # so the one opened by handle is all it has + await connection.executemany( + "INSERT INTO channel_sessions (" + "id, channel, provider_thread_id, target_kind, following, " + "provider_identity_ref_json, target_handle, target_handle_key, " + "created_at_ms, updated_at_ms, agent_id" + ") VALUES (?, 'telegram', ?, ?, 1, '{}', ?, ?, 1, 1, 'agent-1')", + ( + ("channel-group", "telegram:1:-100:0", "group", None, None), + ("channel-kana-handle", "telegram:1:@kana:0", "dm", "kana", "kana"), + ("channel-kana-chat", "telegram:1:7:0", "dm", "kana", "kana"), + ("channel-mika-handle", "telegram:1:@mika:0", "dm", "mika", "mika"), + ), + ) + await connection.executemany( + "INSERT INTO threads (" + "id, channel_session_id, workspace_id, created_at_ms, updated_at_ms, " + "agent_id" + ") VALUES (?, ?, 'agent-1', 1, 1, 'agent-1')", + ( + ("thread-group", "channel-group"), + ("thread-kana-handle", "channel-kana-handle"), + ("thread-kana-chat", "channel-kana-chat"), + ("thread-mika-handle", "channel-mika-handle"), + ), + ) + await connection.executemany( + "INSERT INTO messages (" + "message_id, seq, direction, agent_id, thread_id, channel_session_id, " + "channel, provider_thread_id, provider_message_id, received_at_ms, " + "sender, sender_id, message_type, target, target_kind, body, " + "mentions_agent, notifies_runtime, metadata_json" + ") VALUES (?, ?, 'inbound', 'agent-1', ?, ?, 'telegram', ?, ?, 1, ?, ?, " + "'text', ?, ?, 'hello', 0, 1, '{\"sender_kind\":\"agent\"}')", + ( + ( + "group-from-kana", + 1, + "thread-group", + "channel-group", + "telegram:1:-100:0", + "11", + "kana", + "7", + "group:channel-group", + "group", + ), + ( + "group-from-mika", + 2, + "thread-group", + "channel-group", + "telegram:1:-100:0", + "12", + "mika", + "8", + "group:channel-group", + "group", + ), + ( + "dm-from-kana", + 5, + "thread-kana-chat", + "channel-kana-chat", + "telegram:1:7:0", + "13", + "kana", + "7", + "dm:channel-kana-chat", + "dm", + ), + ), + ) + await connection.execute( + "INSERT INTO messages (" + "message_id, seq, direction, agent_id, thread_id, channel_session_id, " + "channel, provider_thread_id, message_type, target, target_kind, body, " + "command_id, delivery_state, created_at_ms, provider_attempted_at_ms, " + "attachments_json" + ") VALUES ('dm-to-kana', 9, 'outbound', 'agent-1', 'thread-kana-handle', " + "'channel-kana-handle', 'telegram', 'telegram:1:@kana:0', 'text', " + "'dm:channel-kana-handle', 'dm', 'hello in private', 'command-1', 'sent', " + "9, 9, '[]')" + ) + # the half opened by handle has read further than the peer's own thread, + # where seq 5 is still waiting + await connection.executemany( + "INSERT INTO consumer_cursors (" + "thread_id, delivered_through_seq, updated_at_ms" + ") VALUES (?, ?, 1)", + (("thread-kana-handle", 9), ("thread-kana-chat", 4)), + ) + await connection.commit() + + database = SqliteDatabase() + await database.start(timeout=2) + try: + async with database.reader() as session, session.transaction(): + sessions = await session.fetchall( + "SELECT id, provider_thread_id FROM channel_sessions" + ) + threads = await session.fetchall("SELECT id FROM threads") + moved = await session.fetchone( + "SELECT thread_id, channel_session_id, provider_thread_id, target " + "FROM messages WHERE message_id = 'dm-to-kana'" + ) + cursors = await session.fetchall( + "SELECT thread_id, delivered_through_seq FROM consumer_cursors" + ) + + # every DM now answers to the peer it belongs to + named = {row["id"]: row["provider_thread_id"] for row in sessions} + assert named["channel-kana-chat"] == "telegram:1:7:0" + assert named["channel-mika-handle"] == "telegram:1:8:0" + # the half opened by handle is gone, folded into the peer's own + assert "channel-kana-handle" not in named + thread_ids = {row["id"] for row in threads} + assert "thread-kana-chat" in thread_ids + assert "thread-kana-handle" not in thread_ids + assert moved is not None + assert moved["thread_id"] == "thread-kana-chat" + assert moved["channel_session_id"] == "channel-kana-chat" + assert moved["target"] == "dm:channel-kana-chat" + # how this message was addressed is a fact about the message + assert moved["provider_thread_id"] == "telegram:1:@kana:0" + # the message waiting at seq 5 is still waiting + read_through = { + row["thread_id"]: row["delivered_through_seq"] for row in cursors + } + assert read_through["thread-kana-chat"] == 4 + assert "thread-kana-handle" not in read_through finally: await database.stop(timeout=2) @@ -1291,7 +1446,7 @@ async def test_sqlite_v13_migration_preserves_durable_session_and_attempt_facts( "SELECT agent_id FROM runtime_attempts WHERE turn_id = 'turn-1'" ) assert schema_version is not None - assert schema_version["version"] == 26 + assert schema_version["version"] == 27 assert node_state is None assert [row["agent_id"] for row in ownership_rows] == [ "workspace-1", @@ -1400,7 +1555,7 @@ async def test_sqlite_removes_runtime_events_and_node_state() -> None: assert not runtime_objects assert node_state is None assert schema_version is not None - assert schema_version["version"] == 26 + assert schema_version["version"] == 27 assert marker is not None assert marker["compaction_completed_at_ms"] is not None assert freelist is not None diff --git a/tests/contrib/test_telegram_channel.py b/tests/contrib/test_telegram_channel.py index 165056d6..460fffe6 100644 --- a/tests/contrib/test_telegram_channel.py +++ b/tests/contrib/test_telegram_channel.py @@ -244,19 +244,21 @@ def build_api(*args: object, **kwargs: object) -> _FakeApi: ) assert human is not None assert human.provider_thread_id == f"telegram:{bot_id}:{TEST_USER_ID}:0" - # Bots reach each other by username; a numeric id does not apply. + assert human.delivery_handle is None + # A bot is named by its own id too, and reached by username only for as + # long as that chat does not exist. bot = channel.dm_address( SenderIdentity(id="7", name="kana"), sender_kind=SenderKind.AGENT ) assert bot is not None - assert bot.provider_thread_id == f"telegram:{bot_id}:@kana:0" - # A bot without a username falls back to the numeric id rather than - # minting an address Telegram would reject. + assert bot.provider_thread_id == f"telegram:{bot_id}:7:0" + assert bot.delivery_handle == "kana" nameless_bot = channel.dm_address( SenderIdentity(id="7"), sender_kind=SenderKind.AGENT ) assert nameless_bot is not None assert nameless_bot.provider_thread_id == f"telegram:{bot_id}:7:0" + assert nameless_bot.delivery_handle is None assert ( channel.dm_address( SenderIdentity(id="ou_not_numeric"), sender_kind=SenderKind.HUMAN @@ -391,6 +393,29 @@ def update(sender_id: int, chat: dict[str, Any]) -> dict[str, Any]: assert len(audit.events) == 1 +def test_telegram_reads_the_chat_a_send_landed_in() -> None: + from bazaar_compute_node.contrib.telegram.identity import TelegramThreadIdentity + from bazaar_compute_node.contrib.telegram.outbound import TelegramOutboundChannel + + # a chat opened by name answers under its own id, and the send + # acknowledgement is where that id first appears + opened_by_name = TelegramThreadIdentity(bot_id=1, chat_id="@kana", topic_id=0) + assert ( + TelegramOutboundChannel._acknowledged_thread_id( + {"message_id": 5, "chat": {"id": 7, "type": "private"}}, opened_by_name + ) + == "telegram:1:7:0" + ) + + # an acknowledgement that names no chat leaves the conversation as it is + assert ( + TelegramOutboundChannel._acknowledged_thread_id( + {"message_id": 5}, opened_by_name + ) + is None + ) + + def test_telegram_identity_round_trips_numeric_and_username_chats() -> None: from bazaar_compute_node.contrib.telegram.identity import ( TelegramThreadIdentity, diff --git a/tests/support/src/bcn_test_support/channel.py b/tests/support/src/bcn_test_support/channel.py index ac7066ae..9d4bd15b 100644 --- a/tests/support/src/bcn_test_support/channel.py +++ b/tests/support/src/bcn_test_support/channel.py @@ -63,6 +63,7 @@ def __init__(self) -> None: self.stopped = False self.injected_messages: list[Message] = [] self.send_requests: list[ChannelSendRequest] = [] + self.delivered_thread_ids: dict[str, str] = {} self.send_attempts: list[ChannelSendRequest] = [] self.send_gate: asyncio.Event | None = None self.queued_messages: list[ChannelSendRequest] = [] @@ -91,13 +92,15 @@ def get_identity(self) -> ChannelIdentity | None: def dm_address( self, sender: SenderIdentity, *, sender_kind: SenderKind ) -> DmAddress | None: - del sender_kind if sender.id is None: return None return DmAddress( channel_session_id=f"channel-dm-{sender.id}", thread_id=f"thread-dm-{sender.id}", provider_thread_id=f"test:dm:{sender.id}", + # a peer that holds a handle can be reached by it before its own id + # can reach it + delivery_handle=sender.name if sender_kind is SenderKind.AGENT else None, ) async def start(self, *, timeout: float) -> None: @@ -208,7 +211,11 @@ async def send( result = ProviderCallResult( status=ProviderCallStatus.CONFIRMED, value=ChannelDeliveryReceipt( - provider_message_id=f"test-message-{len(self.send_attempts)}" + provider_message_id=f"test-message-{len(self.send_attempts)}", + # a chat opened by handle answers under the peer's own id + provider_thread_id=self.delivered_thread_ids.get( + request.provider_thread_id + ), ), ) if result.status is ProviderCallStatus.CONFIRMED: diff --git a/tests/support/src/bcn_test_support/storage.py b/tests/support/src/bcn_test_support/storage.py index 6dda8b0c..f61fe3ea 100644 --- a/tests/support/src/bcn_test_support/storage.py +++ b/tests/support/src/bcn_test_support/storage.py @@ -386,6 +386,14 @@ async def _unread_in_scope(self) -> list[Message]: async def count_unread_messages(self) -> int: return len(await self._unread_in_scope()) + async def has_outbound_for_command(self, command_id: str) -> bool: + return any( + message.command_id == command_id + and message.direction is MessageDirection.OUTBOUND + for thread in self._storage.messages.values() + for message in thread + ) + async def find_known_sender(self, token: str) -> KnownSender | None: inbound: list[Message] = [] for thread in self._scoped_threads(): @@ -1021,9 +1029,7 @@ def _validate_outbound_message_input(message: object) -> None: def _validate_outbound_insert(message: Message) -> None: - if message.delivery_state is not OutboundDeliveryState.PENDING: - raise ValueError("a new outbound message must start in pending state") - if any( + if message.delivery_state is OutboundDeliveryState.PENDING and any( value is not None for value in ( message.provider_message_id, @@ -1128,7 +1134,6 @@ def _validate_channel_session_update( if ( existing.channel != incoming.channel or existing.provider_thread_id != incoming.provider_thread_id - or existing.created_at_ms != incoming.created_at_ms ): raise ValueError("channel session identity cannot change") _validate_updated_at(existing.updated_at_ms, incoming.updated_at_ms)