"""Client-side injection of summary messages rendered as RichMessage. A summary is not a real Telegram message: it is a locally built `TLRPC.TL_message` carrying `rich_message`, attributed to a mock sender that exists only in this client's `MessagesController` cache. It is inserted into the open `ChatActivity`'s message list right after the newest analyzed message. Verified against exteraGram 12.5.x (dex inspection): - `Message.rich_message` + `flags2 |= 1 << 12` makes `MessageObject.setType` return 36, which is the gate for `RichMessageLayout`. - `RichMessage.part` makes the layout fade its bottom edge and hide the timestamp, i.e. the app's own "still streaming" presentation. - The mock sender id MUST be positive: negative ids are chat/channel space, so `getUser()` can never resolve one and the cell falls back to "DELETED". """ from typing import Any, Optional import diagnostics from services.rich_markdown import build_rich_message # Above Telegram's issued id range, so this user exists only locally. MOCK_SENDER_ID = 8000000000000001 MOCK_SENDER_NAME = "AI Summary" MOCK_SENDER_USERNAME = "ai_summary" FLAGS2_RICH_MESSAGE = 1 << 12 # Telegram's animated robot custom emoji, used as the mock sender's status. ROBOT_EMOJI_DOCUMENT_ID = 5397634513026043777 def _tlrpc(): from org.telegram.tgnet import TLRPC return TLRPC def virtual_id_for(summary_id: int, anchor_msg_id: int) -> int: """Deterministic negative id, so re-injection is idempotent and detectable.""" return -abs(int(anchor_msg_id) * 1000 + (int(summary_id) % 1000)) or -1 def register_mock_sender(account: int) -> Optional[Any]: """Puts the mock sender into the local user cache so the cell renders a name.""" try: TLRPC = _tlrpc() from org.telegram.messenger import MessagesController user = TLRPC.TL_user() user.id = MOCK_SENDER_ID user.first_name = MOCK_SENDER_NAME user.last_name = "" user.username = MOCK_SENDER_USERNAME user.bot = True user.verified = True user.deleted = False user.access_hash = 0 _attach_robot_status(user, account, TLRPC) MessagesController.getInstance(account).putUser(user, False) return user except Exception as exc: diagnostics.record_error("injector.register_mock_sender", exc) return None def _attach_robot_status(user: Any, account: int, TLRPC: Any) -> None: try: from org.telegram.messenger import AnimatedEmojiDrawable if AnimatedEmojiDrawable.findDocument(account, ROBOT_EMOJI_DOCUMENT_ID) is None: return status = TLRPC.TL_emojiStatus() status.document_id = ROBOT_EMOJI_DOCUMENT_ID user.emoji_status = status except Exception: # Purely cosmetic; a missing emoji pack must not break injection. pass def peer_for_dialog(dialog_id: int) -> Any: """Builds the Peer subtype matching the dialog kind. The subtype matters: `MessageObject.isOutOwner` inspects it, and a `TL_peerUser` in a group makes the summary render as an outgoing message. """ TLRPC = _tlrpc() dialog_id = int(dialog_id) if dialog_id > 0: peer = TLRPC.TL_peerUser() peer.user_id = dialog_id return peer if dialog_id <= -1000000000000: peer = TLRPC.TL_peerChannel() peer.channel_id = -dialog_id - 1000000000000 return peer peer = TLRPC.TL_peerChat() peer.chat_id = -dialog_id return peer def build_summary_message( account: int, dialog_id: int, topic_id: int, anchor_msg_id: int, virtual_id: int, answer_markdown: str, thinking: str = "", streaming: bool = False, status: str = "", ) -> Optional[Any]: """Builds the local MessageObject for a summary, or None on failure.""" try: import time TLRPC = _tlrpc() from org.telegram.messenger import MessageObject msg = TLRPC.TL_message() msg.id = int(virtual_id) msg.dialog_id = int(dialog_id) msg.date = int(time.time()) msg.out = False msg.unread = False msg.post = False msg.flags = 0 msg.message = "" msg.peer_id = peer_for_dialog(dialog_id) sender = TLRPC.TL_peerUser() sender.user_id = MOCK_SENDER_ID msg.from_id = sender # Reply header points at the newest analyzed message, so the summary # visually attaches to the range it covers and taps navigate there. if anchor_msg_id: try: header = TLRPC.TL_messageReplyHeader() header.reply_to_msg_id = int(anchor_msg_id) if topic_id: header.reply_to_top_id = int(topic_id) msg.reply_to = header except Exception as exc: diagnostics.record_error("injector.reply_header", exc) msg.rich_message = build_rich_message(answer_markdown, thinking, streaming, status) msg.flags2 = int(getattr(msg, "flags2", 0) or 0) | FLAGS2_RICH_MESSAGE message_object = MessageObject(account, msg, True, True) # Pin attribution rather than let it be derived: in Saved Messages # isOutOwner() returns true for anything without forward info. message_object.isOutOwnerCached = False return message_object except Exception as exc: diagnostics.record_error("injector.build_summary_message", exc) return None def get_messages_list(chat_activity: Any) -> Optional[Any]: """Returns the ChatActivity's live ArrayList, or None.""" try: from hook_utils import get_private_field candidate = getattr(chat_activity, "messages", None) or get_private_field(chat_activity, "messages") if candidate is not None and hasattr(candidate, "size"): return candidate diagnostics.note( f"injector.get_messages_list: unusable candidate={type(candidate).__name__ if candidate is not None else None}" ) except Exception as exc: diagnostics.record_error("injector.get_messages_list", exc) return None def get_chat_adapter(chat_activity: Any) -> Optional[Any]: try: from hook_utils import get_private_field return getattr(chat_activity, "chatAdapter", None) or get_private_field(chat_activity, "chatAdapter") except Exception: return None def find_index_for_anchor(messages_list: Any, anchor_msg_id: int, require_anchor: bool = True) -> int: """Index at which to insert so the summary sits directly after the anchor. ChatActivity.messages is newest-first, so "directly after the anchor" chronologically means "immediately before it" in list order. Returns -1 when the anchor message is not loaded in this view and `require_anchor` is set. Positioning by id order alone is not safe: with a non-zero offset, or a range that has since paged out, the nearest-by-id slot can sit among messages the summary never analyzed, which is how cards ended up attached to the wrong message. Callers skip the insert instead and try again once the anchor is paged in. """ anchor_msg_id = int(anchor_msg_id) try: size = messages_list.size() for i in range(size): owner = getattr(messages_list.get(i), "messageOwner", None) if owner is not None and int(getattr(owner, "id", 0) or 0) == anchor_msg_id: return i if require_anchor: return -1 # Anchor absent: insert before the first real message whose id is at # most the anchor's, i.e. the newest message not newer than the # analyzed range. Injected cards carry negative ids and are skipped so # they cannot drag the position around. for i in range(size): owner = getattr(messages_list.get(i), "messageOwner", None) msg_id = int(getattr(owner, "id", 0) or 0) if owner is not None else 0 if 0 < msg_id <= anchor_msg_id: return i # Every loaded message is newer than the anchor, so the analyzed range # ends above everything here: the card belongs at the very top. return size except Exception as exc: diagnostics.record_error("injector.find_index_for_anchor", exc) return -1 def find_existing(messages_list: Any, virtual_id: int) -> int: """Index of an already-injected message with this virtual id, else -1.""" try: size = messages_list.size() for i in range(size): owner = getattr(messages_list.get(i), "messageOwner", None) if owner is not None and int(getattr(owner, "id", 0) or 0) == int(virtual_id): return i except Exception: pass return -1 def notify_adapter(chat_activity: Any) -> None: adapter = get_chat_adapter(chat_activity) if adapter is None: return def run(): try: adapter.notifyDataSetChanged() except Exception as exc: diagnostics.record_error("injector.notifyDataSetChanged", exc) try: from android_utils import run_on_ui_thread run_on_ui_thread(run) except Exception: run() class SummaryInjection: """A single injected summary message, updatable in place while streaming.""" def __init__(self, chat_activity: Any, account: int, dialog_id: int, topic_id: int, anchor_msg_id: int, virtual_id: int): self.chat_activity = chat_activity self.account = int(account) self.dialog_id = int(dialog_id) self.topic_id = int(topic_id) self.anchor_msg_id = int(anchor_msg_id) self.virtual_id = int(virtual_id) self.message_object: Optional[Any] = None # Every id this injection has ever placed in the list. reanchor() # changes the id mid-flight, so an update must be able to find and # remove the card under its previous id instead of adding a second one. self._used_ids = {int(virtual_id)} def insert( self, answer_markdown: str = "", thinking: str = "", streaming: bool = True, status: str = "", ) -> bool: messages_list = get_messages_list(self.chat_activity) if messages_list is None: diagnostics.note("injector: ChatActivity messages list not found") return False register_mock_sender(self.account) message_object = build_summary_message( self.account, self.dialog_id, self.topic_id, self.anchor_msg_id, self.virtual_id, answer_markdown, thinking, streaming, status, ) if message_object is None: return False self.message_object = message_object # The index must be computed on the UI thread, immediately before the # mutation that uses it. Computing it here would race: loading a large # range takes long enough for paging or an incoming message to shift the # list, and the stale index then lands the card after the wrong message. def run(): try: messages_list_now = get_messages_list(self.chat_activity) if messages_list_now is None: return existing = find_existing(messages_list_now, self.virtual_id) if existing >= 0: messages_list_now.set(existing, message_object) else: index = find_index_for_anchor(messages_list_now, self.anchor_msg_id) if index < 0: # Anchor not loaded in this view: placing the card by id # order alone would put it next to unrelated messages. diagnostics.note( f"injector: anchor {self.anchor_msg_id} not loaded, skipping insert" ) return messages_list_now.add(index, message_object) adapter = get_chat_adapter(self.chat_activity) if adapter is not None: adapter.notifyDataSetChanged() # The analyzed range can end far up the history, so the card # may be injected off-screen. Bring it into view once, on the # first insert only -- scrolling on every streaming update # would fight the user for control of the list. if existing < 0: self.scroll_into_view() except Exception as exc: diagnostics.record_error("injector.insert", exc) _on_ui(run) return True def scroll_into_view(self) -> bool: """Scrolls the chat so the summary card is visible. Targets the anchor, not the card: the anchor is a real message the app can resolve and load, and the card sits directly above it, so both end up on screen. ChatActivity.scrollToMessageId(id, fromMessageId, select, loadIndex, forceScroll, unreadCount) -- arity verified in the APK. """ if not self.anchor_msg_id: return False try: self.chat_activity.scrollToMessageId( int(self.anchor_msg_id), 0, False, 0, True, 0 ) return True except Exception as exc: diagnostics.record_error("injector.scroll_into_view", exc) return False def update( self, answer_markdown: str = "", thinking: str = "", streaming: bool = True, status: str = "", ) -> None: """Re-renders the card in place, without rebuilding the whole list. Mutating rich_message alone is not enough: MessageObject caches messageText, type, and its measured layout, and ChatMessageCell keeps its own cached sizes, so a bare swap renders as an empty bubble. The app's own live-update path (ChatActivity.applyFullRichMessageLocally) does exactly three things: set rich_message, null richLayout, then call chatMessageCellDelegate.forceUpdate(cell, true, true). That repaints the single affected cell and is what keeps a streaming card visually stable. notifyDataSetChanged() rebinds every row instead, which is what made the card flicker and lose scroll position on each delta; it is only used as a fallback when the live cell cannot be found. """ if self.message_object is None: self.insert(answer_markdown, thinking, streaming, status) return replacement = build_summary_message( self.account, self.dialog_id, self.topic_id, self.anchor_msg_id, self.virtual_id, answer_markdown, thinking, streaming, status, ) if replacement is None: return self.message_object = replacement def run(): try: messages_list = get_messages_list(self.chat_activity) inserted = False if messages_list is not None: # Replace at the current id; drop any card left behind # under an id this injection used earlier, so a reanchor # mid-stream cannot leave two cards on screen. index = find_existing(messages_list, self.virtual_id) if index >= 0: messages_list.set(index, replacement) else: index = find_index_for_anchor(messages_list, self.anchor_msg_id) if index < 0: # Anchor paged out mid-stream. Keep the streamed # text; it is re-inserted once the anchor is back. return messages_list.add(index, replacement) inserted = True for stale_id in self._used_ids: if stale_id == self.virtual_id: continue stale = find_existing(messages_list, stale_id) if stale >= 0: messages_list.remove(stale) inserted = True # A row was added or removed, so the adapter's item count # changed and only a full rebind is correct. if inserted or not self._force_update_cell(replacement): adapter = get_chat_adapter(self.chat_activity) if adapter is not None: adapter.notifyDataSetChanged() except Exception as exc: diagnostics.record_error("injector.update", exc) _on_ui(run) def _force_update_cell(self, message_object: Any) -> bool: """Repaints just this card's cell. Returns False if it is not on screen. Mirrors ChatActivity.applyFullRichMessageLocally: locate the live ChatMessageCell bound to this message and hand it to chatMessageCellDelegate.forceUpdate, which remeasures and redraws that one row instead of rebinding the whole list. """ try: from hook_utils import get_private_field delegate = getattr(self.chat_activity, "chatMessageCellDelegate", None) or get_private_field( self.chat_activity, "chatMessageCellDelegate" ) list_view = getattr(self.chat_activity, "chatListView", None) or get_private_field( self.chat_activity, "chatListView" ) if delegate is None or list_view is None: return False for i in range(list_view.getChildCount()): child = list_view.getChildAt(i) if child is None or not hasattr(child, "getMessageObject"): continue bound = child.getMessageObject() owner = getattr(bound, "messageOwner", None) if bound is not None else None if owner is None or int(getattr(owner, "id", 0) or 0) != self.virtual_id: continue # setMessageObject(messageObject, groupedMessages, pinnedBottom, # pinnedTop, firstInChat) -- verified arity against the APK. child.setMessageObject(message_object, None, False, False, False) delegate.forceUpdate(child, False, True) child.invalidate() return True except Exception as exc: diagnostics.record_error("injector._force_update_cell", exc) return False def reanchor(self, anchor_msg_id: int, virtual_id: int) -> None: """Re-targets an already-visible card once the analyzed range is known. The card is injected on click, before any messages are fetched, so both the anchor and the DB row id only become known later. Moving it keeps the promise that a summary sits directly after its newest analyzed message. """ anchor_msg_id = int(anchor_msg_id) virtual_id = int(virtual_id) if anchor_msg_id == self.anchor_msg_id and virtual_id == self.virtual_id: return message_object = self.message_object old_virtual_id = self.virtual_id self.anchor_msg_id = anchor_msg_id self.virtual_id = virtual_id self._used_ids.add(virtual_id) if message_object is None: return def run(): try: owner = message_object.messageOwner messages_list = get_messages_list(self.chat_activity) # Locate the card by its OLD id before rewriting it, otherwise # the lookup misses and the card is never moved. current = find_existing(messages_list, old_virtual_id) if messages_list is not None else -1 owner.id = virtual_id try: header = _tlrpc().TL_messageReplyHeader() header.reply_to_msg_id = anchor_msg_id if self.topic_id: header.reply_to_top_id = int(self.topic_id) owner.reply_to = header except Exception as exc: diagnostics.record_error("injector.reanchor.reply_header", exc) if messages_list is not None and current >= 0: messages_list.remove(current) # A card may already sit under the new id -- e.g. the # restore hook re-injected this summary's saved row. Claim # that id rather than ending up with two cards. duplicate = find_existing(messages_list, virtual_id) if duplicate >= 0: messages_list.remove(duplicate) # The card was just removed, so it must go back somewhere: # the new anchor came from the loaded list and is normally # present, but if it paged out in between, fall back to id # ordering rather than dropping the card entirely. target = find_index_for_anchor(messages_list, anchor_msg_id) if target < 0: target = find_index_for_anchor( messages_list, anchor_msg_id, require_anchor=False ) messages_list.add(max(target, 0), message_object) adapter = get_chat_adapter(self.chat_activity) if adapter is not None: adapter.notifyDataSetChanged() # The card just moved to the real end of the analyzed range, # which may be somewhere else entirely, so follow it there. self.scroll_into_view() except Exception as exc: diagnostics.record_error("injector.reanchor", exc) _on_ui(run) def _on_ui(fn) -> None: try: from android_utils import run_on_ui_thread run_on_ui_thread(fn) except Exception: fn() def loaded_message_ids(messages_list: Any) -> set: """Ids currently loaded in the view, in one pass. Restoring used to scan the whole list once per saved summary; a single set keeps it linear in the list length no matter how many summaries exist. """ ids = set() try: for i in range(messages_list.size()): owner = getattr(messages_list.get(i), "messageOwner", None) if owner is not None: ids.add(int(getattr(owner, "id", 0) or 0)) except Exception as exc: diagnostics.record_error("injector.loaded_message_ids", exc) return ids def restore_saved_summaries(chat_activity: Any, account: int, dialog_id: int, topic_id: int, summaries) -> int: """Re-injects persisted summaries into a freshly opened chat. Only summaries whose anchor is currently loaded are injected: a card whose anchor is outside the loaded window has no correct position, and guessing one by id order is what stacked every saved card together at the bottom of the chat. The rest are picked up on a later call, once paging has brought their anchor in. Returns how many were dispatched for insertion. """ messages_list = get_messages_list(chat_activity) if messages_list is None or not summaries: return 0 loaded = loaded_message_ids(messages_list) pending = [] for summary in summaries: virtual_id = virtual_id_for(summary.id, summary.anchor_msg_id) if virtual_id in loaded: continue if int(summary.anchor_msg_id) not in loaded: continue message_object = build_summary_message( account, dialog_id, topic_id, summary.anchor_msg_id, virtual_id, summary.text, summary.thinking, False, ) if message_object is not None: pending.append((summary.anchor_msg_id, virtual_id, message_object)) if not pending: return 0 register_mock_sender(account) # Newest anchor first: each insert shifts everything below it, so placing # the newest card first keeps the remaining lookups pointing at unmoved # rows. The index is recomputed per card against the live list anyway. pending.sort(key=lambda item: item[0], reverse=True) def run(): try: # Re-resolve on the UI thread: the list this was queued against may # have paged or been replaced in between. live = get_messages_list(chat_activity) if live is None: return added = 0 for anchor_msg_id, virtual_id, message_object in pending: if find_existing(live, virtual_id) >= 0: continue index = find_index_for_anchor(live, anchor_msg_id) if index < 0: continue live.add(index, message_object) added += 1 if added: adapter = get_chat_adapter(chat_activity) if adapter is not None: adapter.notifyDataSetChanged() except Exception as exc: diagnostics.record_error("injector.restore", exc) _on_ui(run) return len(pending)