chatSummaries/tools/verify_bundle.py

#!/usr/bin/env python3
"""Offline verification harness for a built .plugin bundle. Runs without a phone.

Usage: python3 tools/verify_bundle.py <bundle.plugin>
"""
from __future__ import annotations

import ast
import json
import sys
import tempfile
import types
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

REPO_ROOT = Path(__file__).resolve().parent.parent
PLUGIN_JSON = REPO_ROOT / "plugin.json"

_FAILURES = []


def check(name: str, condition: bool, detail: str = "") -> None:
    if condition:
        print(f"PASS {name}")
    else:
        message = f"FAIL {name}"
        if detail:
            message += f": {detail}"
        print(message, file=sys.stderr)
        _FAILURES.append(name)


LOG_LINES: List[str] = []
MENU_ITEMS: List[Any] = []
BULLETINS: List[tuple] = []
HOOKS_INSTALLED: List[Any] = []
LAST_REQUEST: Dict[str, Any] = {}


class MenuItemType:
    MESSAGE_CONTEXT_MENU = "MESSAGE_CONTEXT_MENU"
    DRAWER_MENU = "DRAWER_MENU"
    MAIN_MENU = "MAIN_MENU"
    CHAT_ACTION_MENU = "CHAT_ACTION_MENU"
    PROFILE_ACTION_MENU = "PROFILE_ACTION_MENU"


class MenuItemData:
    def __init__(self, menu_type=None, text="", subtext="", icon=None, on_click=None, **kwargs):
        self.menu_type = menu_type
        self.text = text
        self.subtext = subtext
        self.icon = icon
        self.on_click = on_click
        self.__dict__.update(kwargs)


class MethodHook:
    def before_hooked_method(self, param):
        pass

    def after_hooked_method(self, param):
        pass


class BasePlugin:
    def __init__(self):
        self._settings: Dict[str, Any] = {}

    def get_setting(self, key, default=None):
        return self._settings.get(key, default)

    def set_setting(self, key, value, reload_settings=False):
        self._settings[key] = value

    def add_menu_item(self, item):
        MENU_ITEMS.append(item)
        return f"menu_item_{len(MENU_ITEMS)}"

    def hook_method(self, method, hook_instance, priority=0):
        HOOKS_INSTALLED.append((method, hook_instance))
        return f"hook_ref_{len(HOOKS_INSTALLED)}"

    def unhook_method(self, ref):
        pass

    def on_plugin_load(self):
        pass

    def on_plugin_unload(self):
        pass

    def create_settings(self):
        return []


def fake_log(*args, **kwargs):
    LOG_LINES.append(" ".join(str(a) for a in args))


class _SettingsRow:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


class _JavaList(list):
    """Python list that also answers the Java ArrayList calls our code makes."""

    def add(self, *args):
        if len(args) == 2:
            self.insert(int(args[0]), args[1])
        else:
            self.append(args[0])
        return True

    def size(self):
        return len(self)

    def get(self, index):
        return self[index]

    def set(self, index, value):
        previous = self[index]
        self[index] = value
        return previous

    def remove(self, index):
        return self.pop(int(index))


class _JavaArrayList:
    """A Java ArrayList as Chaquopy exposes it: size()/get() but NOT iterable.

    list()/for-in over one raises TypeError on-device, so the fake withholds
    the iterator protocol to keep that mistake failing here.
    """

    def __init__(self, items=()):
        self._items = list(items)

    def size(self):
        return len(self._items)

    def get(self, index):
        return self._items[index]

    def add(self, item):
        self._items.append(item)
        return True


class _TLObj:
    """Duck-typed stand-in for a TLRPC/TL_iv object: any attribute is settable.

    `blocks`/`items`/`texts` start as list-likes because the production code
    calls `.add()` on them without constructing them first, exactly as the Java
    classes pre-initialize their ArrayList fields.
    """

    def __init__(self, tl_name):
        self.tl_name = tl_name
        self.blocks = _JavaList()
        self.items = _JavaList()
        self.texts = _JavaList()
        self.entities = _JavaList()
        self.flags = 0
        self.flags2 = 0

    def __repr__(self):
        return f"<{self.tl_name}>"


def _tl_iv_factory(name):
    return lambda: _TLObj(name)


class _FakeMessageObject:
    """Mirrors the MessageObject fields/behaviour the injector relies on."""

    TYPE_RICH_MESSAGE = 36

    def __init__(self, account, message, generate_layout=False, check_media=False):
        self.currentAccount = account
        self.messageOwner = message
        self.richLayout = None
        self.isOutOwnerCached = None
        self.type = None
        self.messageText = ""
        self.updateMessageText()
        self.setType()

    def updateMessageText(self):
        rich = getattr(self.messageOwner, "rich_message", None)
        self.messageText = "<rich>" if rich is not None else getattr(self.messageOwner, "message", "")

    def setType(self):
        rich = getattr(self.messageOwner, "rich_message", None)
        self.type = self.TYPE_RICH_MESSAGE if rich is not None else 0

    def isOutOwner(self):
        if self.isOutOwnerCached is not None:
            return self.isOutOwnerCached
        return bool(getattr(self.messageOwner, "out", False))

    def getId(self):
        return getattr(self.messageOwner, "id", 0)


class _FakeMessagesController:
    """Per-account user cache; rejects non-positive ids the way getUser does."""

    _instances = {}

    def __init__(self):
        self.users = {}

    @classmethod
    def getInstance(cls, account):
        return cls._instances.setdefault(account, cls())

    def putUser(self, user, from_cache):
        self.users[int(user.id)] = user

    def getUser(self, user_id):
        user_id = int(user_id)
        if user_id <= 0:
            return None
        return self.users.get(user_id)


class _FakeRequestCallback:
    """Mirrors client_utils.RequestCallback: the Java-listener wrapper.
    On-device, send_request hands its second argument to Java. A bare Python
    function is accepted but never invoked, so the response is silently
    dropped; only a RequestCallback-wrapped callable runs. The fake therefore
    refuses anything else, so that mistake fails here instead of on the phone.
    """

    def __init__(self, fn):
        self.fn = fn

    def run(self, response, error):
        return self.fn(response, error)


def _fake_send_request(request, callback, *, account=None):
    if not isinstance(callback, _FakeRequestCallback):
        raise TypeError(
            "send_request callback must be wrapped in RequestCallback(...); "
            f"got {type(callback).__name__}"
        )
    callback.run(None, None)
    return 1


_KNOWN_METHODS = {
    "org.telegram.ui.LaunchActivity": ["dispatchTouchEvent"],
    "org.telegram.messenger.MessagesController": ["deleteMessages", "deleteMessages"],
    "org.telegram.ui.ChatActivity": [
        "onPageDownClicked",
        "updateMessagesVisiblePart",
        "didReceivedNotification_messagesDidLoad",
    ],
}
_KNOWN_CONSTRUCTOR_COUNTS = {
    "org.telegram.ui.Cells.ChatUnreadCell": 1,
}


class _FakeJavaMethod:
    """Simulates a resolved java.lang.reflect.Method for a known method name."""

    def __init__(self, class_name, method_name):
        self.class_name = class_name
        self.method_name = method_name

    def getName(self):
        return self.method_name


class _FakeJavaConstructor:
    def __init__(self, class_name):
        self.class_name = class_name


class _FakeJavaReflectClass:
    """Simulates the real java.lang.Class object returned by .getClass()."""

    def __init__(self, name):
        self._name = name

    def getDeclaredMethods(self):
        return [_FakeJavaMethod(self._name, m) for m in _KNOWN_METHODS.get(self._name, [])]

    def getMethods(self):
        return self.getDeclaredMethods()

    def getDeclaredConstructors(self):
        count = _KNOWN_CONSTRUCTOR_COUNTS.get(self._name, 0)
        return [_FakeJavaConstructor(self._name) for _ in range(count)]


class _FakeChaquopyClassWrapper:
    """Simulates what hook_utils.find_class(...) actually returns on-device:
    a Chaquopy Python-side class wrapper, NOT a java.lang.Class. It exposes
    no getDeclaredMethod(s) directly -- only .getClass() reaches the real
    reflection object. A fake that skipped this distinction (returning None,
    or returning something with getDeclaredMethods directly) would have
    hidden the real on-device bug where code called getDeclaredMethod(...)
    straight on find_class(...)'s result.
    """

    def __init__(self, name):
        self._name = name

    def getClass(self):
        return _FakeJavaReflectClass(self._name)


def fake_find_class(name: str):
    return _FakeChaquopyClassWrapper(name)


def fake_get_private_field(obj, field_name):
    return None


class FakeBulletinHelper:
    @staticmethod
    def show_info(message, fragment=None):
        BULLETINS.append(("info", message))

    @staticmethod
    def show_error(message, fragment=None):
        BULLETINS.append(("error", message))

    @staticmethod
    def show_success(message, fragment=None):
        BULLETINS.append(("success", message))


class FakeAlertDialogBuilder:
    ALERT_TYPE_MESSAGE = 0
    ALERT_TYPE_SPINNER = 1
    BUTTON_POSITIVE = 1
    BUTTON_NEGATIVE = 2

    built: List[Any] = []

    def __init__(self, activity, progress_style=0, resources_provider=None):
        self.activity = activity
        self._view = None
        self._positive = None
        self._negative = None
        self.red_buttons = []
        self.shown = False
        FakeAlertDialogBuilder.built.append(self)

    def set_title(self, *a, **k):
        return self

    def set_message(self, *a, **k):
        return self

    def set_view(self, view, height=-2):
        self._view = view
        return self

    def set_positive_button(self, text, listener=None):
        self._positive = listener
        return self

    def set_negative_button(self, text, listener=None):
        self._negative = listener
        return self

    def set_cancelable(self, *a, **k):
        return self

    def create(self):
        return self

    def show(self):
        self.shown = True
        return self

    def make_button_red(self, button_type):
        self.red_buttons.append(button_type)

    def dismiss(self):
        pass


def _java_widget_stub(name):
    def __init__(self, *args, **kwargs):
        self._children = []
        self._text = ""

    def setOrientation(self, *a, **k):
        pass

    def setGravity(self, *a, **k):
        pass

    def setPadding(self, *a, **k):
        pass

    def addView(self, view, *a, **k):
        self._children.append(view)

    def setText(self, text):
        self._text = text

    def getText(self):
        return self._text

    def setHint(self, *a, **k):
        pass

    def setSingleLine(self, *a, **k):
        pass

    def setMinLines(self, *a, **k):
        pass

    def setMaxLines(self, *a, **k):
        pass

    def setTextSize(self, *a, **k):
        pass

    def setTextColor(self, *a, **k):
        pass

    def setHintTextColor(self, *a, **k):
        pass

    def setLineColors(self, *a, **k):
        pass

    def setCursorColor(self, *a, **k):
        pass

    def setBackgroundColor(self, *a, **k):
        pass

    return type(
        name,
        (object,),
        {
            "__init__": __init__,
            "setOrientation": setOrientation,
            "setGravity": setGravity,
            "setPadding": setPadding,
            "addView": addView,
            "setText": setText,
            "getText": getText,
            "setHint": setHint,
            "setSingleLine": setSingleLine,
            "setMinLines": setMinLines,
            "setMaxLines": setMaxLines,
            "setTextSize": setTextSize,
            "setTextColor": setTextColor,
            "setHintTextColor": setHintTextColor,
            "setLineColors": setLineColors,
            "setCursorColor": setCursorColor,
            "setBackgroundColor": setBackgroundColor,
        },
    )


