Name a DM by the conversation the provider delivered into

plans/2026-09-05-dm-address-resolution.md

@@ -23,13 +23,14 @@
 
 各 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:<bot>: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`），两种发送者共用同一结构，
@@ -50,7 +51,8 @@
 2. 未命中且目标形如 `dm:@name` 时，在当前 actor 可达范围内按 handle 查历史入站发送者，
    取其 `sender_id` 与所属 channel。
 3. 把该身份交给 channel 换取私聊的 provider 地址。
-4. 换到地址则建立 `channel_sessions` 与 `threads` 两行并重新解析；换不到则抛
+4. 按 `(channel, provider_thread_id)` 找现有会话——会话行的 id 由建立它的那一刻决定，不必等于此刻
+   由身份算出的那个；找不到才建立 `channel_sessions` 与 `threads` 两行，然后重新解析；换不到地址则抛
    `InboxTargetResolutionError`，`bcc` 调用方看到的仍是「找不到」。
 
 发送失败不回滚已建立的映射。地址换不出来时没有别的补救手段，保留映射与删除映射对调用方没有区别。
@@ -80,6 +82,35 @@
 看到并且可以照抄的那一段（`resources/command/sender.tpl` 在没有 handle 时渲染 `@<sender_id>`），
 因此按 `messages.sender_id` 匹配这一级足以覆盖飞书，不需要为飞书补 handle。
 
+## 会话身份由 provider 认定
+
+会话行的键是对方的 provider 身份，不是任何可被改名或转让的名字。Telegram 上首次发往一个尚不存在的
+私聊必须用 `@username`，这是投递地址，与会话身份是两件事：混用会让「我方发出的」与「对方回来的」
+落在两行上，同一段对话被劈成两半。
+
+`sendMessage` 成功时返回发出的那条 Message，其中必然带 `chat`，因此每一次投递都附带 provider 自己认定
+的会话 id，与本次用什么形式寻址无关。据此：
+
+- 出站回执 `ChannelDeliveryReceipt` 与 `OutboundDeliveryResult` 携带 `provider_thread_id`，由 channel 从
+  回执中取出并按自己的身份串格式拼出。
+- 一次投递之后，若回执给出的身份与会话行不同，则以回执为准。目标身份已经有行时不改，两行同一个身份
+  比一个滞后的名字更糟；`channel_sessions` 上的 provider 身份索引不是唯一索引，重复不会被数据库挡下。
+- 身份变更是一次窄操作 `rebind_channel_session`，不走 `save_channel_session`——后者明确禁止改动
+  provider 身份，该限制在其余路径上继续成立。
+
+## 首次投递携带 handle
+
+`DmAddress` 携带 `delivery_handle`：打开一个尚不存在的私聊所需的名字，由 channel 自己给出——需不需要
+一个名字、什么样的名字算数，只有 channel 知道。telegram 对 bot 给出其 username，对人不给，其余 channel
+不给。mint 把 channel 给出的值写入会话行的 `provider_identity_ref_json`，`ChannelSendRequest` 再把它带回
+channel；telegram 出站在有该值时用 `@handle` 作为 `chat_id`，否则用身份中的数字 id。打字状态的路由仍按
+数字身份注册，`@handle` 只进入本次发送的 payload。
+
+投递一旦有任何一段被确认，该私聊就已经存在，数字 id 足以送达，会话行上的 `delivery_handle` 随即清除；
+多段发送中前面几段成功而后面失败时，整体结果可能是 `unknown`，此时回执里仍带着被确认那段所属的会话 id，
+按它同样清除。清除前重新读取该行：清一个键不该带上发送前的快照，否则会把期间到达的入站回执盖掉。
+清除与身份变更同时发生时，先清除后变更：变更之后再保存整行会带上该行已经没有的身份。
+
 ## Tasks
 
 ### Task 1：按 `@` 位置的取值查历史发送者
@@ -98,8 +129,8 @@
 两个 id 由各 channel 用**自己的身份串**做 uuid5 得出，core 无法从 `provider_thread_id` 推导。若由 core
 自行编号，等对方之后真的发消息进来，channel 会算出另一组 id，同一个人会出现两条会话。
 
-`sender_kind` 供 Telegram 区分 bot 与人，其余 channel 忽略。该参数不进入 agent 的视野：agent 始终只写
-`dm:@name`，人与 bot 的差别只存在于内部解析。
+`sender_kind` 用于排除来路不明的发送者，并决定 `delivery_handle`：以频道或匿名管理员身份发到群里的消息带的是该群的 id 而没有
+`from`，按它建立的私聊会把内容发回原群。该参数不进入 agent 的视野：agent 始终只写 `dm:@name`。
 
 ### Task 3：未命中时建立映射
 
@@ -113,7 +144,23 @@
 冷启动私聊的目标会话必然零入站，与该校验直接冲突。该校验没有任何测试覆盖，也不来自任何既定要求，
 因此删除；配套的 `ErrorKind.TARGET_NOT_REPLYABLE` 随之成为死代码，一并删除。
 
-### Task 4：端到端测试
+### Task 4：迁移既有的按 handle 命名的会话
+
+按 handle 命名的 dm 行的正身可从入站消息推出：入站同时带着对方说话时用的 handle 与其 `sender_id`，
+取该 handle 下 seq 最大的一条，与 `_resolve_or_mint` 的取法一致。据此：
+
+- 对方已经有按其 id 命名的会话时，把按 handle 命名的那行的 messages、thread、reminder 与 cursor
+  并入该会话后删除该行。消息上的 `provider_thread_id` 不改写——它记录的是该条消息当时如何被寻址；
+  `target` 改写为留存会话的 `canonical_target`。
+- 对方还没有这样的会话时，就地把该行改名为按 id 命名，并把原 handle 写入 `delivery_handle`，
+  使下一次发送仍能打开尚未建立的私聊。
+
+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` 以覆盖「找不到」路径。

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,
 )
 
 

src/bazaar_compute_node/contrib/sqlite/migrations/v27_name_dms_by_the_peer_id.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from .model import Migration
+
+# A DM opened by handle kept that handle as its name, while the peer speaks
+# under its own id, so one conversation ended 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.
+_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
+            ),
+            provider_identity_ref_json = JSON_SET(
+                CASE
+                    WHEN JSON_VALID(provider_identity_ref_json)
+                    THEN provider_identity_ref_json
+                    ELSE '{}'
+                END,
+                '$.delivery_handle',
+                target_handle
+            )
+        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"]

src/bazaar_compute_node/contrib/sqlite/repository/sessions.py

@@ -44,6 +44,26 @@ async def find_channel_session(
         )
         return channel_session_from_row(row) if row is not None else None
 
+    async def rebind_channel_session(
+        self,
+        channel_session_id: str,
+        *,
+        provider_thread_id: str,
+        updated_at_ms: int,
+    ) -> None:
+        """Let the provider rename a conversation this node opened blind.
+
+        Addressing a chat that does not exist yet takes a name rather than an
+        id, and the provider answers with the id it keeps. Everything already
+        written stays where it is; only the name this row answers to moves.
+        """
+
+        await self.execute(
+            "UPDATE channel_sessions SET provider_thread_id = ?, updated_at_ms = ? "
+            "WHERE agent_id = /*agent_id*/? AND id = ?",
+            (provider_thread_id, updated_at_ms, channel_session_id),
+        )
+
     async def get_channel_session(
         self, channel_session_id: str
     ) -> ChannelSession | None:

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, which is a matter of delivery.
+        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,
+            # a chat with a bot that has never been spoken to is only reachable
+            # by its 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(