From 535359129d2f625e43c6cb583e0abad536d41b6e Mon Sep 17 00:00:00 2001 From: Hanchin Hsieh Date: Wed, 9 Sep 2026 15:28:17 -0400 Subject: [PATCH 1/2] fix: reach one peer's DM by every name it answers to Co-Authored-By: Claude Opus 5 --- .../contrib/lark/channel.py | 2 +- .../contrib/sqlite/repository/facade.py | 2 +- .../contrib/sqlite/repository/sessions.py | 18 ++++- .../contrib/telegram/channel.py | 31 ++++++- .../contrib/wecom/channel.py | 2 +- src/bazaar_compute_node/core/channel.py | 12 +-- .../core/orchestration/command.py | 13 +-- .../core/orchestration/orchestrator.py | 37 ++++++++- src/bazaar_compute_node/core/storage.py | 2 +- tests/contrib/test_orchestration.py | 81 +++++++++++++++++++ tests/contrib/test_telegram_channel.py | 14 +++- tests/core/test_ports.py | 2 +- tests/support/src/bcn_test_support/channel.py | 8 +- tests/support/src/bcn_test_support/storage.py | 8 +- 14 files changed, 202 insertions(+), 30 deletions(-) diff --git a/src/bazaar_compute_node/contrib/lark/channel.py b/src/bazaar_compute_node/contrib/lark/channel.py index 6f9ef724..892c3d15 100644 --- a/src/bazaar_compute_node/contrib/lark/channel.py +++ b/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( diff --git a/src/bazaar_compute_node/contrib/sqlite/repository/facade.py b/src/bazaar_compute_node/contrib/sqlite/repository/facade.py index 4dcb6e7b..3ec1f143 100644 --- a/src/bazaar_compute_node/contrib/sqlite/repository/facade.py +++ b/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( diff --git a/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py b/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py index 1d61d3b4..06a3025b 100644 --- a/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py +++ b/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( diff --git a/src/bazaar_compute_node/contrib/telegram/channel.py b/src/bazaar_compute_node/contrib/telegram/channel.py index 0b2a932c..155f9d94 100644 --- a/src/bazaar_compute_node/contrib/telegram/channel.py +++ b/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, diff --git a/src/bazaar_compute_node/contrib/wecom/channel.py b/src/bazaar_compute_node/contrib/wecom/channel.py index d5b31ced..8d62aeff 100644 --- a/src/bazaar_compute_node/contrib/wecom/channel.py +++ b/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( diff --git a/src/bazaar_compute_node/core/channel.py b/src/bazaar_compute_node/core/channel.py index 82bd1f3c..719436ae 100644 --- a/src/bazaar_compute_node/core/channel.py +++ b/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( diff --git a/src/bazaar_compute_node/core/orchestration/command.py b/src/bazaar_compute_node/core/orchestration/command.py index 86b1132f..c2c7640d 100644 --- a/src/bazaar_compute_node/core/orchestration/command.py +++ b/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, diff --git a/src/bazaar_compute_node/core/orchestration/orchestrator.py b/src/bazaar_compute_node/core/orchestration/orchestrator.py index e09b7b1d..45f577c4 100644 --- a/src/bazaar_compute_node/core/orchestration/orchestrator.py +++ b/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 diff --git a/src/bazaar_compute_node/core/storage.py b/src/bazaar_compute_node/core/storage.py index 4c612d21..f2b4bffe 100644 --- a/src/bazaar_compute_node/core/storage.py +++ b/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( diff --git a/tests/contrib/test_orchestration.py b/tests/contrib/test_orchestration.py index 4e79d982..d56f97cd 100644 --- a/tests/contrib/test_orchestration.py +++ b/tests/contrib/test_orchestration.py @@ -4869,6 +4869,87 @@ async def test_a_sender_the_channel_cannot_address_stays_not_found() -> None: await orchestrator.stop(timeout=1) +@pytest.mark.asyncio +async def test_one_peer_addressed_two_ways_stays_one_conversation() -> None: + orchestrator, channel, _, storage, _ = await make_node( + mode=Mode.DANGEROUS_INDIVIDUAL + ) + try: + # This peer's DM was opened under its handle and carries no handle of + # its own, so `dm:@kana` cannot resolve to it and has to mint. + await channel.inject( + Message( + direction=MessageDirection.INBOUND, + seq=1, + message_id="message-dm-1", + thread_id="bcn-dm-handle", + channel_session_id="channel-dm-handle", + channel="test", + provider_thread_id="test:dm:@kana", + provider_message_id="provider-dm-1", + received_at_ms=1, + sender=SenderIdentity(id="peer-1", name="kana"), + message_type="text", + target="dm:channel-dm-handle", + target_kind=ChannelTargetKind.DM, + body="hello", + metadata={"sender_kind": SenderKind.AGENT.value}, + ) + ) + await wait_until(lambda: len(storage.channel_sessions) == 1) + + 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, + ) + + # Minting reached the conversation that was already open instead of + # opening a second half of it. + assert len(storage.channel_sessions) == 1 + assert ( + storage.channel_sessions["channel-dm-handle"].provider_thread_id + == "test:dm:@kana" + ) + + # The peer answers under its provider id, which is the other way to + # reach this same conversation. + await channel.inject( + Message( + direction=MessageDirection.INBOUND, + seq=2, + message_id="message-dm-2", + thread_id="bcn-dm-id", + channel_session_id="channel-dm-peer-1", + channel="test", + provider_thread_id="test:dm:peer-1", + provider_message_id="provider-dm-2", + received_at_ms=3, + sender=SenderIdentity(id="peer-1", name="kana"), + message_type="text", + target="dm:channel-dm-peer-1", + target_kind=ChannelTargetKind.DM, + body="hello back", + metadata={"sender_kind": SenderKind.AGENT.value}, + ) + ) + await wait_until( + lambda: ( + len( + _stored_messages( + storage, "bcn-dm-handle", direction=MessageDirection.INBOUND + ) + ) + == 2 + ) + ) + assert len(storage.channel_sessions) == 1 + 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_telegram_channel.py b/tests/contrib/test_telegram_channel.py index 165056d6..79b591dd 100644 --- a/tests/contrib/test_telegram_channel.py +++ b/tests/contrib/test_telegram_channel.py @@ -243,20 +243,28 @@ def build_api(*args: object, **kwargs: object) -> _FakeApi: sender_kind=SenderKind.HUMAN, ) assert human is not None - assert human.provider_thread_id == f"telegram:{bot_id}:{TEST_USER_ID}:0" + assert human.provider_thread_ids == ( + f"telegram:{bot_id}:{TEST_USER_ID}:0", + f"telegram:{bot_id}:@human:0", + ) # Bots reach each other by username; a numeric id does not apply. 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" + # That same bot speaks with a numeric chat id, so the conversation this + # address opens is the one its messages already arrive in. + assert bot.provider_thread_ids == ( + f"telegram:{bot_id}:@kana:0", + f"telegram:{bot_id}:7:0", + ) # A bot without a username falls back to the numeric id rather than # minting an address Telegram would reject. 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.provider_thread_ids == (f"telegram:{bot_id}:7:0",) assert ( channel.dm_address( SenderIdentity(id="ou_not_numeric"), sender_kind=SenderKind.HUMAN diff --git a/tests/core/test_ports.py b/tests/core/test_ports.py index 3fcdb7ef..36c59005 100644 --- a/tests/core/test_ports.py +++ b/tests/core/test_ports.py @@ -133,7 +133,7 @@ async def test_a_minted_dm_address_carries_the_same_namespace_as_a_received_one( body="hello", attachments=(), target_kind=ChannelTargetKind.DM, - provider_thread_id=address.provider_thread_id, + provider_thread_id=address.provider_thread_ids[0], ), timeout=1, ) diff --git a/tests/support/src/bcn_test_support/channel.py b/tests/support/src/bcn_test_support/channel.py index ac7066ae..970ebd4f 100644 --- a/tests/support/src/bcn_test_support/channel.py +++ b/tests/support/src/bcn_test_support/channel.py @@ -97,7 +97,13 @@ def dm_address( 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 is reachable by it too, and that is + # the form an already open conversation may be stored under + provider_thread_ids=( + (f"test:dm:{sender.id}", f"test:dm:@{sender.name}") + if sender.name + else (f"test:dm:{sender.id}",) + ), ) async def start(self, *, timeout: float) -> None: diff --git a/tests/support/src/bcn_test_support/storage.py b/tests/support/src/bcn_test_support/storage.py index 6dda8b0c..ca01af75 100644 --- a/tests/support/src/bcn_test_support/storage.py +++ b/tests/support/src/bcn_test_support/storage.py @@ -219,7 +219,7 @@ async def record_inbound( message = cast(Message, existing_message) channel_session = await self.find_channel_session( channel=channel, - provider_thread_id=provider_thread_id, + provider_thread_ids=(provider_thread_id,), ) channel_session_created = channel_session is None if channel_session is None: @@ -315,14 +315,14 @@ async def find_channel_session( self, *, channel: str, - provider_thread_id: str, + provider_thread_ids: tuple[str, ...], ) -> ChannelSession | None: matches = [ session for session in self._storage.channel_sessions.values() if ( session.channel == channel - and session.provider_thread_id == provider_thread_id + and session.provider_thread_id in provider_thread_ids ) ] if len(matches) > 1: @@ -739,7 +739,7 @@ async def save_channel_session(self, session: ChannelSession) -> None: else: 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( From edf88e6562f4b4964819b5a70b96a695bd761356 Mon Sep 17 00:00:00 2001 From: Hanchin Hsieh Date: Wed, 9 Sep 2026 21:02:02 -0400 Subject: [PATCH 2/2] fix: merge the DM conversations an earlier release split in two Co-Authored-By: Claude Opus 5 --- .../contrib/sqlite/migrations/registry.py | 2 + .../v27_merge_split_dm_conversations.py | 118 +++++++++++++++ tests/contrib/test_sqlite_database.py | 140 +++++++++++++++++- 3 files changed, 256 insertions(+), 4 deletions(-) create mode 100644 src/bazaar_compute_node/contrib/sqlite/migrations/v27_merge_split_dm_conversations.py diff --git a/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py b/src/bazaar_compute_node/contrib/sqlite/migrations/registry.py index e3004c14..b5ab583d 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_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, ) diff --git a/src/bazaar_compute_node/contrib/sqlite/migrations/v27_merge_split_dm_conversations.py b/src/bazaar_compute_node/contrib/sqlite/migrations/v27_merge_split_dm_conversations.py new file mode 100644 index 00000000..8280a0ad --- /dev/null +++ b/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"] diff --git a/tests/contrib/test_sqlite_database.py b/tests/contrib/test_sqlite_database.py index 1b6cf46a..2e350cf1 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,139 @@ 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_merges_a_dm_that_was_stored_under_two_names() -> 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 was reached by handle and answers by chat id: one conversation. + # mika is two different peers who happen to share a display name. + 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', ?, 'dm', 1, '{}', ?, ?, 1, 1, 'agent-1')", + ( + ("channel-kana-handle", "telegram:1:@kana:0", "kana", "kana"), + ("channel-kana-chat", "telegram:1:7:0", "kana", "kana"), + ("channel-mika-one", "telegram:1:8:0", "mika", "mika"), + ("channel-mika-two", "telegram:1:9:0", "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-kana-handle", "channel-kana-handle"), + ("thread-kana-chat", "channel-kana-chat"), + ("thread-mika-one", "channel-mika-one"), + ("thread-mika-two", "channel-mika-two"), + ), + ) + 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 ('outbound-to-kana', 1, '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', 1, 1, '[]')" + ) + await connection.execute( + "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-from-kana', 2, 'inbound', 'agent-1', " + "'thread-kana-chat', 'channel-kana-chat', 'telegram', 'telegram:1:7:0', " + "'11', 2, 'kana', '7', 'text', 'dm:@kana', 'dm', 'hello back', 0, 1, " + '\'{"sender_kind":"agent"}\')' + ) + await connection.executemany( + "INSERT INTO consumer_cursors (" + "thread_id, delivered_through_seq, updated_at_ms" + ") VALUES (?, ?, 1)", + (("thread-kana-handle", 1), ("thread-kana-chat", 0)), + ) + await connection.execute( + "INSERT INTO reminders (" + "reminder_id, owner_thread_id, anchor_message_id, title, state, " + "next_fire_at_ms, revision, last_occurrence_no, created_at_ms, " + "updated_at_ms, agent_id" + ") VALUES ('018f0000-0000-7000-8000-00000000002a', 'thread-kana-handle', " + "'inbound-from-kana', 'follow up with kana', 'scheduled', 100, 1, 0, 1, " + "1, 'agent-1')" + ) + 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 FROM channel_sessions ORDER BY id" + ) + threads = await session.fetchall("SELECT id FROM threads ORDER BY id") + moved = await session.fetchone( + "SELECT thread_id, channel_session_id, provider_thread_id, target " + "FROM messages WHERE message_id = 'outbound-to-kana'" + ) + cursors = await session.fetchall( + "SELECT thread_id, delivered_through_seq FROM consumer_cursors " + "ORDER BY thread_id" + ) + reminder = await session.fetchone("SELECT owner_thread_id FROM reminders") + + # the two halves of kana's conversation are one, and the two peers who + # share a display name are still two + assert [row["id"] for row in sessions] == [ + "channel-kana-chat", + "channel-mika-one", + "channel-mika-two", + ] + assert [row["id"] for row in threads] == [ + "thread-kana-chat", + "thread-mika-one", + "thread-mika-two", + ] + assert moved is not None + assert moved["thread_id"] == "thread-kana-chat" + assert moved["channel_session_id"] == "channel-kana-chat" + assert moved["target"] == "dm:@kana" + # how this message was addressed is a fact about the message, and the + # merge does not get to rewrite it + assert moved["provider_thread_id"] == "telegram:1:@kana:0" + # the surviving cursor covers what it inherited + assert [ + (row["thread_id"], row["delivered_through_seq"]) for row in cursors + ] == [("thread-kana-chat", 1)] + assert reminder is not None + assert reminder["owner_thread_id"] == "thread-kana-chat" finally: await database.stop(timeout=2) @@ -1291,7 +1423,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 +1532,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