Inject the summary card only once the answer starts streaming

Pressing Summarize used to inject a status card immediately, at a guessed
anchor, and move it once the fetch returned. Any failure before the model
answered (fetch error, empty history, unreachable host, bad key, HTTP error)
then left an error bubble in the chat and had already scrolled the user away
from where they were reading.

The pipeline is now fetch -> stream -> card. A bulletin acknowledges the
press; every failure before the first chunk is a bulletin only and leaves the
chat untouched. On the first chunk the card is created with its final anchor
and virtual id and placed strictly after the newest analyzed message. If that
message is not loaded, the chat is asked to load it and placement retries; the
card never falls back to the chat bottom. Placement then hard-jumps the layout
manager to the card's row: scrollToMessageId only smooth-scrolls from cell
positions that are stale right after an insert. The provisional anchor and
reanchor() machinery is gone.

The row is reserved with INSERT .. DO NOTHING, so re-summarizing a saved range
no longer blanks its text before the new answer arrives; a fresh row whose
stream breaks is deleted instead of left empty.

Also:
- Finished summaries end with #aisummaries as a native TL_iv.textHashtag
  (rendered by RichMessageLayout like a hashtag entity) and carry a plain-text
  body with a TL_messageEntityHashtag.
- Deleting a card (MessagesController.deleteMessages, all overloads) removes
  its saved row and the card, with a "Summary deleted" bulletin.
- Settings show the saved summary count and approximate size, and a confirmed
  "Clear all summaries" action.
- Version 1.1.0.

Verified on device: no card before streaming, correct placement and jump,
restore on reopen, deletion, and the hashtag rendering. Telegram's search is
server-side, so local cards do not appear in its results.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tools/verify_bundle.py

@@ -162,6 +162,7 @@ class _TLObj:
         self.blocks = _JavaList()
         self.items = _JavaList()
         self.texts = _JavaList()
+        self.entities = _JavaList()
         self.flags = 0
         self.flags2 = 0
 
@@ -254,6 +255,7 @@ def _fake_send_request(request, callback, *, account=None):
 
 _KNOWN_METHODS = {
     "org.telegram.ui.LaunchActivity": ["dispatchTouchEvent"],
+    "org.telegram.messenger.MessagesController": ["deleteMessages", "deleteMessages"],
     "org.telegram.ui.ChatActivity": [
         "onPageDownClicked",
         "updateMessagesVisiblePart",
@@ -343,9 +345,16 @@ class FakeAlertDialogBuilder:
     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
@@ -372,8 +381,12 @@ class FakeAlertDialogBuilder:
         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
 
@@ -572,6 +585,7 @@ def install_fakes() -> None:
             "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
@@ -588,7 +602,7 @@ def install_fakes() -> None:
         "textFixed textUnderline textUrl pageBlockParagraph pageBlockHeading1 "
         "pageBlockHeading2 pageBlockHeading3 pageBlockPreformatted pageBlockBlockquote "
         "pageBlockDivider pageBlockThinking pageBlockList pageBlockOrderedList "
-        "pageBlockDetails TL_pageListItemText TL_pageListOrderedItemText"
+        "pageBlockDetails TL_pageListItemText TL_pageListOrderedItemText textHashtag"
     ).split()
     tgnet_tl_mod.TL_iv = type(
         "TL_iv",
@@ -711,6 +725,242 @@ def expected_dunders(meta: dict) -> Dict[str, Any]:
     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:
@@ -1007,44 +1257,26 @@ def main(argv=None) -> int:
             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)}",
+            "delete_hook hooks every MessagesController.deleteMessages overload",
+            len(plugin.delete_hook._unhooks) == 2,
+            f"installed hooks: {len(plugin.delete_hook._unhooks)} (expected 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])
+        # 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(
-            "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)}",
+            "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)}",
         )
 
     # ------------------------------------------------------------------
@@ -1176,6 +1408,14 @@ def main(argv=None) -> int:
             [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",
@@ -1481,6 +1721,25 @@ def main(argv=None) -> int:
             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")
@@ -1547,6 +1806,10 @@ def main(argv=None) -> int:
         # 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
 
@@ -1581,6 +1844,13 @@ def main(argv=None) -> int:
         class _FakeListView:
             def __init__(self):
                 self.cells = []
+                self.stopped = 0
+
+            def getMeasuredHeight(self):
+                return 900
+
+            def stopScroll(self):
+                self.stopped += 1
 
             def getChildCount(self):
                 return len(self.cells)
@@ -1588,6 +1858,13 @@ def main(argv=None) -> int:
             def getChildAt(self, index):
                 return self.cells[index]
 
+        class _FakeLayoutManager:
+            def __init__(self):
+                self.jumps = []
+
+            def scrollToPositionWithOffset(self, position, offset, bottom):
+                self.jumps.append((position, offset, bottom))
+
         class _FakeChatActivity:
             def __init__(self, ids):
                 self.messages = _JavaList()
@@ -1598,13 +1875,24 @@ def main(argv=None) -> int:
                 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(
@@ -1684,7 +1972,7 @@ def main(argv=None) -> int:
 
         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)
+        inserted = inj.insert(thinking="first chunk", streaming=True)
         check(
             "injection inserts one message at the anchor index and refreshes the adapter",
             inserted
@@ -1695,12 +1983,18 @@ def main(argv=None) -> int:
         )
 
         # 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.
+        # 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(
-            "inserting the card scrolls the chat to the analyzed range",
-            activity.scrolled_to == [555],
+            "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}",
         )
 
@@ -1720,8 +2014,8 @@ def main(argv=None) -> int:
         )
         check(
             "a streaming update does not scroll again and fight the user",
-            activity.scrolled_to == [555],
-            f"scrolled_to={activity.scrolled_to}",
+            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
@@ -1750,46 +2044,6 @@ def main(argv=None) -> int:
             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])
@@ -1801,18 +2055,101 @@ def main(argv=None) -> int:
             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]
+        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 collide_provisional not in ids,
+            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}",
+        )
+
+        # 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 and never.scrolled_to == [300],
+            f"ids={never.ids()} 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)
@@ -2015,23 +2352,7 @@ def main(argv=None) -> int:
             f"dispatched={sep_n} ids={sep_ids}",
         )
 
-        # A live status card is the only feedback that pressing Summarize did
-        # anything, so it must appear even when the provisional anchor cannot be
-        # placed: a large offset reaches past the loaded window, and an
-        # unresolved anchor arrives as 0. Requiring a loaded anchor here left the
-        # user staring at an unchanged chat.
-        for label, anchor in (("unresolved anchor 0", 0), ("anchor past the window", 4242)):
-            live = _FakeChatActivity([700, 690, 680])
-            live_injection = injector.SummaryInjection(
-                live, 0, -4242, 0, anchor, injector.virtual_id_for(0, anchor or 1)
-            )
-            shown = live_injection.insert(status="Fetching messages…", streaming=True)
-            live_ids = [m.messageOwner.id for m in live.messages]
-            check(
-                f"the status card still appears with an {label}",
-                shown and len([i for i in live_ids if i < 0]) == 1,
-                f"inserted={shown} ids={live_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",