def install_fakes() -> None:
    base_plugin_mod = types.ModuleType("base_plugin")
    base_plugin_mod.BasePlugin = BasePlugin
    base_plugin_mod.MenuItemData = MenuItemData
    base_plugin_mod.MenuItemType = MenuItemType
    base_plugin_mod.MethodHook = MethodHook
    sys.modules["base_plugin"] = base_plugin_mod

    android_utils_mod = types.ModuleType("android_utils")
    android_utils_mod.log = fake_log
    android_utils_mod.run_on_ui_thread = lambda fn, delay=0: fn()
    android_utils_mod.copy_to_clipboard = lambda text: None
    android_utils_mod.OnClickListener = lambda fn: fn
    android_utils_mod.OnLongClickListener = lambda fn: fn
    sys.modules["android_utils"] = android_utils_mod

    hook_utils_mod = types.ModuleType("hook_utils")
    hook_utils_mod.find_class = fake_find_class
    hook_utils_mod.get_private_field = fake_get_private_field
    hook_utils_mod.set_private_field = lambda obj, field, value: False
    sys.modules["hook_utils"] = hook_utils_mod

    client_utils_mod = types.ModuleType("client_utils")
    client_utils_mod.PLUGINS_QUEUE = "pluginsQueue"
    client_utils_mod.run_on_queue = lambda fn, queue=None, delay=0: fn()
    client_utils_mod.get_last_fragment = lambda: None
    client_utils_mod.get_messages_controller = lambda: None
    client_utils_mod.get_selected_account = lambda: 0
    client_utils_mod.RequestCallback = _FakeRequestCallback
    client_utils_mod.send_request = _fake_send_request
    sys.modules["client_utils"] = client_utils_mod

    # summary_db writes a real SQLite file; point it at a scratch dir so the
    # harness exercises the actual schema and queries.
    file_utils_mod = types.ModuleType("file_utils")
    file_utils_mod.get_files_dir = lambda: tempfile.mkdtemp(prefix="chat_summaries_verify_")
    sys.modules["file_utils"] = file_utils_mod

    ui_mod = types.ModuleType("ui")
    ui_mod.__path__ = []
    sys.modules["ui"] = ui_mod

    ui_settings_mod = types.ModuleType("ui.settings")
    for cls_name in ("Header", "Divider", "Switch", "Selector", "Input", "Text", "EditText", "Custom"):
        setattr(ui_settings_mod, cls_name, type(cls_name, (_SettingsRow,), {}))
    sys.modules["ui.settings"] = ui_settings_mod
    ui_mod.settings = ui_settings_mod

    ui_alert_mod = types.ModuleType("ui.alert")
    ui_alert_mod.AlertDialogBuilder = FakeAlertDialogBuilder
    sys.modules["ui.alert"] = ui_alert_mod
    ui_mod.alert = ui_alert_mod

    ui_bulletin_mod = types.ModuleType("ui.bulletin")
    ui_bulletin_mod.BulletinHelper = FakeBulletinHelper
    sys.modules["ui.bulletin"] = ui_bulletin_mod
    ui_mod.bulletin = ui_bulletin_mod

    android_mod = types.ModuleType("android")
    android_mod.__path__ = []
    sys.modules["android"] = android_mod

    android_view_mod = types.ModuleType("android.view")
    android_view_mod.Gravity = type("Gravity", (), {"CENTER": 17})
    android_view_mod.View = type("View", (), {})
    sys.modules["android.view"] = android_view_mod
    android_mod.view = android_view_mod

    class _FakeLayoutParams:
        def __init__(self, *a, **k):
            pass

    android_widget_mod = types.ModuleType("android.widget")
    for widget_name in ("LinearLayout", "TextView", "ScrollView", "FrameLayout", "Button"):
        cls = _java_widget_stub(widget_name)
        cls.VERTICAL = 1
        cls.HORIZONTAL = 0
        cls.LayoutParams = _FakeLayoutParams
        setattr(android_widget_mod, widget_name, cls)
    sys.modules["android.widget"] = android_widget_mod
    android_mod.widget = android_widget_mod

    org_mod = types.ModuleType("org")
    org_mod.__path__ = []
    sys.modules["org"] = org_mod
    org_telegram_mod = types.ModuleType("org.telegram")
    org_telegram_mod.__path__ = []
    sys.modules["org.telegram"] = org_telegram_mod

    tgnet_mod = types.ModuleType("org.telegram.tgnet")

    class _TLRequestStub:
        def __init__(self):
            self.peer = None
            self.offset_id = 0
            self.offset_date = 0
            self.add_offset = 0
            self.limit = 0
            self.max_id = 0
            self.min_id = 0
            self.hash = 0
            self.msg_id = 0

    tgnet_mod.TLRPC = type(
        "TLRPC",
        (),
        {
            "TL_messages_getHistory": staticmethod(_TLRequestStub),
            "TL_messages_getReplies": staticmethod(_TLRequestStub),
            "TL_message": staticmethod(lambda: _TLObj("TL_message")),
            "TL_user": staticmethod(lambda: _TLObj("TL_user")),
            "TL_peerUser": staticmethod(lambda: _TLObj("TL_peerUser")),
            "TL_peerChat": staticmethod(lambda: _TLObj("TL_peerChat")),
            "TL_peerChannel": staticmethod(lambda: _TLObj("TL_peerChannel")),
            "TL_messageReplyHeader": staticmethod(lambda: _TLObj("TL_messageReplyHeader")),
            "TL_emojiStatus": staticmethod(lambda: _TLObj("TL_emojiStatus")),
            "TL_messageEntityHashtag": staticmethod(lambda: _TLObj("TL_messageEntityHashtag")),
        },
    )
    sys.modules["org.telegram.tgnet"] = tgnet_mod
    org_telegram_mod.tgnet = tgnet_mod

    # TL_iv: instant-view block/rich-text constructors used by rich_markdown.
    # Every block type is the same duck-typed bag of attributes; `blocks`,
    # `items`, and `texts` behave like Java ArrayList (.add()).
    tgnet_tl_mod = types.ModuleType("org.telegram.tgnet.tl")
    tgnet_tl_mod.__path__ = []

    _TL_IV_NAMES = (
        "RichMessage textPlain textEmpty textConcat textBold textItalic textStrike "
        "textFixed textUnderline textUrl pageBlockParagraph pageBlockHeading1 "
        "pageBlockHeading2 pageBlockHeading3 pageBlockPreformatted pageBlockBlockquote "
        "pageBlockDivider pageBlockThinking pageBlockList pageBlockOrderedList "
        "pageBlockDetails TL_pageListItemText TL_pageListOrderedItemText textHashtag"
    ).split()
    tgnet_tl_mod.TL_iv = type(
        "TL_iv",
        (),
        {name: staticmethod(_tl_iv_factory(name)) for name in _TL_IV_NAMES},
    )
    sys.modules["org.telegram.tgnet.tl"] = tgnet_tl_mod
    tgnet_mod.tl = tgnet_tl_mod

    messenger_mod = types.ModuleType("org.telegram.messenger")
    messenger_mod.AndroidUtilities = type("AndroidUtilities", (), {"dp": staticmethod(lambda v: v)})
    messenger_mod.R = type("R", (), {})
    messenger_mod.MessageObject = _FakeMessageObject
    messenger_mod.MessagesController = _FakeMessagesController
    messenger_mod.AnimatedEmojiDrawable = type(
        "AnimatedEmojiDrawable", (), {"findDocument": staticmethod(lambda account, doc_id: None)}
    )
    sys.modules["org.telegram.messenger"] = messenger_mod
    org_telegram_mod.messenger = messenger_mod

    ui_pkg_mod = types.ModuleType("org.telegram.ui")
    ui_pkg_mod.__path__ = []
    sys.modules["org.telegram.ui"] = ui_pkg_mod
    org_telegram_mod.ui = ui_pkg_mod

    components_mod = types.ModuleType("org.telegram.ui.Components")
    edit_text_cls = _java_widget_stub("EditTextBoldCursor")
    components_mod.EditTextBoldCursor = edit_text_cls
    components_mod.LayoutHelper = type(
        "LayoutHelper",
        (),
        {
            "MATCH_PARENT": -1,
            "WRAP_CONTENT": -2,
            "createLinear": staticmethod(lambda *a, **k: None),
        },
    )
    sys.modules["org.telegram.ui.Components"] = components_mod
    ui_pkg_mod.Components = components_mod

    actionbar_mod = types.ModuleType("org.telegram.ui.ActionBar")

    class _FakeTheme:
        # A handful of real key_* names used by src/ui/theme.py; getattr on
        # any key_* name returns the name itself (getColor below just
        # echoes a stable fake int derived from it).
        key_dialogBackground = "key_dialogBackground"
        key_dialogTextBlack = "key_dialogTextBlack"
        key_dialogTextGray3 = "key_dialogTextGray3"
        key_windowBackgroundWhiteHintText = "key_windowBackgroundWhiteHintText"
        key_windowBackgroundWhiteBlackText = "key_windowBackgroundWhiteBlackText"
        key_dialogInputField = "key_dialogInputField"
        key_dialogInputFieldActivated = "key_dialogInputFieldActivated"
        key_windowBackgroundWhiteBlueText = "key_windowBackgroundWhiteBlueText"

        @staticmethod
        def getColor(key):
            return -1  # any deterministic int; value is never asserted on

    actionbar_mod.Theme = _FakeTheme
    sys.modules["org.telegram.ui.ActionBar"] = actionbar_mod
    ui_pkg_mod.ActionBar = actionbar_mod

    requests_mod = types.ModuleType("requests")

    # Captures the last request body so checks can assert what was actually
    # sent (system prompt, JSON transcript) rather than trusting the caller.
    LAST_REQUEST.clear()

    class _FakeResponse:
        status_code = 200
        text = "{}"

        def iter_lines(self):
            for line in (
                b'data: {"choices":[{"delta":{"reasoning_content":"Thinking."}}]}',
                b"",
                b'data: {"choices":[{"delta":{"content":"Fake summary."}}]}',
                b"data: [DONE]",
            ):
                yield line

        def close(self):
            pass

    def fake_post(url, headers=None, json=None, timeout=None, stream=False):
        LAST_REQUEST["url"] = url
        LAST_REQUEST["headers"] = headers or {}
        LAST_REQUEST["json"] = json or {}
        return _FakeResponse()

    requests_mod.post = fake_post
    sys.modules["requests"] = requests_mod


KEY_TO_DUNDER = {
    "id": "__id__",
    "name": "__name__",
    "description": "__description__",
    "author": "__author__",
    "version": "__version__",
    "icon": "__icon__",
    "app_version": "__app_version__",
    "sdk_version": "__sdk_version__",
    "requirements": "__requirements__",
}


def expected_dunders(meta: dict) -> Dict[str, Any]:
    result = {}
    for key, dunder in KEY_TO_DUNDER.items():
        if key not in meta:
            continue
        value = meta[key]
        if key in ("description", "icon") and value == "":
            continue
        if key == "requirements" and value == []:
            continue
        result[dunder] = value
    return result


