From eee262c8448e2981a856a482f727ac50e159b277 Mon Sep 17 00:00:00 2001 From: Hanchin Hsieh Date: Thu, 10 Sep 2026 00:13:19 -0400 Subject: [PATCH 1/3] docs: name a DM by its peer rather than by a name it can lose Co-Authored-By: Claude Opus 5 --- plans/2026-09-05-dm-address-resolution.md | 69 +++++++++++++++++++---- 1 file changed, 58 insertions(+), 11 deletions(-) diff --git a/plans/2026-09-05-dm-address-resolution.md b/plans/2026-09-05-dm-address-resolution.md index 3871787a..bc2c6d76 100644 --- a/plans/2026-09-05-dm-address-resolution.md +++ b/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::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 时渲染 `@`）， 因此按 `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` 以覆盖「找不到」路径。 From 51fc91d7415024b9de1cdf477008ee4e62247f94 Mon Sep 17 00:00:00 2001 From: Hanchin Hsieh Date: Thu, 10 Sep 2026 00:13:20 -0400 Subject: [PATCH 2/3] fix: name a DM by its peer and open a new chat by handle Co-Authored-By: Claude Opus 5 --- .../contrib/sqlite/repository/sessions.py | 20 ++++ .../contrib/telegram/channel.py | 25 ++--- .../contrib/telegram/outbound.py | 49 ++++++++- src/bazaar_compute_node/core/channel.py | 10 ++ .../core/orchestration/command.py | 73 ++++++++++++- .../core/orchestration/delivery.py | 9 ++ src/bazaar_compute_node/core/outcomes.py | 1 + src/bazaar_compute_node/core/storage.py | 11 ++ tests/contrib/test_orchestration.py | 100 ++++++++++++++++++ tests/contrib/test_telegram_channel.py | 28 ++++- tests/support/src/bcn_test_support/channel.py | 9 +- tests/support/src/bcn_test_support/storage.py | 16 +++ 12 files changed, 325 insertions(+), 26 deletions(-) diff --git a/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py b/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py index 1d61d3b4..46d543e4 100644 --- a/src/bazaar_compute_node/contrib/sqlite/repository/sessions.py +++ b/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: diff --git a/src/bazaar_compute_node/contrib/telegram/channel.py b/src/bazaar_compute_node/contrib/telegram/channel.py index 0b2a932c..f5e250f9 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, 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( diff --git a/src/bazaar_compute_node/contrib/telegram/outbound.py b/src/bazaar_compute_node/contrib/telegram/outbound.py index a9f02548..9bc3b13b 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,19 @@ def receipt(self) -> Mapping[str, object]: if receipt.get("state") == "confirmed" and isinstance(receipt.get("provider_message_id"), str) ) + thread_ids = tuple( + receipt["provider_thread_id"] + for receipt in self.receipts + if receipt.get("state") == "confirmed" + and isinstance(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, } @@ -110,6 +117,8 @@ def _outbound_route( "Telegram outbound route belongs to another bot", ) 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 +334,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 +435,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 +678,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 from here + on, and the send acknowledgement is the first place that id 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 +709,20 @@ def _channel_receipt(receipts: list[dict[str, object]]) -> ChannelDeliveryReceip raise AssertionError( "confirmed Telegram delivery requires provider message id" ) + provider_thread_id = next( + ( + value + for receipt in receipts + if isinstance(value := receipt.get("provider_thread_id"), str) + ), + None, + ) return ChannelDeliveryReceipt( provider_message_id=confirmed_ids[0], provider_receipt_ref=( confirmed_ids[-1] if len(confirmed_ids) > 1 else None ), + provider_thread_id=provider_thread_id, ) diff --git a/src/bazaar_compute_node/core/channel.py b/src/bazaar_compute_node/core/channel.py index 82bd1f3c..a01f3a59 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 was addressed by a name rather than by its id + 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,9 @@ class ChannelSendRequest: target_kind: ChannelTargetKind provider_thread_id: str provider_reply_to_message_id: str | None = None + # a chat this node has never spoken in cannot be reached by its id yet, and + # this is the name that opens it + delivery_handle: str | None = None @dataclass(frozen=True, slots=True) @@ -134,6 +140,9 @@ class DmAddress: channel_session_id: str thread_id: str provider_thread_id: str + # what it takes to open this chat while its id cannot reach it yet, for a + # channel where that is a different thing from naming it + 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..1775592b 100644 --- a/src/bazaar_compute_node/core/orchestration/command.py +++ b/src/bazaar_compute_node/core/orchestration/command.py @@ -266,8 +266,9 @@ 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 + stored_session = await self._storage.find_channel_session( + channel=known.channel, + provider_thread_id=address.provider_thread_id, ) if stored_session is None: stored_session = ChannelSession( @@ -277,6 +278,11 @@ async def _resolve_or_mint( created_at_ms=now, updated_at_ms=now, target_kind=ChannelTargetKind.DM, + metadata=( + {"delivery_handle": address.delivery_handle} + if address.delivery_handle is not None + else {} + ), ) # The name this conversation already answers to comes first, then what # the provider calls the peer, and only then the token that found it. @@ -289,11 +295,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, @@ -403,6 +409,14 @@ async def _transmit( target_kind=channel_session.target_kind, provider_thread_id=channel_session.provider_thread_id, provider_reply_to_message_id=prepared.reply_to_provider_message_id, + delivery_handle=( + handle + if isinstance( + handle := channel_session.metadata.get("delivery_handle"), str + ) + and handle + else None + ), ) ) outbound = replace( @@ -425,8 +439,59 @@ async def _transmit( "delivery_receipt": dict(delivery_result.receipt), }, ) + await self._settle_delivered_conversation(channel_session, delivery_result) return outbound, delivery_result + async def _settle_delivered_conversation( + self, + channel_session: ChannelSession, + delivery_result: OutboundDeliveryResult, + ) -> None: + """Take what a delivery says about the conversation it landed in. + + A chat that has now been spoken in at all is reachable by its own id, + so the name that opened it stops being needed. The id itself normally only + confirms the name this conversation already has; when it does not, the + provider is the authority on where the message went, unless that + conversation is already open here: two rows under one name is worse + than a name that lags. + """ + + # the handle goes first: saving the row again after a rebind would + # offer an identity the row no longer has + opened = delivery_result.provider_thread_id is not None or ( + delivery_result.state + in (OutboundDeliveryState.SENT, OutboundDeliveryState.PARTIAL) + ) + if opened and "delivery_handle" in channel_session.metadata: + current = await self._storage.get_channel_session(channel_session.id) + if current is not None and "delivery_handle" in current.metadata: + await self._storage.save_channel_session( + replace( + current, + metadata={ + key: value + for key, value in current.metadata.items() + if key != "delivery_handle" + }, + updated_at_ms=self._clock(), + ) + ) + delivered = delivery_result.provider_thread_id + if delivered is None or delivered == channel_session.provider_thread_id: + return + opened = await self._storage.find_channel_session( + channel=channel_session.channel, + provider_thread_id=delivered, + ) + if opened is not None: + return + await self._storage.rebind_channel_session( + channel_session.id, + provider_thread_id=delivered, + updated_at_ms=self._clock(), + ) + async def _record_delivery( self, audit_context: CorrelationContext, diff --git a/src/bazaar_compute_node/core/orchestration/delivery.py b/src/bazaar_compute_node/core/orchestration/delivery.py index 716474c4..358c2e02 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,15 @@ 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 + # a delivery that ended badly can still have opened the conversation + 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_receipt_ref=provider_receipt_ref, + provider_thread_id=provider_thread_id, error_kind=( provider_result.error_kind or ErrorKind.PROVIDER_FAILED.value ), @@ -104,6 +112,7 @@ def _map_provider_result( return OutboundDeliveryResult( state=OutboundDeliveryState.UNKNOWN, 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..1884046a 100644 --- a/src/bazaar_compute_node/core/storage.py +++ b/src/bazaar_compute_node/core/storage.py @@ -576,6 +576,17 @@ async def find_channel_session( async def get_channel_session( self, channel_session_id: str ) -> ChannelSession | 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.""" + ... + async def get_thread(self, thread_id: str) -> Thread | None: ... async def find_thread(self, channel_session_id: str) -> Thread | None: ... async def get_runtime_attempt(self, turn_id: str) -> RuntimeAttempt | None: ... diff --git a/tests/contrib/test_orchestration.py b/tests/contrib/test_orchestration.py index 4e79d982..2bfe7547 100644 --- a/tests/contrib/test_orchestration.py +++ b/tests/contrib/test_orchestration.py @@ -4869,6 +4869,106 @@ async def test_a_sender_the_channel_cannot_address_stays_not_found() -> None: await orchestrator.stop(timeout=1) +@pytest.mark.asyncio +async def test_a_minted_dm_is_named_by_its_peer_and_opened_by_handle() -> None: + orchestrator, channel, _, storage, _ = await make_node( + mode=Mode.DANGEROUS_INDIVIDUAL + ) + try: + 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 the 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 conversation is named after the peer, and the handle rides along + # as what opens a chat that is not there yet. + minted = storage.channel_sessions["channel-dm-peer-1"] + assert minted.provider_thread_id == "test:dm:peer-1" + await wait_until( + lambda: any( + sent.delivery_handle == "kana" and sent.body == "hello in private" + for sent in channel.sent_messages + ) + ) + # the chat has been spoken in now, so its id reaches it and the name + # that opened it is done + await wait_until( + lambda: ( + "delivery_handle" + not in storage.channel_sessions["channel-dm-peer-1"].metadata + ) + ) + + # so the peer answering under its own id reaches that 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:peer-1", + 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_telegram_channel.py b/tests/contrib/test_telegram_channel.py index 165056d6..1a9f4610 100644 --- a/tests/contrib/test_telegram_channel.py +++ b/tests/contrib/test_telegram_channel.py @@ -244,14 +244,13 @@ 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. + # A bot is named by its own id too: a username opens a chat that does + # not exist yet, which is a matter of delivery rather than of identity. 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" nameless_bot = channel.dm_address( SenderIdentity(id="7"), sender_kind=SenderKind.AGENT ) @@ -391,6 +390,27 @@ 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) + acknowledged = TelegramOutboundChannel._acknowledged_thread_id( + {"message_id": 5, "chat": {"id": 7, "type": "private"}}, opened_by_name + ) + assert acknowledged == "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..f77998ea 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,13 @@ 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}", + delivery_handle=(sender.name if sender_kind is SenderKind.AGENT else None), ) async def start(self, *, timeout: float) -> None: @@ -208,7 +209,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 addressed by handle answers under the peer's id + provider_thread_id=self.delivered_thread_ids.get( + request.provider_thread_id, 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..41f6e6b7 100644 --- a/tests/support/src/bcn_test_support/storage.py +++ b/tests/support/src/bcn_test_support/storage.py @@ -334,6 +334,22 @@ async def get_channel_session( ) -> ChannelSession | None: return self._storage.channel_sessions.get(channel_session_id) + async def rebind_channel_session( + self, + channel_session_id: str, + *, + provider_thread_id: str, + updated_at_ms: int, + ) -> None: + session = self._storage.channel_sessions.get(channel_session_id) + if session is None: + return + self._storage.channel_sessions[channel_session_id] = replace( + session, + provider_thread_id=provider_thread_id, + updated_at_ms=updated_at_ms, + ) + async def get_thread(self, thread_id: str) -> Thread | None: session = self._storage.threads.get(thread_id) if session is None or not self._in_scope(session): From 0db548fc3568c58a45b33128fa92b7ca47a8e7ea Mon Sep 17 00:00:00 2001 From: Hanchin Hsieh Date: Thu, 10 Sep 2026 00:13:20 -0400 Subject: [PATCH 3/3] fix: name every DM by the peer it belongs to Co-Authored-By: Claude Opus 5 --- .../contrib/sqlite/migrations/registry.py | 2 + .../migrations/v27_name_dms_by_the_peer_id.py | 168 +++++++++++++++++ tests/contrib/test_sqlite_database.py | 169 +++++++++++++++++- 3 files changed, 335 insertions(+), 4 deletions(-) create mode 100644 src/bazaar_compute_node/contrib/sqlite/migrations/v27_name_dms_by_the_peer_id.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..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..0134065c --- /dev/null +++ b/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"] diff --git a/tests/contrib/test_sqlite_database.py b/tests/contrib/test_sqlite_database.py index 1b6cf46a..85bbf2a8 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,168 @@ 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 exists twice; mika has not answered yet, so + # the DM opened by handle is the only one 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:@kana", + "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 side 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, provider_identity_ref_json " + "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 = 'dm-to-kana'" + ) + cursors = await session.fetchall( + "SELECT thread_id, delivered_through_seq FROM consumer_cursors " + "ORDER BY thread_id" + ) + + # 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 under a handle is gone, folded into the peer's own + assert "channel-kana-handle" not in named + # a chat that was never opened keeps the name that opens it + renamed = next(row for row in sessions if row["id"] == "channel-mika-handle") + assert '"delivery_handle":"mika"' in renamed[ + "provider_identity_ref_json" + ].replace(" ", "") + thread_ids = {row["id"] for row in threads} + assert "thread-kana-chat" in thread_ids + assert "thread-mika-handle" 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 + assert [ + (row["thread_id"], row["delivered_through_seq"]) for row in cursors + ] == [("thread-kana-chat", 4)] finally: await database.stop(timeout=2) @@ -1291,7 +1452,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 +1561,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