Reach one peer's DM by every name it answers to

src/bazaar_compute_node/contrib/lark/channel.py

@@ -964,7 +964,7 @@ def dm_address(
         return DmAddress(
             channel_session_id=thread.channel_session_id,
             thread_id=thread.session_id,
-            provider_thread_id=thread.provider_thread_id,
+            provider_thread_ids=(thread.provider_thread_id,),
         )
 
     async def send(

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_merge_split_dm_conversations import SPLIT_DM_MERGE_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,
+    SPLIT_DM_MERGE_MIGRATION,
 )
 
 

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

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+from .model import Migration
+
+# A DM the agent opened by handle and the same DM the peer speaks in are one
+# conversation stored twice, because each side named it its own way. They are
+# recognisable as a pair: one row is keyed by the very handle both answer to and
+# the other is not. Two different peers who share a display name are not, since
+# neither of their rows is keyed by that name.
+_PAIRS = """
+    SELECT
+        loser.agent_id AS agent_id,
+        loser.id AS loser_id,
+        loser_thread.id AS loser_thread_id,
+        winner.id AS winner_id,
+        winner_thread.id AS winner_thread_id,
+        CASE
+            WHEN winner.target_handle IS NOT NULL THEN 'dm:@' || winner.target_handle
+            ELSE 'dm:' || winner.id
+        END AS winner_target
+    FROM channel_sessions AS loser
+    JOIN threads AS loser_thread
+      ON loser_thread.agent_id = loser.agent_id
+     AND loser_thread.channel_session_id = loser.id
+    JOIN channel_sessions AS winner
+      ON winner.agent_id = loser.agent_id
+     AND winner.channel = loser.channel
+     AND winner.target_kind = 'dm'
+     AND winner.target_handle_key = loser.target_handle_key
+     AND winner.id <> loser.id
+     AND LOWER(winner.provider_thread_id)
+         NOT LIKE '%:@' || winner.target_handle_key || ':%'
+    JOIN threads AS winner_thread
+      ON winner_thread.agent_id = winner.agent_id
+     AND winner_thread.channel_session_id = winner.id
+    WHERE loser.target_kind = 'dm'
+      AND loser.target_handle_key IS NOT NULL
+      AND LOWER(loser.provider_thread_id)
+          LIKE '%:@' || loser.target_handle_key || ':%'
+      AND (
+        SELECT COUNT(*) FROM channel_sessions AS peer
+        WHERE peer.agent_id = loser.agent_id
+          AND peer.channel = loser.channel
+          AND peer.target_kind = 'dm'
+          AND peer.target_handle_key = loser.target_handle_key
+      ) = 2
+"""
+
+SPLIT_DM_MERGE_MIGRATION = Migration(
+    version=27,
+    name="merge_split_dm_conversations",
+    statements=(
+        # the rows that identify a pair are the rows this merge deletes, so the
+        # set has to be settled before the first of them goes
+        f"""
+        CREATE TEMPORARY TABLE bcn_split_dm_pairs AS {_PAIRS}
+        """,
+        # the surviving cursor cannot fall behind messages it is about to own
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        UPDATE consumer_cursors SET delivered_through_seq = MAX(
+            delivered_through_seq,
+            COALESCE((
+                SELECT losing.delivered_through_seq
+                FROM consumer_cursors AS losing
+                JOIN pairs ON pairs.loser_thread_id = losing.thread_id
+                WHERE pairs.winner_thread_id = consumer_cursors.thread_id
+            ), 0)
+        )
+        WHERE thread_id IN (SELECT winner_thread_id FROM pairs)
+        """,
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        DELETE FROM consumer_cursors
+        WHERE thread_id IN (SELECT loser_thread_id FROM pairs)
+        """,
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        UPDATE reminders SET owner_thread_id = (
+            SELECT winner_thread_id FROM pairs
+            WHERE pairs.loser_thread_id = reminders.owner_thread_id
+        )
+        WHERE owner_thread_id IN (SELECT loser_thread_id FROM pairs)
+        """,
+        # provider_thread_id stays as it was: it records how this very message
+        # was addressed, and rewriting it would claim it arrived another way
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        UPDATE messages SET
+            thread_id = (
+                SELECT winner_thread_id FROM pairs
+                WHERE pairs.loser_id = messages.channel_session_id
+            ),
+            target = (
+                SELECT winner_target FROM pairs
+                WHERE pairs.loser_id = messages.channel_session_id
+            ),
+            channel_session_id = (
+                SELECT winner_id FROM pairs
+                WHERE pairs.loser_id = messages.channel_session_id
+            )
+        WHERE channel_session_id IN (SELECT loser_id FROM pairs)
+        """,
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        DELETE FROM threads WHERE id IN (SELECT loser_thread_id FROM pairs)
+        """,
+        """
+        WITH pairs AS (SELECT * FROM bcn_split_dm_pairs)
+        DELETE FROM channel_sessions WHERE id IN (SELECT loser_id FROM pairs)
+        """,
+        """
+        DROP TABLE bcn_split_dm_pairs
+        """,
+    ),
+)
+
+__all__ = ["SPLIT_DM_MERGE_MIGRATION"]

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

@@ -37,7 +37,7 @@ async def _inbound_channel_session(
 
         channel_session = await self.find_channel_session(
             channel=channel,
-            provider_thread_id=provider_thread_id,
+            provider_thread_ids=(provider_thread_id,),
         )
         if channel_session is None:
             channel_session = ChannelSession(

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

@@ -30,16 +30,26 @@ async def find_channel_session(
         self,
         *,
         channel: str,
-        provider_thread_id: str,
+        provider_thread_ids: tuple[str, ...],
     ) -> ChannelSession | None:
+        """Find the conversation stored under any of these identities.
+
+        A channel that addresses one peer several ways offers every form it
+        knows, and they answer to a single row: whichever form opened the
+        conversation is the one it keeps.
+        """
+
+        if not provider_thread_ids:
+            return None
+        placeholders = ", ".join("?" for _ in provider_thread_ids)
         row = await self._fetch_one_or_conflict(
             "SELECT id, channel, provider_thread_id, target_kind, following, "
             "created_at_ms, updated_at_ms, last_inbound_at_ms, last_outbound_at_ms, "
             "target_display_name, target_handle, target_handle_key, "
             "provider_identity_ref_json FROM channel_sessions "
             "WHERE agent_id = /*agent_id*/? AND channel = ? "
-            "AND provider_thread_id = ? ORDER BY rowid",
-            (channel, provider_thread_id),
+            f"AND provider_thread_id IN ({placeholders}) ORDER BY rowid",
+            (channel, *provider_thread_ids),
             "channel provider identity",
         )
         return channel_session_from_row(row) if row is not None else None
@@ -142,7 +152,7 @@ async def save_channel_session(self, session: ChannelSession) -> None:
         if existing is None:
             duplicate = await self.find_channel_session(
                 channel=session.channel,
-                provider_thread_id=session.provider_thread_id,
+                provider_thread_ids=(session.provider_thread_id,),
             )
             if duplicate is not None:
                 raise ValueError(

src/bazaar_compute_node/contrib/telegram/channel.py

@@ -3,7 +3,7 @@
 import asyncio
 import logging
 from collections.abc import AsyncIterator, Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from time import time_ns
 from unicodedata import category
 
@@ -468,9 +468,36 @@ def dm_address(
         return DmAddress(
             channel_session_id=identity.channel_session_id,
             thread_id=identity.session_id,
-            provider_thread_id=identity.provider_thread_id,
+            provider_thread_ids=self._dm_identities(identity, sender),
         )
 
+    @staticmethod
+    def _dm_identities(
+        identity: TelegramThreadIdentity, sender: SenderIdentity | None
+    ) -> tuple[str, ...]:
+        """Every identity this one DM answers to, the given one first.
+
+        A peer reached by `@username` and the same peer reached by its numeric
+        id are one conversation, and either form may be the one that opened it.
+        """
+
+        identities = (identity.provider_thread_id,)
+        if sender is None:
+            return identities
+        chat_id: int | str
+        if isinstance(identity.chat_id, str):
+            if sender.id is None:
+                return identities
+            try:
+                chat_id = int(sender.id)
+            except ValueError:
+                return identities
+        elif sender.name:
+            chat_id = f"@{sender.name}"
+        else:
+            return identities
+        return (*identities, replace(identity, chat_id=chat_id).provider_thread_id)
+
     async def send(
         self,
         request: ChannelSendRequest,

src/bazaar_compute_node/contrib/wecom/channel.py

@@ -495,7 +495,7 @@ def dm_address(
         return DmAddress(
             channel_session_id=str(uuid5(NAMESPACE_URL, identity)),
             thread_id=str(uuid5(NAMESPACE_URL, f"bcn:{identity}")),
-            provider_thread_id=sender.id,
+            provider_thread_ids=(sender.id,),
         )
 
     async def send(

src/bazaar_compute_node/core/channel.py

@@ -126,14 +126,16 @@ async def request_approval(
 class DmAddress:
     """A DM conversation in one channel's own terms.
 
-    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.
+    A channel that can address the same peer more than one way reaches the same
+    conversation under any of `provider_thread_ids`: an inbound message that
+    arrived under one of them already opened this conversation, and opening a
+    second one would split it in half. A conversation that does not exist yet is
+    opened under the first.
     """
 
     channel_session_id: str
     thread_id: str
-    provider_thread_id: str
+    provider_thread_ids: tuple[str, ...]
 
 
 class IChannel(IAsyncLifecycle, IApproval, Protocol):
@@ -257,7 +259,7 @@ def dm_address(
                 address.channel_session_id,
             ),
             thread_id=thread_id,
-            provider_thread_id=address.provider_thread_id,
+            provider_thread_ids=address.provider_thread_ids,
         )
 
     async def send(

src/bazaar_compute_node/core/orchestration/command.py

@@ -266,14 +266,17 @@ async def _resolve_or_mint(
             if address is None:
                 raise
         now = self._clock()
-        stored_session = await self._storage.get_channel_session(
-            address.channel_session_id
+        # the peer may already have written first under another of its
+        # identities, and that conversation is this one
+        stored_session = await self._storage.find_channel_session(
+            channel=known.channel,
+            provider_thread_ids=address.provider_thread_ids,
         )
         if stored_session is None:
             stored_session = ChannelSession(
                 id=address.channel_session_id,
                 channel=known.channel,
-                provider_thread_id=address.provider_thread_id,
+                provider_thread_id=address.provider_thread_ids[0],
                 created_at_ms=now,
                 updated_at_ms=now,
                 target_kind=ChannelTargetKind.DM,
@@ -289,11 +292,11 @@ async def _resolve_or_mint(
                 target_handle_key=handle.casefold(),
             )
         )
-        stored_thread = await self._storage.get_thread(address.thread_id)
+        stored_thread = await self._storage.find_thread(stored_session.id)
         if stored_thread is None:
             stored_thread = ConversationRow(
                 id=address.thread_id,
-                channel_session_id=address.channel_session_id,
+                channel_session_id=stored_session.id,
                 workspace_id=self._actors.agent_id,
                 created_at_ms=now,
                 updated_at_ms=now,

src/bazaar_compute_node/core/orchestration/orchestrator.py

@@ -19,6 +19,7 @@
 from ..lifecycle import IAsyncLifecycle, TimeoutBudget
 from ..models import (
     ChannelSession,
+    ChannelTargetKind,
     Message,
     RuntimeAttempt,
     RuntimeEventState,
@@ -1034,12 +1035,46 @@ def _runtime_worker_done(
         )
         self._start_runtime_worker(actor, queue)
 
+    async def _settle_provider_thread_id(self, message: Message) -> Message:
+        """Name this conversation the way it is already stored.
+
+        A channel that can address one peer several ways may hand this message
+        the name that peer speaks under while the conversation was opened under
+        another. Only the stored row settles which one it answers to, and from
+        here on there is one name again.
+        """
+
+        if (
+            message.target_kind is not ChannelTargetKind.DM
+            or message.channel is None
+            or message.provider_thread_id is None
+            or message.sender is None
+        ):
+            return message
+        address = self._channel.dm_address(
+            message.sender, sender_kind=message.sender_kind
+        )
+        if address is None:
+            return message
+        candidates = tuple(
+            dict.fromkeys((message.provider_thread_id, *address.provider_thread_ids))
+        )
+        if len(candidates) == 1:
+            return message
+        stored = await self._storage.find_channel_session(
+            channel=message.channel,
+            provider_thread_ids=candidates,
+        )
+        if stored is None or stored.provider_thread_id == message.provider_thread_id:
+            return message
+        return replace(message, provider_thread_id=stored.provider_thread_id)
+
     async def _record_inbound(
         self,
         message: Message,
     ) -> tuple[_DurableTurnContext | None, Message, bool]:
         recorded = await self._storage.record_inbound(
-            message,
+            await self._settle_provider_thread_id(message),
             now_ms=self._clock(),
         )
         message = recorded.message

src/bazaar_compute_node/core/storage.py

@@ -570,7 +570,7 @@ async def finalize_outbound_delivery(
     ) -> Message[OutboundAttachment]: ...
 
     async def find_channel_session(
-        self, *, channel: str, provider_thread_id: str
+        self, *, channel: str, provider_thread_ids: tuple[str, ...]
     ) -> ChannelSession | None: ...
 
     async def get_channel_session(