def run_pipeline_checks(plugin, injector, summary_db_mod, chat_activity_cls, message_cls) -> None:
    """Drives the whole Summarize pipeline: fetch -> stream -> card.

    The rule under test: the chat is not touched until the endpoint has
    actually started answering. Every earlier failure is a bulletin only.
    """
    if plugin is None or message_cls is None:
        check("pipeline checks need the plugin class", False, "plugin not loaded")
        return

    dialog_id = -4242
    requests_mod = sys.modules["requests"]
    real_post = requests_mod.post
    cu_mod = sys.modules["client_utils"]
    real_last_fragment = cu_mod.get_last_fragment
    plugin.set_setting("custom_base_url", "https://example.test/v1")

    def fresh_db():
        plugin.summary_db = summary_db_mod.SummaryDb(str(Path(tempfile.mkdtemp()) / "summaries.db"))
        return plugin.summary_db

    def feed(messages=None, error=None):
        plugin.fetcher.fetch_history = lambda d, c, o, t, cb: cb(messages, error)

    def run(activity):
        BULLETINS.clear()
        plugin._run_summary(None, activity, dialog_id, 0, 3, 0, "Summarize")

    analyzed = [
        message_cls("Alice", "a", id=500, sender_id=1),
        message_cls("Bob", "b", id=555, sender_id=2),
        message_cls("Alice", "c", id=520, sender_id=1),
    ]

    class _ErrorResponse:
        status_code = 500
        text = "upstream exploded"

        def close(self):
            pass

    class _BreakingResponse:
        status_code = 200
        text = ""

        def iter_lines(self):
            yield b'data: {"choices":[{"delta":{"content":"Partial"}}]}'
            raise IOError("connection reset")

        def close(self):
            pass

    try:
        # 1. History fetch fails: bulletin only, chat untouched.
        db = fresh_db()
        activity = chat_activity_cls([600, 555, 500])
        feed(error="flood wait")
        run(activity)
        check(
            "a failed history fetch shows a bulletin and leaves the chat untouched",
            activity.ids() == [600, 555, 500]
            and activity.chatLayoutManager.jumps == []
            and activity.scrolled_to == []
            and ("error", "Chat Summaries: flood wait") in BULLETINS,
            f"ids={activity.ids()} bulletins={BULLETINS}",
        )

        feed(messages=[])
        run(activity)
        check(
            "an empty history shows a bulletin and leaves the chat untouched",
            activity.ids() == [600, 555, 500] and any("no messages" in b[1] for b in BULLETINS),
            f"ids={activity.ids()} bulletins={BULLETINS}",
        )

        # 2. Endpoint fails before the first chunk (HTTP error, unreachable
        # host): no card, no scroll, no DB row.
        feed(messages=analyzed)
        requests_mod.post = lambda *a, **k: _ErrorResponse()
        run(activity)
        check(
            "an HTTP error before the first chunk injects nothing",
            activity.ids() == [600, 555, 500]
            and activity.chatLayoutManager.jumps == []
            and any("HTTP 500" in b[1] for b in BULLETINS if b[0] == "error"),
            f"ids={activity.ids()} bulletins={BULLETINS}",
        )

        def unreachable(*a, **k):
            raise ConnectionError("no route to host")

        requests_mod.post = unreachable
        run(activity)
        check(
            "an unreachable endpoint injects nothing and reports the failure",
            activity.ids() == [600, 555, 500]
            and any("request failed" in b[1] for b in BULLETINS if b[0] == "error"),
            f"ids={activity.ids()} bulletins={BULLETINS}",
        )
        check(
            "a failure before the first chunk leaves no row in the store",
            db.stats()[0] == 0,
            f"rows={db.stats()[0]}",
        )

        # 3. Success: the card appears right after the newest analyzed message
        # (555, not the chat's newest 600), the chat jumps to it once, and the
        # finished summary is tagged and persisted.
        requests_mod.post = real_post
        plugin.restore_hook._scope_state["done"].add(("stale",))
        run(activity)
        ids = activity.ids()
        cards = [i for i in ids if i < 0]
        card_index = ids.index(cards[0]) if cards else -1
        check(
            "a successful stream places exactly one card directly after the newest analyzed message",
            len(cards) == 1 and card_index == ids.index(555) - 1,
            f"ids={ids}",
        )
        check(
            "the chat hard-jumps to the card exactly once",
            activity.chatLayoutManager.jumps == [(2 + card_index, 300, False)],
            f"jumps={activity.chatLayoutManager.jumps}",
        )
        final_owner = activity.messages.get(card_index).messageOwner if cards else None
        check(
            "the finished card is tagged #aisummaries and carries the answer",
            final_owner is not None
            and final_owner.message == "Fake summary.\n\n#aisummaries",
            f"message={getattr(final_owner, 'message', None)!r}",
        )
        rows = db.get_for_scope(dialog_id, 0)
        check(
            "the finished summary is saved under the analyzed anchor without the tag",
            len(rows) == 1 and rows[0].anchor_msg_id == 555 and rows[0].text == "Fake summary."
            and cards and cards[0] == injector.virtual_id_for(rows[0].id, 555),
            f"rows={rows}",
        )
        check(
            "saving a summary invalidates the restore cache",
            ("stale",) not in plugin.restore_hook._scope_state["done"],
        )

        # 4. Re-running a failing stream over a saved range keeps the saved text.
        requests_mod.post = lambda *a, **k: _BreakingResponse()
        run(activity)
        rows = db.get_for_scope(dialog_id, 0)
        check(
            "a stream that breaks mid-way does not wipe the previously saved summary",
            len(rows) == 1 and rows[0].text == "Fake summary.",
            f"rows={rows}",
        )
        check(
            "a mid-stream failure reuses the existing card and reports the error on it",
            len([i for i in activity.ids() if i < 0]) == 1
            and "Error:" in (activity.messages.get(card_index).messageOwner.message or "")
            and any("stream failed" in b[1] for b in BULLETINS if b[0] == "error"),
            f"ids={activity.ids()} bulletins={BULLETINS}",
        )

        # A fresh range that breaks mid-way leaves no empty row behind.
        broken_activity = chat_activity_cls([700, 690])
        feed(messages=[message_cls("Alice", "x", id=690, sender_id=1)])
        run(broken_activity)
        check(
            "a fresh stream that breaks mid-way removes its reserved row",
            [r.anchor_msg_id for r in db.get_for_dialog(dialog_id)] == [555],
            f"rows={db.get_for_dialog(dialog_id)}",
        )
        requests_mod.post = real_post

        # 5. Deleting the card through the app's own delete path forgets it.
        card_id = cards[0]
        cu_mod.get_last_fragment = lambda: activity
        delete_hooks = [
            hook for method, hook in HOOKS_INSTALLED
            if getattr(method, "class_name", "") == "org.telegram.messenger.MessagesController"
        ]
        BULLETINS.clear()
        plugin.restore_hook._scope_state["done"].add(("stale",))
        param = types.SimpleNamespace(
            args=[_JavaArrayList([card_id, 600]), None, None, dialog_id, False, 0]
        )
        for hook in delete_hooks:
            hook.before_hooked_method(param)
        check(
            "deleting a summary card removes its row, its card, and says so once",
            db.get_for_dialog(dialog_id) == []
            and card_id not in activity.ids()
            and BULLETINS == [("info", "Summary deleted")]
            and ("stale",) not in plugin.restore_hook._scope_state["done"],
            f"rows={db.get_for_dialog(dialog_id)} ids={activity.ids()} bulletins={BULLETINS}",
        )
        BULLETINS.clear()
        for hook in delete_hooks:
            hook.before_hooked_method(types.SimpleNamespace(args=[_JavaArrayList([600]), None, None, dialog_id]))
        check(
            "deleting ordinary messages never touches the summary store",
            BULLETINS == [],
            f"bulletins={BULLETINS}",
        )

        # 6. Clear all: confirmed through a red-buttoned alert.
        db.save(dialog_id, 0, 555, "one", "", 1)
        db.save(-1, 0, 7, "two", "", 1)
        count, size = db.stats()
        check("stats report row count and a non-zero file size", count == 2 and size > 0, f"{count} {size}")

        class _Fragment:
            def getParentActivity(self):
                return object()

        cu_mod.get_last_fragment = lambda: _Fragment()
        FakeAlertDialogBuilder.built.clear()
        BULLETINS.clear()
        plugin._on_clear_all_click(None)
        dialog = FakeAlertDialogBuilder.built[-1] if FakeAlertDialogBuilder.built else None
        check(
            "Clear all asks for confirmation with a red button before deleting",
            dialog is not None and dialog.shown
            and dialog.red_buttons == [FakeAlertDialogBuilder.BUTTON_POSITIVE]
            and db.stats()[0] == 2,
            f"dialog={getattr(dialog, '__dict__', None)}",
        )
        if dialog is not None and dialog._positive is not None:
            dialog._positive(dialog, FakeAlertDialogBuilder.BUTTON_POSITIVE)
        check(
            "confirming Clear all wipes every summary and reports it",
            db.stats()[0] == 0 and ("success", "Deleted 2 summaries") in BULLETINS,
            f"rows={db.stats()[0]} bulletins={BULLETINS}",
        )
    finally:
        requests_mod.post = real_post
        cu_mod.get_last_fragment = real_last_fragment


