exteraGramPlugins/chatSummaries
2
0 Fork 0
generated from exteraGramPlugins/pluginBase
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity
Files
b05d1a274cd00eb4cbcd91b434c5b7ea1c8e2145
chatSummaries/tools/verify_bundle.py
T
lyuksovannyy b05d1a274c Never match a date separator as a summary anchor 
Date separators are local MessageObjects that live in the same
ChatActivity.messages list as real messages, and ChatActivity builds them
with messageOwner.id left at 0 (their stable ids come from a separate
dateObjectsStableIds space, so they never collide with injected cards).

find_index_for_anchor scanned for an exact id match without requiring a
positive id, so looking up anchor 0 returned the index of the first
separator and would have placed a card against a divider rather than a
message. loaded_message_ids had the matching flaw: it recorded the
separators' 0, which made a summary with anchor id 0 look restorable.

Reject non-positive anchors outright and keep separators out of the loaded
id set. Every real chat interleaves separators, so the harness now models
them: they must not be picked as the insertion point, must not satisfy an
anchor-loaded check, and summaries anchored on real messages must still
restore correctly with separators in between.
2026-09-19 16:54:25 +02:00

1989 lines
77 KiB
Python
Executable File
Raw Blame History

#!/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.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.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
def __init__(self, activity, progress_style=0, resources_provider=None):
self.activity = activity
self._view = None
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):
return self
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")),
},
)
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"
).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 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)",
)
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}",
)
# Provisional anchor, chosen before the fetch returns: with an offset
# the analyzed range ends partway up the history, so the guess has to
# skip that many messages instead of grabbing the newest.
class _AnchorList(list):
def size(self):
return len(self)
def get(self, index):
return self[index]
class _AnchorActivity:
def __init__(self, ids):
self.messages = _AnchorList()
for mid in ids:
owner_obj = _TLObj("TL_message")
owner_obj.id = mid
holder = types.SimpleNamespace(messageOwner=owner_obj)
self.messages.append(holder)
anchor_activity = _AnchorActivity([900, 800, 700, 600])
check(
"with no offset the provisional anchor is the newest message",
plugin._newest_visible_message_id(anchor_activity) == 900,
f"got {plugin._newest_visible_message_id(anchor_activity)}",
)
check(
"an offset moves the provisional anchor up the history",
plugin._newest_visible_message_id(anchor_activity, skip=2) == 700,
f"got {plugin._newest_visible_message_id(anchor_activity, skip=2)}",
)
# Injected cards carry negative ids and must never be chosen as the
# anchor, or a second summary would anchor to the first one's card.
mixed_anchor = _AnchorActivity([-700007, 900, 800])
check(
"an existing injected card is never picked as the anchor",
plugin._newest_visible_message_id(mixed_anchor) == 900,
f"got {plugin._newest_visible_message_id(mixed_anchor)}",
)
# ------------------------------------------------------------------
# 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]}",
)
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))}",
)
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:
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 = []
def getChildCount(self):
return len(self.cells)
def getChildAt(self, index):
return self.cells[index]
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.scrolled_to = []
def getDialogId(self):
return -4242
def scrollToMessageId(self, msg_id, from_id, select, load_index, force, unread):
self.scrolled_to.append(msg_id)
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(status="Fetching messages…", 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. The anchor is the
# scroll target because it is a real message the app can resolve, and
# the card sits directly above it.
check(
"inserting the card scrolls the chat to the analyzed range",
activity.scrolled_to == [555],
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 == [555],
f"scrolled_to={activity.scrolled_to}",
)
# 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})",
)
# The card is injected before the fetch, so its anchor starts
# provisional and must be corrected once the real range is known.
reanchor_activity = _FakeChatActivity([600, 555, 500])
provisional_id = injector.virtual_id_for(0, 600)
reinj = injector.SummaryInjection(reanchor_activity, 0, -4242, 0, 600, provisional_id)
reinj.insert(status="Fetching messages…", streaming=True)
check(
"provisional card lands at the newest message before the fetch",
reanchor_activity.messages.get(0).messageOwner.id == provisional_id,
f"ids={[m.messageOwner.id for m in reanchor_activity.messages]}",
)
final_id = injector.virtual_id_for(11, 555)
reinj.reanchor(555, final_id)
ids = [m.messageOwner.id for m in reanchor_activity.messages]
check(
"reanchor moves the card to the real anchor and adopts its virtual id",
reanchor_activity.messages.size() == 4
and ids.index(final_id) == 1
and reanchor_activity.messages.get(1).messageOwner.reply_to.reply_to_msg_id == 555,
f"ids={ids}",
)
check(
"reanchor follows the card to the real end of the analyzed range",
reanchor_activity.scrolled_to == [600, 555],
f"scrolled_to={reanchor_activity.scrolled_to}",
)
# Regression: a streaming update after a reanchor used to add a second
# card, because the lookup missed the old id and fell through to add().
reinj.update(status="Summarizing 98 messages…", streaming=True)
ids = [m.messageOwner.id for m in reanchor_activity.messages]
check(
"updating after a reanchor leaves exactly one summary card",
reanchor_activity.messages.size() == 4
and ids.count(final_id) == 1
and provisional_id not in ids,
f"ids={ids}",
)
# 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_provisional = injector.virtual_id_for(0, 600)
collide = injector.SummaryInjection(collide_activity, 0, -4242, 0, 600, collide_provisional)
collide.insert(status="Fetching messages…", streaming=True)
collide.reanchor(555, saved_id)
collide.update(status="Summarizing…", streaming=True)
ids = [m.messageOwner.id for m in collide_activity.messages]
check(
"re-summarizing a restored range reuses its card instead of duplicating",
ids.count(saved_id) == 1 and collide_provisional not in ids,
f"ids={ids}",
)
# 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}",
)
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())
Reference in new issue View Git Blame Copy Permalink