def main(argv=None) -> int:
    argv = sys.argv[1:] if argv is None else argv
    if len(argv) != 1:
        print("usage: tools/verify_bundle.py <bundle.plugin>", file=sys.stderr)
        return 1
    bundle_path = Path(argv[0])
    if not bundle_path.is_file():
        print(f"error: bundle not found at {bundle_path}", file=sys.stderr)
        return 1

    install_fakes()
    source = bundle_path.read_text(encoding="utf-8")

    modules_before = set(sys.modules)
    ns = {"__name__": "chat_summaries"}
    exec(compile(source, str(bundle_path), "exec"), ns)
    new_modules = set(sys.modules) - modules_before

    meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
    expected = expected_dunders(meta)
    mismatches = [
        f"{dunder}: expected {value!r}, got {ns.get(dunder)!r}"
        for dunder, value in expected.items()
        if ns.get(dunder) != value
    ]
    check("metadata constants match plugin.json", not mismatches, "; ".join(mismatches))

    tree = ast.parse(source, filename=str(bundle_path))
    top_level_assigned = {
        target.id
        for node in tree.body
        if isinstance(node, ast.Assign)
        for target in node.targets
        if isinstance(target, ast.Name)
    }
    check(
        "metadata exposed as top-level literal assignments (AST)",
        set(expected) <= top_level_assigned,
        f"missing: {set(expected) - top_level_assigned}",
    )

    plugin_classes = [
        value
        for value in ns.values()
        if isinstance(value, type) and issubclass(value, BasePlugin) and value is not BasePlugin
    ]
    check(
        "exactly one BasePlugin subclass defined",
        len(plugin_classes) == 1,
        f"found {len(plugin_classes)}",
    )
    plugin_cls = plugin_classes[0] if len(plugin_classes) == 1 else None

    if plugin_cls is not None:
        check(
            "plugin class __module__ matches executed namespace",
            plugin_cls.__module__ == ns["__name__"],
            f"__module__={plugin_cls.__module__!r} != {ns['__name__']!r}",
        )

        plugin = plugin_cls()
        MENU_ITEMS.clear()
        LOG_LINES.clear()
        plugin.on_plugin_load()

        check(
            "on_plugin_load emits load log line",
            any("chat_summaries loaded" in line for line in LOG_LINES),
            f"log lines: {LOG_LINES}",
        )
        check(
            "on_plugin_load registers Summarize and Copy diagnostics CHAT_ACTION_MENU items",
            len(MENU_ITEMS) == 2
            and all(m.menu_type == MenuItemType.CHAT_ACTION_MENU for m in MENU_ITEMS)
            and {m.text for m in MENU_ITEMS} == {"Summarize", "Copy diagnostics"},
            f"menu items: {[(getattr(m, 'text', None), getattr(m, 'menu_type', None)) for m in MENU_ITEMS]}",
        )
        check(
            "unread_hook installs dispatchTouchEvent, onPageDownClicked, and ChatUnreadCell hooks",
            len(plugin.unread_hook._unhooks) == 3,
            f"installed hooks: {len(plugin.unread_hook._unhooks)} (expected 3)",
        )

        # unread_count decays while a chat is open (markDialogAsRead runs as
        # messages are viewed), so pressing the unread button a second time used
        # to prefill a far smaller range than the first press: 1100 then 15 then
        # 75. The resolved count must not shrink while the chat stays open.
        class _UnreadDialog:
            def __init__(self, count):
                self.unread_count = count

        class _UnreadFragment:
            def __init__(self, divider, ids):
                self.createUnreadMessageAfterId = divider
                self.messages = _JavaList()
                for mid in ids:
                    owner_obj = _TLObj("TL_message")
                    owner_obj.id = mid
                    self.messages.add(_FakeMessageObject(0, owner_obj))

        unread_dialog = _UnreadDialog(1100)
        cu_mod = sys.modules["client_utils"]
        real_controller = cu_mod.get_messages_controller

        class _UnreadController:
            class dialogs_dict:
                @staticmethod
                def get(_dialog_id):
                    return unread_dialog

        cu_mod.get_messages_controller = lambda: _UnreadController
        try:
            unread_fragment = _UnreadFragment(171000, list(range(172700, 172670, -1)))
            resolved = []
            for decayed in (1100, 15, 75):
                unread_dialog.unread_count = decayed
                resolved.append(
                    plugin.unread_hook._resolve_unread_count(unread_fragment, -1978105175)
                )
        finally:
            cu_mod.get_messages_controller = real_controller
        check(
            "the unread prefill does not shrink as messages are marked read",
            resolved == [1100, 1100, 1100],
            f"resolved={resolved}",
        )

        settings_rows = plugin.create_settings()
        settings_keys = {row.key for row in settings_rows if hasattr(row, "key")}
        expected_keys = {
            "custom_base_url",
            "custom_api_key",
            "custom_model",
            "default_count",
            "default_offset",
            "global_prompt",
            "include_attachments",
        }
        check(
            "create_settings exposes endpoint and default settings keys",
            expected_keys == settings_keys,
            f"missing: {expected_keys - settings_keys}; unexpected: {settings_keys - expected_keys}",
        )

        # Per-chat/topic prompt override: falls back to global, then to a
        # saved override once one is set for that specific (dialog, topic).
        dialog_id, topic_id = -100123456, 7
        global_prompt_key = "global_prompt"
        plugin.set_setting(global_prompt_key, "Global default prompt")
        before_override = plugin._get_prompt_for(dialog_id, topic_id)
        check(
            "prompt falls back to global default with no override",
            before_override == "Global default prompt",
            f"got {before_override!r}",
        )
        plugin._save_prompt_for(dialog_id, topic_id, "Chat-specific prompt")
        after_override = plugin._get_prompt_for(dialog_id, topic_id)
        check(
            "per-chat/topic prompt override takes precedence once saved",
            after_override == "Chat-specific prompt",
            f"got {after_override!r}",
        )
        other_topic_prompt = plugin._get_prompt_for(dialog_id, topic_id + 1)
        check(
            "prompt override is scoped per-topic, not just per-chat",
            other_topic_prompt == "Global default prompt",
            f"got {other_topic_prompt!r}",
        )

        # Dispatcher: OpenAI path against the faked requests module. FormattedMessage
        # is a private bundled module (never in sys.modules), so duck-type instead
        # of importing it -- the dispatcher only needs a message with these attrs.
        class _FakeMessage:
            def __init__(self, sender_name, text, id=0, sender_id=0, reply_to_id=0,
                         reply_highlight="", attachments=None):
                self.sender_name = sender_name
                self.text = text
                self.id = id
                self.date = 1700000000
                self.sender_id = sender_id
                self.reply_to_id = reply_to_id
                self.reply_highlight = reply_highlight
                self.attachments = list(attachments or [])

        fake_messages = [_FakeMessage("Alice", "hello", id=10, sender_id=111)]
        plugin.set_setting("custom_base_url", "https://example.test/v1")
        plugin.set_setting("custom_api_key", "sk-test")
        results = {}

        def on_done(answer, thinking, error):
            results["answer"] = answer
            results["error"] = error

        plugin.dispatcher.summarize_streaming(fake_messages, "Summarize", lambda k, t: None, on_done)
        check(
            "dispatcher streams a summary from a faked SSE response",
            results.get("error") is None and results.get("answer") == "Fake summary.",
            f"results: {results}",
        )

        # The request must carry three separate roles, not one concatenated
        # blob: script-owned system prompt, the user's instructions, and the
        # anonymized JSON transcript.
        sent = LAST_REQUEST.get("json") or {}
        roles = [m["role"] for m in sent.get("messages", [])]
        contents = [m["content"] for m in sent.get("messages", [])]
        check(
            "request sends a system prompt, the user prompt, and the transcript",
            roles == ["system", "user", "user"]
            and "anonymized" in contents[0]
            and contents[1] == "Summarize"
            and contents[2].startswith('{"messages":['),
            f"roles={roles} third={contents[2][:60] if len(contents) > 2 else None!r}",
        )
        check(
            "the system prompt documents the mention syntax the model must use",
            "{userN}" in contents[0] and "{msgN}" in contents[0],
        )
        check(
            "streaming is requested so deltas arrive incrementally",
            sent.get("stream") is True,
            f"stream={sent.get('stream')}",
        )

        # Endpoint is mandatory now that there is no built-in provider: an
        # unconfigured base URL must produce a clear error, not a crash.
        plugin.set_setting("custom_base_url", "")
        unconfigured = {}

        def on_unconfigured(answer, thinking, error):
            unconfigured["answer"] = answer
            unconfigured["error"] = error

        plugin.dispatcher.summarize_streaming(
            fake_messages, "Summarize", lambda k, t: None, on_unconfigured
        )
        check(
            "dispatcher reports a clear error when no base URL is configured",
            unconfigured.get("answer") is None
            and "base URL" in (unconfigured.get("error") or ""),
            f"results: {unconfigured}",
        )
        plugin.set_setting("custom_base_url", "https://example.test/v1")

        # MessageFetcher: no controller available -> graceful error, not a crash.
        fetch_results = {}

        def on_fetched(messages, error):
            fetch_results["messages"] = messages
            fetch_results["error"] = error

        plugin.fetcher.fetch_history(dialog_id=dialog_id, count=10, offset=0, callback=on_fetched)
        check(
            "fetch_history reports an error when no messages controller is available",
            fetch_results.get("error") is not None,
            f"fetch_results: {fetch_results}",
        )

        # Telegram returns java.util.ArrayList, which Chaquopy exposes without
        # the iterator protocol: list()/for-in raise TypeError. The fetcher must
        # read it through size()/get(i).
        page_results = {}

        def on_page(messages, error):
            page_results["messages"] = messages
            page_results["error"] = error

        fake_response = _TLObj("TL_messages_messages")
        raw_message = _TLObj("TL_message")
        raw_message.id = 4242
        raw_message.date = 1700000000
        raw_message.message = "hello from java"
        raw_message.from_id = None
        fake_response.messages = _JavaArrayList([raw_message])
        fake_response.users = _JavaArrayList()
        fake_response.chats = _JavaArrayList()

        plugin.fetcher._handle_response(
            fake_response,
            None,
            [],
            on_page,
            dialog_id,
            10,
            0,
            0,
            100,
        )
        check(
            "fetcher reads a non-iterable Java ArrayList response via size()/get()",
            page_results.get("error") is None
            and len(page_results.get("messages") or []) == 1
            and page_results["messages"][0].text == "hello from java",
            f"page_results={page_results}",
        )

        check(
            "delete_hook hooks every MessagesController.deleteMessages overload",
            len(plugin.delete_hook._unhooks) == 2,
            f"installed hooks: {len(plugin.delete_hook._unhooks)} (expected 2)",
        )

        # Storage settings: an approximate size readout and a destructive,
        # confirmed "clear all" action.
        storage_rows = {getattr(r, "text", None): r for r in settings_rows}
        clear_row = storage_rows.get("Clear all summaries")
        size_row = storage_rows.get("Saved summaries")
        check(
            "settings show the saved summary count and approximate size",
            size_row is not None and "saved" in (size_row.subtext or "") and "B" in size_row.subtext,
            f"subtext={getattr(size_row, 'subtext', None)!r}",
        )
        check(
            "settings offer a red Clear all summaries action",
            clear_row is not None and clear_row.red is True and callable(clear_row.on_click),
            f"row={getattr(clear_row, '__dict__', None)}",
        )

    # ------------------------------------------------------------------
    # Bundled-module checks: markdown serialization, SSE streaming, and the
    # SQLite store. Reached through the bundle's private registry, since
    # bundled modules deliberately never enter sys.modules.
    # ------------------------------------------------------------------
    loaded = ns.get("_LOADED") or {}
    bundle_import = ns.get("_bundle_load")

    def bundled(name):
        if name in loaded:
            return loaded[name]
        if bundle_import is not None:
            try:
                return bundle_import(name)
            except Exception:
                return None
        return None

    rich_markdown = bundled("services.rich_markdown")
    if rich_markdown is None:
        check("services.rich_markdown is importable from the bundle", False, "module not found")
    else:
        blocks = rich_markdown.markdown_to_blocks(
            "# Title\n\n"
            "Plain **bold** and `code`.\n\n"
            "- one\n- two\n\n"
            "1. first\n2. second\n\n"
            "> quoted\n\n"
            "---\n\n"
            "```python\nprint(1)\n```\n"
        )
        kinds = [b.tl_name for b in blocks]
        check(
            "markdown_to_blocks maps every markdown construct to its block type",
            kinds
            == [
                "pageBlockHeading1",
                "pageBlockParagraph",
                "pageBlockList",
                "pageBlockOrderedList",
                "pageBlockBlockquote",
                "pageBlockDivider",
                "pageBlockPreformatted",
            ],
            f"got {kinds}",
        )

        code_block = blocks[-1]
        check(
            "fenced code block keeps its language and body",
            code_block.language == "python" and code_block.text.text == "print(1)",
            f"language={code_block.language!r} text={getattr(code_block.text, 'text', None)!r}",
        )

        check(
            "list blocks carry one item per bullet",
            len(blocks[2].items) == 2 and len(blocks[3].items) == 2,
            f"bullet items={len(blocks[2].items)} ordered items={len(blocks[3].items)}",
        )

        check(
            "ordered list items are numbered from the markdown start value",
            [i.num for i in blocks[3].items] == ["1", "2"],
            f"nums={[i.num for i in blocks[3].items]}",
        )

        # Inline styling must nest into a RichText tree, not be dropped.
        inline = rich_markdown.build_rich_text("a **b** c `d` [e](http://x)")
        inline_kinds = [p.tl_name for p in inline.texts]
        check(
            "inline markdown becomes bold/fixed/url RichText nodes",
            "textBold" in inline_kinds and "textFixed" in inline_kinds and "textUrl" in inline_kinds,
            f"kinds={inline_kinds}",
        )

        # Reasoning still arriving (no answer yet) animates as a thinking
        # block; once the answer starts it is finished, so it collapses into a
        # tappable details container that can be revealed and hidden.
        reasoning_only = rich_markdown.build_rich_message("", thinking="reasoning", streaming=True)
        check(
            "reasoning still streaming renders as an animated thinking block",
            [b.tl_name for b in reasoning_only.blocks] == ["pageBlockThinking"],
            f"blocks={[b.tl_name for b in reasoning_only.blocks]}",
        )

        streaming_rm = rich_markdown.build_rich_message("answer", thinking="reasoning", streaming=True)
        details = streaming_rm.blocks[0]
        check(
            "finished reasoning collapses into a closed details block",
            details.tl_name == "pageBlockDetails"
            and details.open is False
            and details.title.text == rich_markdown.THINKING_TITLE
            and len(details.blocks) == 1,
            f"first={details.tl_name} open={getattr(details, 'open', None)}",
        )
        # `part` drives RichMessageLayout's partial branch, which also draws a
        # server-backed "Show more" button; it must stay off for local summaries.
        final_rm = rich_markdown.build_rich_message("answer", thinking="reasoning", streaming=False)
        check(
            "part stays unset so no Show more collapse is drawn",
            streaming_rm.part is False and final_rm.part is False,
            f"streaming={streaming_rm.part} final={final_rm.part}",
        )

        # Stage feedback: the card is injected on click, before any content
        # exists, so a bare status must still render as a shimmer block.
        status_rm = rich_markdown.build_rich_message("", status="Fetching messages…", streaming=True)
        check(
            "status-only RichMessage renders the stage as a thinking block",
            len(status_rm.blocks) == 1 and status_rm.blocks[0].tl_name == "pageBlockThinking",
            f"blocks={[b.tl_name for b in status_rm.blocks]}",
        )
        superseded = rich_markdown.build_rich_message(
            "", thinking="real reasoning", status="Fetching messages…", streaming=True
        )
        check(
            "real reasoning supersedes the status placeholder",
            len(superseded.blocks) == 1
            and superseded.blocks[0].text.text == "real reasoning",
            f"blocks={[b.tl_name for b in superseded.blocks]}",
        )
        answered = rich_markdown.build_rich_message(
            "# Answer", status="Summarizing…", streaming=True
        )
        check(
            "status is dropped once answer text arrives",
            [b.tl_name for b in answered.blocks] == ["pageBlockHeading1"],
            f"blocks={[b.tl_name for b in answered.blocks]}",
        )
        plain = rich_markdown.markdown_to_plain(
            "# Title\n\n**Bold** and *it* with `code` by [Alice](tg://user?id=1)\n> quote\n```\nx = 1\n```"
        )
        check(
            "markdown_to_plain keeps the readable text and drops the markup",
            plain == "Title\n\nBold and it with code by Alice\nquote\nx = 1",
            f"plain={plain!r}",
        )
        empty_rm = rich_markdown.build_rich_message("", streaming=True)
        check(
            "an empty RichMessage still carries one block so the cell renders",
            len(empty_rm.blocks) == 1,
            f"blocks={len(empty_rm.blocks)}",
        )

    transcript_mod = bundled("services.transcript")
    if transcript_mod is None:
        check("services.transcript is importable from the bundle", False, "module not found")
    else:
        class _Msg:
            def __init__(self, id, sender_id, sender_name, text,
                         reply_to_id=0, reply_highlight="", attachments=None,
                         sender_kind="user"):
                self.id = id
                self.date = 1700000000
                self.sender_id = sender_id
                self.sender_name = sender_name
                self.sender_kind = sender_kind
                self.text = text
                self.reply_to_id = reply_to_id
                self.reply_highlight = reply_highlight
                self.attachments = list(attachments or [])

        convo = [
            _Msg(5001, 111, "Alice", "got new crocs", attachments=["randhash.png"]),
            _Msg(5002, 222, "Bob", "cap", reply_to_id=5001, reply_highlight="new"),
            _Msg(5003, 111, "Alice", "fr", reply_to_id=4000),
            _Msg(5004, 1001234567890, "Some Channel", "announcement", sender_kind="chat"),
        ]
        payload, mapping = transcript_mod.build_transcript(convo)
        parsed = json.loads(payload)

        check(
            "transcript numbers messages from 1 and senders from 1 by first appearance",
            [(m["id"], m["user"]) for m in parsed["messages"]] == [(1, 1), (2, 2), (3, 1), (4, 3)],
            f"got {[(m['id'], m['user']) for m in parsed['messages']]}",
        )
        check(
            "transcript is compact JSON with no spacing",
            " " not in payload.replace("got new crocs", "").replace("randhash.png", ""),
            f"payload={payload[:120]}",
        )
        check(
            "reply_to is remapped to the anonymized id and carries the highlight",
            parsed["messages"][1].get("reply_to") == 1
            and parsed["messages"][1].get("reply_highlight") == "new",
            f"entry={parsed['messages'][1]}",
        )
        check(
            "real Telegram ids never reach the transcript",
            "5001" not in payload and "111" not in payload and "4000" not in payload,
            f"payload={payload}",
        )

        # Attachments are opt-in: message text is always sent, file names only
        # when the setting is enabled.
        check(
            "attachments are omitted unless explicitly included",
            "attachments" not in payload,
            f"payload={payload}",
        )
        with_attachments, _ = transcript_mod.build_transcript(convo, include_attachments=True)
        check(
            "attachments are listed when included",
            json.loads(with_attachments)["messages"][0].get("attachments") == ["randhash.png"],
            f"entry={json.loads(with_attachments)['messages'][0]}",
        )

        # A reply whose parent was not analyzed loses meaning, so the parent is
        # injected with a negative id marking it as outside the range.
        parent = _Msg(4000, 333, "Carol", "original point")
        ranged, ranged_map = transcript_mod.build_transcript(convo, extra_by_id={4000: parent})
        ranged_entries = json.loads(ranged)["messages"]
        ids = [m["id"] for m in ranged_entries]
        check(
            "an out-of-range reply parent is injected with a negative id before its child",
            ids == [1, 2, -1, 3, 4] and ranged_entries[2]["text"] == "original point",
            f"ids={ids}",
        )
        check(
            "the injected parent is the reply target of its child",
            ranged_entries[3].get("reply_to") == -1,
            f"entry={ranged_entries[3]}",
        )
        check(
            "out-of-range parents stay resolvable back to their real id",
            ranged_map.real_message_id(-1) == 4000,
            f"mapped={ranged_map.real_message_id(-1)}",
        )

        # The model answers with placeholders; substitution supplies the real
        # display name and turns citations into tappable deep links.
        answer = "{user1} said it broke ({msg1}, {msg2}), {user2} agreed."
        rendered = transcript_mod.deanonymize(answer, mapping, -1001234567890)
        check(
            "user placeholders resolve to the real display name and profile link",
            "[Alice](tg://user?id=111)" in rendered
            and "[Bob](tg://user?id=222)" in rendered,
            f"rendered={rendered}",
        )
        check(
            "message placeholders become numbered links into this dialog",
            "[№2](tg://openmessage?chat_id=1234567890&message_id=5002)" in rendered,
            f"rendered={rendered}",
        )
        check(
            "no placeholder survives substitution",
            "{user" not in rendered and "{msg" not in rendered,
            f"rendered={rendered}",
        )
        check(
            "private-chat message links use user_id rather than chat_id",
            "user_id=777" in transcript_mod.message_link(777, 42),
            transcript_mod.message_link(777, 42),
        )

        # A group or channel sender is not a user: tg://user?id= on one reports
        # "user not found", so chat senders need the chat link form.
        check(
            "a chat/channel sender links to the chat, not a user profile",
            "[Some Channel](tg://openmessage?chat_id=1234567890)"
            in transcript_mod.deanonymize("{user3} posted", mapping, -1),
            transcript_mod.deanonymize("{user3} posted", mapping, -1),
        )
        check(
            "sender_link uses the user form only for real users",
            transcript_mod.sender_link(111, "user") == "tg://user?id=111"
            and transcript_mod.sender_link(1001234567890, "chat")
            == "tg://openmessage?chat_id=1234567890",
            f"user={transcript_mod.sender_link(111, 'user')} "
            f"chat={transcript_mod.sender_link(1001234567890, 'chat')}",
        )

        # tg://openmessage carries no thread parameter, so inside a forum it
        # lands in the supergroup's global feed. The t.me/c form is the only
        # one that opens the right topic.
        check(
            "a topic message link targets the topic, not the supergroup feed",
            transcript_mod.message_link(-1001234567890, 42, 7)
            == "https://t.me/c/1234567890/7/42",
            transcript_mod.message_link(-1001234567890, 42, 7),
        )
        check(
            "a group without a topic still uses the openmessage form",
            transcript_mod.message_link(-1001234567890, 42, 0)
            == "tg://openmessage?chat_id=1234567890&message_id=42",
            transcript_mod.message_link(-1001234567890, 42, 0),
        )
        topic_payload, topic_mapping = transcript_mod.build_transcript(convo, topic_id=9)
        check(
            "the topic travels on the mapping so citations open in-topic",
            topic_mapping.topic_id == 9
            and "/c/1234567890/9/5002"
            in transcript_mod.deanonymize("{msg2}", topic_mapping, -1001234567890),
            transcript_mod.deanonymize("{msg2}", topic_mapping, -1001234567890),
        )
        check(
            "the topic id is not leaked into the transcript payload",
            '"topic' not in topic_payload,
            f"payload={topic_payload[:80]}",
        )
        check(
            "an unknown user placeholder degrades to readable text",
            transcript_mod.deanonymize("{user99} spoke", mapping, -1) == "someone spoke",
            transcript_mod.deanonymize("{user99} spoke", mapping, -1),
        )
        check(
            "an unknown message placeholder is dropped rather than left dangling",
            transcript_mod.deanonymize("see {msg99}", mapping, -1) == "see ",
            transcript_mod.deanonymize("see {msg99}", mapping, -1),
        )

        # Streaming safety: this is why placeholders replaced Markdown links.
        # A half-received token must not render as literal text.
        check(
            "a placeholder still being streamed is suppressed, not shown raw",
            transcript_mod.strip_partial_placeholder("build broke ({msg1") == "build broke (",
            transcript_mod.strip_partial_placeholder("build broke ({msg1"),
        )
        check(
            "complete placeholders earlier in the text survive stripping",
            transcript_mod.strip_partial_placeholder("a {msg1} b {us") == "a {msg1} b ",
            transcript_mod.strip_partial_placeholder("a {msg1} b {us"),
        )
        check(
            "text with no trailing placeholder is left untouched",
            transcript_mod.strip_partial_placeholder("all done.") == "all done.",
        )

        # Reasoning is model-authored text too, so it carries placeholders and
        # must go through the same substitution -- leaving them raw would show
        # the reader the anonymized namespace.
        reasoning = "Looking at {user1} and {msg2} to decide."
        rendered_reasoning = transcript_mod.deanonymize(reasoning, mapping, -1001234567890)
        check(
            "reasoning text is substituted, not left with raw placeholders",
            "{user" not in rendered_reasoning
            and "{msg" not in rendered_reasoning
            and "Alice" in rendered_reasoning
            and "tg://" in rendered_reasoning,
            f"rendered={rendered_reasoning}",
        )

        # Privacy boundary: only text and attachments cross the wire verbatim.
        # Names, ids and topic stay local, while the DB keeps the real thing.
        check(
            "sender names never reach the transcript payload",
            "Alice" not in payload and "Bob" not in payload and "Some Channel" not in payload,
            f"payload={payload}",
        )
        check(
            "the mapping retains real names and ids for local substitution",
            mapping.user_name(1) == "Alice"
            and mapping.real_user_id(1) == 111
            and mapping.real_message_id(2) == 5002
            and mapping.sender_kind(3) == "chat",
            f"name={mapping.user_name(1)} id={mapping.real_user_id(1)} kind={mapping.sender_kind(3)}",
        )
        check(
            "message text and attachment names are the only verbatim fields",
            "got new crocs" in payload
            and "randhash.png" in json.loads(with_attachments)["messages"][0]["attachments"],
            f"payload={payload[:100]}",
        )

    streaming_mod = bundled("providers.streaming")
    if streaming_mod is None:
        check("providers.streaming is importable from the bundle", False, "module not found")
    else:
        # Reasoning fields differ per vendor; all three must be recognised.
        check(
            "reasoning_content and reasoning deltas are classified as thinking",
            streaming_mod._extract_delta({"choices": [{"delta": {"reasoning_content": "r"}}]}) == ("thinking", "r")
            and streaming_mod._extract_delta({"choices": [{"delta": {"reasoning": "r"}}]}) == ("thinking", "r"),
        )
        check(
            "content deltas are classified as answer",
            streaming_mod._extract_delta({"choices": [{"delta": {"content": "c"}}]}) == ("answer", "c"),
        )

        # A <think> tag split across chunks must not leak into the answer.
        splitter = streaming_mod._ThinkTagSplitter()
        got = []
        for piece in ["before<thi", "nk>rea", "soning</thi", "nk>after"]:
            got.extend(splitter.feed(piece))
        got.extend(splitter.flush())
        joined = [
            ("thinking", "".join(t for k, t in got if k == "thinking")),
            ("answer", "".join(t for k, t in got if k == "answer")),
        ]
        check(
            "think tags split across SSE chunks are reassembled correctly",
            joined == [("thinking", "reasoning"), ("answer", "beforeafter")],
            f"got {joined}",
        )

        sse_lines = [
            b'data: {"choices":[{"delta":{"content":"hi"}}]}',
            b"",
            b"data: [DONE]",
            b'data: {"choices":[{"delta":{"content":"never"}}]}',
        ]
        objects = list(streaming_mod.iter_sse_objects(iter(sse_lines)))
        check(
            "SSE parsing stops at [DONE] and ignores blank lines",
            len(objects) == 1,
            f"parsed {len(objects)} objects",
        )

    summary_db_mod = bundled("services.summary_db")
    if summary_db_mod is None:
        check("services.summary_db is importable from the bundle", False, "module not found")
    else:
        db = summary_db_mod.SummaryDb(str(Path(tempfile.mkdtemp()) / "summaries.db"))
        first = db.save(-100, 0, 555, "text one", "thinking one", 42)
        rows = db.get_for_scope(-100, 0)
        check(
            "summary_db persists text, anchor id, and message count",
            len(rows) == 1
            and rows[0].anchor_msg_id == 555
            and rows[0].text == "text one"
            and rows[0].thinking == "thinking one"
            and rows[0].msg_count == 42,
            f"rows={rows}",
        )

        # Re-saving the same anchor must update in place: streaming reserves a
        # row first and rewrites it when the stream finishes.
        second = db.save(-100, 0, 555, "text two", "thinking two", 43)
        rows = db.get_for_scope(-100, 0)
        check(
            "re-saving the same anchor updates the existing row instead of duplicating",
            second == first and len(rows) == 1 and rows[0].text == "text two",
            f"first={first} second={second} rows={rows}",
        )

        db.save(-100, 7, 555, "topic text", "", 1)
        check(
            "summaries are scoped per topic, not just per chat",
            len(db.get_for_scope(-100, 0)) == 1 and len(db.get_for_scope(-100, 7)) == 1,
            f"topic0={len(db.get_for_scope(-100, 0))} topic7={len(db.get_for_scope(-100, 7))}",
        )

        reserved, created = db.reserve(-100, 0, 555, 9)
        check(
            "reserving an anchor that already has a summary keeps its text",
            reserved == first and created is False and db.get(first).text == "text two",
            f"reserved={reserved} created={created} text={db.get(first).text!r}",
        )
        fresh, fresh_created = db.reserve(-100, 0, 777, 9)
        check(
            "reserving a new anchor creates an empty row",
            fresh_created is True and db.get(fresh).text == "",
            f"created={fresh_created}",
        )
        check(
            "get_for_dialog spans every topic of the chat",
            sorted((r.topic_id, r.anchor_msg_id) for r in db.get_for_dialog(-100))
            == [(0, 555), (0, 777), (7, 555)],
            f"rows={db.get_for_dialog(-100)}",
        )

        check("summary_db.delete removes a row", db.delete(first) and db.get(first) is None)

    injector = bundled("services.rich_injector")
    if injector is None:
        check("services.rich_injector is importable from the bundle", False, "module not found")
    else:
        check(
            "mock sender id is positive so getUser can resolve it",
            injector.MOCK_SENDER_ID > 0,
            f"id={injector.MOCK_SENDER_ID}",
        )
        user = injector.register_mock_sender(0)
        from org.telegram.messenger import MessagesController  # noqa: E402

        resolved = MessagesController.getInstance(0).getUser(injector.MOCK_SENDER_ID)
        check(
            "mock sender resolves from the local user cache with a name",
            resolved is not None and resolved.first_name == injector.MOCK_SENDER_NAME,
            f"resolved={resolved}",
        )

        # Peer subtype drives isOutOwner; a wrong type renders as outgoing.
        check(
            "peer_for_dialog picks the peer subtype matching the dialog kind",
            injector.peer_for_dialog(12345).tl_name == "TL_peerUser"
            and injector.peer_for_dialog(-4242).tl_name == "TL_peerChat"
            and injector.peer_for_dialog(-1001234567890).tl_name == "TL_peerChannel",
        )
        check(
            "channel peer id is converted out of dialog-id space",
            injector.peer_for_dialog(-1001234567890).channel_id == 1234567890,
            f"channel_id={injector.peer_for_dialog(-1001234567890).channel_id}",
        )

        message_object = injector.build_summary_message(
            0, -4242, 0, 555, injector.virtual_id_for(1, 555), "# Answer", "reasoning", True
        )
        owner = message_object.messageOwner
        check(
            "summary message carries rich_message with the type-36 flag bit",
            owner.rich_message is not None
            and bool(owner.flags2 & injector.FLAGS2_RICH_MESSAGE)
            and message_object.type == 36,
            f"flags2={owner.flags2} type={message_object.type}",
        )
        check(
            "summary message is attributed to the mock sender and not outgoing",
            owner.from_id.user_id == injector.MOCK_SENDER_ID
            and owner.out is False
            and message_object.isOutOwner() is False,
        )
        check(
            "summary message replies to the newest analyzed message",
            owner.reply_to.reply_to_msg_id == 555,
            f"reply_to={owner.reply_to}",
        )
        check(
            "virtual ids are negative and stable per (summary, anchor)",
            injector.virtual_id_for(1, 555) < 0
            and injector.virtual_id_for(1, 555) == injector.virtual_id_for(1, 555)
            and injector.virtual_id_for(2, 555) != injector.virtual_id_for(1, 555),
        )

        # Insertion order: ChatActivity.messages is newest-first, so the
        # summary must land at the anchor's index to sit after it on screen.
        class _FakeAdapter:
            # ChatActivityAdapter rows before the first message (header,
            # loading row): the jump must offset the list index by this.
            messagesStartRow = 2

            def __init__(self):
                self.notified = 0

            def notifyDataSetChanged(self):
                self.notified += 1

        class _FakeCell:
            """Live ChatMessageCell: the row the targeted repaint should hit."""

            def __init__(self, message_object):
                self._message_object = message_object
                self.rebound = 0
                self.invalidated = 0

            def getMessageObject(self):
                return self._message_object

            def setMessageObject(self, mo, grouped, pinned_bottom, pinned_top, first):
                self._message_object = mo
                self.rebound += 1

            def invalidate(self):
                self.invalidated += 1

        class _FakeDelegate:
            def __init__(self):
                self.forced = 0

            def forceUpdate(self, cell, a, b):
                self.forced += 1

        class _FakeListView:
            def __init__(self):
                self.cells = []
                self.stopped = 0
                self.scroll_state = 0

            def getScrollState(self):
                return self.scroll_state

            def getMeasuredHeight(self):
                return 900

            def stopScroll(self):
                self.stopped += 1

            def getChildCount(self):
                return len(self.cells)

            def getChildAt(self, index):
                return self.cells[index]

        class _FakeLayoutManager:
            """Records jumps and reports the visible row range they produce.

            `hijack` simulates ChatActivity's own postponed scroll landing
            right after a jump and carrying the view elsewhere.
            """

            def __init__(self):
                self.jumps = []
                self.visible = (-1, -1)
                self.hijack = 0

            def scrollToPositionWithOffset(self, position, offset, bottom):
                self.jumps.append((position, offset, bottom))
                if self.hijack:
                    self.hijack -= 1
                    self.visible = (40, 46)
                else:
                    self.visible = (max(0, position - 3), position + 3)

            def findFirstVisibleItemPosition(self):
                return self.visible[0]

            def findLastVisibleItemPosition(self):
                return self.visible[1]

        class _FakeChatActivity:
            def __init__(self, ids):
                self.messages = _JavaList()
                for mid in ids:
                    owner_obj = _TLObj("TL_message")
                    owner_obj.id = mid
                    self.messages.add(_FakeMessageObject(0, owner_obj))
                self.chatAdapter = _FakeAdapter()
                self.chatMessageCellDelegate = _FakeDelegate()
                self.chatListView = _FakeListView()
                self.chatLayoutManager = _FakeLayoutManager()
                self.scrolled_to = []
                # Ids scrollToMessageId will "load" into the list, mimicking
                # the history request it issues for an unloaded id.
                self.loadable = {}

            def getDialogId(self):
                return -4242

            def scrollToMessageId(self, msg_id, from_id, select, load_index, force, unread):
                self.scrolled_to.append(msg_id)
                for mid in self.loadable.pop(msg_id, ()):
                    owner_obj = _TLObj("TL_message")
                    owner_obj.id = mid
                    self.messages.add(_FakeMessageObject(0, owner_obj))

            def ids(self):
                return [m.messageOwner.id for m in self.messages]

        activity = _FakeChatActivity([600, 555, 500])
        check(
            "insertion index places the summary directly after its anchor",
            injector.find_index_for_anchor(activity.messages, 555) == 1,
            f"index={injector.find_index_for_anchor(activity.messages, 555)}",
        )

        # The anchor is the newest ANALYZED message, which with a non-zero
        # offset is not the newest message in the chat. The card must land
        # there, never at the bottom.
        offset_activity = _FakeChatActivity([900, 800, 700, 600, 500])
        check(
            "with an offset the card lands at the analyzed range, not the chat bottom",
            injector.find_index_for_anchor(offset_activity.messages, 700) == 2,
            f"index={injector.find_index_for_anchor(offset_activity.messages, 700)}",
        )

        # Anchor not loaded in this view: falling back to index 0 would put the
        # card below messages the summary never saw, which is the bug this
        # ordering fallback exists to prevent.
        # These exercise the id-ordering fallback itself, which is now opt-in:
        # callers that must place a card somewhere (reanchor) ask for it, while
        # the default refuses to guess a position for an unloaded anchor.
        check(
            "requested id ordering places an unloaded anchor by id",
            injector.find_index_for_anchor(offset_activity.messages, 750, require_anchor=False) == 2,
            f"index={injector.find_index_for_anchor(offset_activity.messages, 750, require_anchor=False)}",
        )
        check(
            "an anchor older than everything loaded lands at the top of the list",
            injector.find_index_for_anchor(offset_activity.messages, 100, require_anchor=False)
            == offset_activity.messages.size(),
            f"index={injector.find_index_for_anchor(offset_activity.messages, 100, require_anchor=False)}",
        )
        check(
            "an anchor newer than everything loaded lands at the bottom",
            injector.find_index_for_anchor(offset_activity.messages, 1000, require_anchor=False) == 0,
            f"index={injector.find_index_for_anchor(offset_activity.messages, 1000, require_anchor=False)}",
        )

        # The real invariant is positional, not a fixed index: every message
        # newer than the anchor stays below the card, every older one above,
        # regardless of injected cards already sitting in the list.
        def neighbours(activity_obj, anchor):
            # require_anchor=False so the invariant is checked for unloaded
            # anchors too: when a position is chosen at all, it must be correct.
            index = injector.find_index_for_anchor(
                activity_obj.messages, anchor, require_anchor=False
            )
            ids = [m.messageOwner.id for m in activity_obj.messages]
            real = [i for i in ids if i > 0]
            below = [i for i in ids[:index] if i > 0]
            above = [i for i in ids[index:] if i > 0]
            return below, above, real

        mixed = _FakeChatActivity([900, 800, 700])
        injected_owner = _TLObj("TL_message")
        injected_owner.id = -800007
        mixed.messages.add(1, _FakeMessageObject(0, injected_owner))
        below, above, _ = neighbours(mixed, 750)
        check(
            "an injected card already in the list does not misplace the next one",
            all(i > 750 for i in below) and all(i < 750 for i in above),
            f"below={below} above={above}",
        )

        # Same invariant across the offset cases, which is what "injected only
        # after the latest analyzed message" actually means.
        for anchor in (700, 750, 900, 100, 1000):
            below, above, _ = neighbours(offset_activity, anchor)
            check(
                f"no message newer than anchor {anchor} ends up below the card",
                all(i > anchor for i in below) and all(i <= anchor for i in above),
                f"anchor={anchor} below={below} above={above}",
            )

        virtual_id = injector.virtual_id_for(9, 555)
        inj = injector.SummaryInjection(activity, 0, -4242, 0, 555, virtual_id)
        inserted = inj.insert(thinking="first chunk", streaming=True)
        check(
            "injection inserts one message at the anchor index and refreshes the adapter",
            inserted
            and activity.messages.size() == 4
            and activity.messages.get(1).messageOwner.id == virtual_id
            and activity.chatAdapter.notified >= 1,
            f"size={activity.messages.size()} notified={activity.chatAdapter.notified}",
        )

        # The analyzed range can end far up the history, so the card may be
        # injected off-screen; the chat has to move to it. scrollToMessageId
        # only smooth-scrolls from stale cell positions right after an insert,
        # so placement hard-jumps the layout manager to the card's own row.
        check(
            "placing the card hard-jumps the layout manager to the card's row",
            activity.chatLayoutManager.jumps == [(2 + 1, 300, False)]
            and activity.chatListView.stopped == 1,
            f"jumps={activity.chatLayoutManager.jumps} stopped={activity.chatListView.stopped}",
        )
        check(
            "a loaded anchor needs no history load before placement",
            activity.scrolled_to == [],
            f"scrolled_to={activity.scrolled_to}",
        )

        before = activity.messages.get(1)
        inj.update(answer_markdown="final answer", thinking="reasoning", streaming=False)
        after = activity.messages.get(1)
        check(
            "update swaps in a rebuilt MessageObject rather than mutating in place",
            activity.messages.size() == 4
            and after is not before
            and after.messageOwner.id == virtual_id
            and after.type == 36
            and [b.tl_name for b in after.messageOwner.rich_message.blocks]
            == ["pageBlockDetails", "pageBlockParagraph"],
            f"size={activity.messages.size()} same={after is before} "
            f"blocks={[b.tl_name for b in after.messageOwner.rich_message.blocks]}",
        )
        check(
            "a streaming update does not scroll again and fight the user",
            activity.scrolled_to == [] and len(activity.chatLayoutManager.jumps) == 1,
            f"scrolled_to={activity.scrolled_to} jumps={activity.chatLayoutManager.jumps}",
        )

        # Streaming repaints should hit the one visible cell, not rebind every
        # row: notifyDataSetChanged() on each delta is what made the card
        # flicker and lose scroll position.
        activity.chatListView.cells = [_FakeCell(activity.messages.get(1))]
        notified_before = activity.chatAdapter.notified
        inj.update(answer_markdown="more text", thinking="reasoning", streaming=True)
        check(
            "a streaming repaint updates the live cell instead of the whole list",
            activity.chatMessageCellDelegate.forced == 1
            and activity.chatListView.cells[0].rebound == 1
            and activity.chatAdapter.notified == notified_before,
            f"forced={activity.chatMessageCellDelegate.forced} "
            f"rebound={activity.chatListView.cells[0].rebound} "
            f"notified={activity.chatAdapter.notified} (was {notified_before})",
        )

        # Off-screen card: no cell to repaint, so the adapter must be told.
        activity.chatListView.cells = []
        notified_before = activity.chatAdapter.notified
        inj.update(answer_markdown="offscreen", thinking="reasoning", streaming=True)
        check(
            "an off-screen card falls back to a full adapter refresh",
            activity.chatAdapter.notified == notified_before + 1,
            f"notified={activity.chatAdapter.notified} (was {notified_before})",
        )

        # Re-summarizing a range whose saved card the restore hook already
        # re-injected must claim that card, not sit beside it.
        collide_activity = _FakeChatActivity([600, 555, 500])
        saved_id = injector.virtual_id_for(12, 555)
        injector.restore_saved_summaries(
            collide_activity,
            0,
            -4242,
            0,
            [summary_db_mod.StoredSummary(12, -4242, 0, 555, "previous run", "", 5, 0)],
        )
        collide = injector.SummaryInjection(collide_activity, 0, -4242, 0, 555, saved_id)
        collide.insert(thinking="new run", streaming=True)
        collide.update(answer_markdown="new answer", streaming=True)
        ids = collide_activity.ids()
        check(
            "re-summarizing a restored range reuses its card instead of duplicating",
            ids.count(saved_id) == 1 and len(ids) == 4,
            f"ids={ids}",
        )

        # The anchor is the newest ANALYZED message; with an offset it can be
        # outside the loaded window. The card must never fall back to the chat
        # bottom: the chat is asked to load the anchor, and the card is placed
        # directly after it once it arrives.
        far = _FakeChatActivity([900, 800])
        far.loadable[300] = [310, 300, 290]
        far_card = injector.virtual_id_for(5, 300)
        far_inj = injector.SummaryInjection(far, 0, -4242, 0, 300, far_card)
        far_inj.insert(thinking="chunk", streaming=True)
        far_ids = far.ids()
        check(
            "an unloaded anchor is loaded first, then the card lands right after it",
            far.scrolled_to == [300]
            and far_inj.placed
            and far_ids.index(far_card) == far_ids.index(300) - 1,
            f"scrolled_to={far.scrolled_to} ids={far_ids}",
        )
        check(
            "the card is never parked at the bottom while its anchor loads",
            far_ids[0] == 900,
            f"ids={far_ids}",
        )

        # Loading a far anchor makes ChatActivity run its own postponed scroll,
        # which can land after the jump and carry the view away from the card.
        # The settle checks must bring it back.
        hijacked = _FakeChatActivity([900, 800])
        hijacked.loadable[300] = [310, 300, 290]
        hijacked.chatLayoutManager.hijack = 1
        hijacked_card = injector.virtual_id_for(7, 300)
        injector.SummaryInjection(hijacked, 0, -4242, 0, 300, hijacked_card).insert(
            thinking="chunk", streaming=True
        )
        card_row = 2 + hijacked.ids().index(hijacked_card)
        first, last = hijacked.chatLayoutManager.visible
        check(
            "a jump overridden by the app's own scroll is redone until the card is on screen",
            len(hijacked.chatLayoutManager.jumps) == 2 and first <= card_row <= last,
            f"jumps={hijacked.chatLayoutManager.jumps} visible={(first, last)} row={card_row}",
        )

        # Once the user takes hold of the list, the plugin stops steering it.
        dragged = _FakeChatActivity([600, 555, 500])
        dragged.chatLayoutManager.hijack = 1
        dragged.chatListView.scroll_state = 1
        injector.SummaryInjection(dragged, 0, -4242, 0, 555, injector.virtual_id_for(8, 555)).insert(
            thinking="chunk", streaming=True
        )
        check(
            "no re-jump while the user is dragging the list",
            len(dragged.chatLayoutManager.jumps) == 1,
            f"jumps={dragged.chatLayoutManager.jumps}",
        )

        # A cache-then-server reload can replace the list under a fresh card.
        reloaded = _FakeChatActivity([600, 555, 500])
        reloaded_card = injector.virtual_id_for(9, 555)
        reloaded_inj = injector.SummaryInjection(reloaded, 0, -4242, 0, 555, reloaded_card)
        real_on_ui_for_reload = injector._on_ui
        dropped = []

        def _reloading_on_ui(fn, delay_ms=0):
            if delay_ms and not dropped:
                # First settle check: the chat reloads its window first.
                reloaded.messages[:] = [m for m in reloaded.messages if m.messageOwner.id > 0]
                dropped.append(True)
            fn()

        injector._on_ui = _reloading_on_ui
        try:
            reloaded_inj.insert(thinking="chunk", streaming=True)
        finally:
            injector._on_ui = real_on_ui_for_reload
        reloaded_ids = reloaded.ids()
        check(
            "a card dropped by a list reload right after placement is put back",
            dropped and reloaded_ids.count(reloaded_card) == 1
            and reloaded_ids.index(reloaded_card) == reloaded_ids.index(555) - 1,
            f"ids={reloaded_ids}",
        )

        # An anchor that never loads: no card anywhere, rather than a card
        # attached to messages it did not analyze.
        never = _FakeChatActivity([900, 800])
        never_card = injector.virtual_id_for(6, 300)
        never_inj = injector.SummaryInjection(never, 0, -4242, 0, 300, never_card)
        never_inj.insert(thinking="chunk", streaming=True)
        never_inj.update(answer_markdown="more", streaming=True)
        check(
            "an anchor that never loads places no card at a wrong position",
            never_card not in never.ids() and not never_inj.placed,
            f"ids={never.ids()}",
        )
        check(
            "a far anchor that does not arrive is requested again, not just once",
            len(never.scrolled_to) >= 2 and set(never.scrolled_to) == {300},
            f"scrolled_to={never.scrolled_to}",
        )
        # ...and the card shows up once the user scrolls the anchor in.
        for mid in (310, 300):
            owner_obj = _TLObj("TL_message")
            owner_obj.id = mid
            never.messages.add(_FakeMessageObject(0, owner_obj))
        never_inj.update(answer_markdown="final", streaming=False, tagged=True)
        never_ids = never.ids()
        check(
            "a card whose anchor loads late is placed on the next render",
            never_inj.placed and never_ids.index(never_card) == never_ids.index(300) - 1,
            f"ids={never_ids}",
        )

        # Finished summaries carry the #aisummaries tag, in the card and in
        # the plain-text body (with a hashtag entity), so copied text and any
        # text matching see it. Streaming renders stay untagged.
        tagged_mo = injector.build_summary_message(
            0, -4242, 0, 555, injector.virtual_id_for(1, 555),
            "**Done** see [Alice](tg://user?id=1)", "", False, tagged=True,
        )
        tagged_owner = tagged_mo.messageOwner
        last_block = tagged_owner.rich_message.blocks[-1]
        entity = tagged_owner.entities[0] if len(tagged_owner.entities) else None
        tag_node = getattr(last_block, "text", None)
        check(
            "a finished summary ends with a native #aisummaries hashtag node",
            last_block.tl_name == "pageBlockParagraph"
            and getattr(tag_node, "tl_name", None) == "textHashtag"
            and getattr(tag_node.text, "text", None) == "#aisummaries",
            f"last={last_block.tl_name} node={getattr(tag_node, '__dict__', None)}",
        )
        check(
            "the plain-text body is searchable text plus a hashtag entity",
            tagged_owner.message == "Done see Alice\n\n#aisummaries"
            and entity is not None
            and entity.tl_name == "TL_messageEntityHashtag"
            and entity.offset == len("Done see Alice\n\n")
            and entity.length == len("#aisummaries"),
            f"message={tagged_owner.message!r} entity={getattr(entity, '__dict__', None)}",
        )
        streaming_mo = injector.build_summary_message(
            0, -4242, 0, 555, injector.virtual_id_for(1, 555), "partial", "", True
        )
        check(
            "a streaming render is not tagged",
            "#aisummaries" not in streaming_mo.messageOwner.message
            and len(streaming_mo.messageOwner.entities) == 0,
            f"message={streaming_mo.messageOwner.message!r}",
        )

        # Re-inserting the same virtual id must replace, never duplicate.
        inj2 = injector.SummaryInjection(activity, 0, -4242, 0, 555, virtual_id)
        inj2.insert(answer_markdown="again", thinking="", streaming=False)
        check(
            "re-injecting an existing virtual id replaces instead of duplicating",
            activity.messages.size() == 4,
            f"size={activity.messages.size()}",
        )

        restore_activity = _FakeChatActivity([600, 555, 500])
        restored = injector.restore_saved_summaries(
            restore_activity,
            0,
            -4242,
            0,
            [summary_db_mod.StoredSummary(3, -4242, 0, 555, "saved text", "", 5, 0)],
        )
        check(
            "stored summaries are re-injected when a chat is reopened",
            restored == 1 and restore_activity.messages.size() == 4,
            f"restored={restored} size={restore_activity.messages.size()}",
        )

        # Several stored summaries for one chat must each land after their own
        # analyzed range, not pile up at one position.
        multi_activity = _FakeChatActivity([900, 800, 700, 600, 500])
        injector.restore_saved_summaries(
            multi_activity,
            0,
            -4242,
            0,
            [
                summary_db_mod.StoredSummary(11, -4242, 0, 600, "older range", "", 5, 0),
                summary_db_mod.StoredSummary(12, -4242, 0, 800, "newer range", "", 5, 0),
            ],
        )
        ids = [m.messageOwner.id for m in multi_activity.messages]
        positions = {}
        for position, msg_id in enumerate(ids):
            positions[msg_id] = position
        older_card = injector.virtual_id_for(11, 600)
        newer_card = injector.virtual_id_for(12, 800)
        check(
            "each restored summary sits directly above its own anchor",
            positions[older_card] == positions[600] - 1
            and positions[newer_card] == positions[800] - 1,
            f"ids={ids}",
        )
        check(
            "restoring several summaries keeps chat messages in order",
            [i for i in ids if i > 0] == [900, 800, 700, 600, 500],
            f"ids={ids}",
        )

        # Bug: every saved summary appeared stacked in a row at the bottom of
        # the chat. Cards whose anchor is outside the loaded window have no
        # correct position, and placing them by id order alone collapsed them
        # together. They must be skipped until paging brings their anchor in.
        narrow = _FakeChatActivity([600, 590, 580])
        stranded = injector.restore_saved_summaries(
            narrow,
            0,
            -4242,
            0,
            [
                summary_db_mod.StoredSummary(21, -4242, 0, 120, "ancient", "", 5, 0),
                summary_db_mod.StoredSummary(22, -4242, 0, 130, "old", "", 5, 0),
                summary_db_mod.StoredSummary(23, -4242, 0, 590, "loaded", "", 5, 0),
            ],
        )
        narrow_ids = [m.messageOwner.id for m in narrow.messages]
        injected = [i for i in narrow_ids if i < 0]
        check(
            "summaries whose anchor is not loaded are not injected at all",
            stranded == 1 and injected == [injector.virtual_id_for(23, 590)],
            f"dispatched={stranded} injected={injected} ids={narrow_ids}",
        )
        check(
            "the one restorable summary still lands directly above its anchor",
            narrow_ids.index(injector.virtual_id_for(23, 590)) == narrow_ids.index(590) - 1,
            f"ids={narrow_ids}",
        )

        # Once paging widens the window, the previously skipped summary has a
        # real anchor and must be restored on the next attempt.
        for older in (130, 120):
            owner_obj = _TLObj("TL_message")
            owner_obj.id = older
            narrow.messages.add(_FakeMessageObject(0, owner_obj))
        after_paging = injector.restore_saved_summaries(
            narrow,
            0,
            -4242,
            0,
            [
                summary_db_mod.StoredSummary(21, -4242, 0, 120, "ancient", "", 5, 0),
                summary_db_mod.StoredSummary(22, -4242, 0, 130, "old", "", 5, 0),
                summary_db_mod.StoredSummary(23, -4242, 0, 590, "loaded", "", 5, 0),
            ],
        )
        paged_ids = [m.messageOwner.id for m in narrow.messages]
        check(
            "summaries are restored once paging loads their anchor",
            after_paging == 2
            and paged_ids.index(injector.virtual_id_for(21, 120)) == paged_ids.index(120) - 1
            and paged_ids.index(injector.virtual_id_for(22, 130)) == paged_ids.index(130) - 1,
            f"dispatched={after_paging} ids={paged_ids}",
        )
        check(
            "re-restoring does not duplicate the summary already on screen",
            paged_ids.count(injector.virtual_id_for(23, 590)) == 1,
            f"ids={paged_ids}",
        )

        # Bug: the card appeared after the wrong message when many messages
        # were included. The insert index was computed on the calling thread and
        # applied later on the UI thread; anything arriving in between shifted
        # the list and invalidated it. A fake dispatcher that mutates the list
        # between computing and applying reproduces that window.
        racing = _FakeChatActivity([700, 690, 680])
        real_on_ui = injector._on_ui

        def _racing_on_ui(fn):
            owner_obj = _TLObj("TL_message")
            owner_obj.id = 701
            racing.messages.add(0, _FakeMessageObject(0, owner_obj))
            fn()

        injector._on_ui = _racing_on_ui
        try:
            racing_injection = injector.SummaryInjection(
                racing, 0, -4242, 0, 690, injector.virtual_id_for(31, 690)
            )
            racing_injection.insert("summary body", "", streaming=False)
        finally:
            injector._on_ui = real_on_ui
        racing_ids = [m.messageOwner.id for m in racing.messages]
        card = injector.virtual_id_for(31, 690)
        check(
            "the card lands above its anchor even if the list shifts first",
            racing_ids.index(card) == racing_ids.index(690) - 1,
            f"ids={racing_ids}",
        )

        # find_index_for_anchor must report an absent anchor rather than
        # guessing index 0 (chat bottom) or a nearest-by-id slot.
        absent = _FakeChatActivity([900, 800])
        check(
            "an unloaded anchor is reported as absent",
            injector.find_index_for_anchor(absent.messages, 123) == -1,
            f"index={injector.find_index_for_anchor(absent.messages, 123)}",
        )
        check(
            "id-ordered placement is still available when explicitly requested",
            injector.find_index_for_anchor(absent.messages, 123, require_anchor=False) == 2,
            f"index={injector.find_index_for_anchor(absent.messages, 123, require_anchor=False)}",
        )

        # Every real chat interleaves date separators, which are local
        # MessageObjects carrying messageOwner.id == 0 (ChatActivity builds them
        # with `iput v12, ...Message->id` where v12 is 0, and gives them ids from
        # a separate dateObjectsStableIds space). They must never be treated as
        # the insertion point or counted as a loaded anchor.
        dated = _FakeChatActivity([700, 690])
        separator_owner = _TLObj("TL_messageService")
        separator_owner.id = 0
        dated.messages.add(1, _FakeMessageObject(0, separator_owner))
        dated_index = injector.find_index_for_anchor(dated.messages, 690)
        dated_ids = [m.messageOwner.id for m in dated.messages]
        check(
            "a date separator is not mistaken for the anchor",
            dated_ids[dated_index] == 690,
            f"index={dated_index} ids={dated_ids}",
        )
        # A separator's id 0 must not make "anchor 0" look loaded, and must not
        # be offered as a position; anchor ids are always positive.
        check(
            "a date separator does not satisfy an anchor-loaded check",
            injector.find_index_for_anchor(dated.messages, 0) == -1,
            f"index={injector.find_index_for_anchor(dated.messages, 0)}",
        )
        # A summary anchored on a real message must still restore correctly when
        # separators sit between the loaded messages.
        sep_restore = _FakeChatActivity([800, 790])
        sep_owner = _TLObj("TL_messageService")
        sep_owner.id = 0
        sep_restore.messages.add(1, _FakeMessageObject(0, sep_owner))
        sep_n = injector.restore_saved_summaries(
            sep_restore,
            0,
            -4242,
            0,
            [summary_db_mod.StoredSummary(41, -4242, 0, 790, "text", "", 5, 0)],
        )
        sep_ids = [m.messageOwner.id for m in sep_restore.messages]
        sep_card = injector.virtual_id_for(41, 790)
        check(
            "summaries restore correctly around date separators",
            sep_n == 1 and sep_ids.index(sep_card) == sep_ids.index(790) - 1,
            f"dispatched={sep_n} ids={sep_ids}",
        )

        run_pipeline_checks(plugin if plugin_cls is not None else None, injector, summary_db_mod, _FakeChatActivity, _FakeMessage if plugin_cls is not None else None)

    check(
        "sys.modules gained no entry for bundled plugin module names",
        not ({"main", "config", "services", "providers", "hooks", "ui.summarize_dialog"} & new_modules),
        f"leaked: {({'main', 'config', 'services', 'providers', 'hooks', 'ui.summarize_dialog'} & new_modules)}",
    )

    if _FAILURES:
        print(f"\n{len(_FAILURES)} check(s) failed: {', '.join(_FAILURES)}", file=sys.stderr)
        return 1
    print("\nall checks passed")
    return 0


if __name__ == "__main__":
    sys.exit(main())