fix(telegram): shield batch flush from follow-up cancel

[necoweb3]

Summary

_enqueue_text_event / _enqueue_photo_event / _enqueue_media_group_event
unconditionally cancel the prior flush task whenever a new chunk lands, and
each _flush_* pops its event out of the pending buffer before awaiting
handle_message. A cancel that arrives after that pop destroys a message that
is in no buffer and was never dispatched.

The user sends two Telegram messages back to back; the agent only ever answers
the second. The first gets no reply and no error.

The Discord adapter already shields this exact path, and its comment describes
this precise failure (#12444). Telegram was never updated.

Problem

_flush_text_batch (plugins/platforms/telegram/adapter.py):

await asyncio.sleep(delay)
event = self._pending_text_batches.pop(key, None)   # event leaves the buffer
if not event:
    return
...
await self.handle_message(event)                    # cancellable, unshielded

_enqueue_text_event:

prior_task = self._pending_text_batch_tasks.get(key)
if prior_task and not prior_task.done():            # a task inside
    prior_task.cancel()                             # handle_message is NOT done

When the follow-up lands after the pop, _pending_text_batches no longer
holds the first event, so the new enqueue starts a fresh batch containing only
the second message — and the cancel tears down the dispatch of the first.

The window is open on every flush, because handle_message always suspends
before it durably queues anything: gateway/platforms/base.py awaits
asyncio.to_thread(self._apply_topic_recovery, event) as its first step. It
widens sharply on exactly the path where users send follow-ups — the busy path
(_handle_active_session_busy_message) awaits two further to_thread round
trips (session-store lock read, then a SQLite compression-lock query) before
_queue_or_replace_pending_event durably queues the event.

Nothing recovers the message:

• No error surfaces. The flush has only try/finally, no
  except asyncio.CancelledError; the task is fire-and-forget via
  asyncio.create_task, and asyncio never reports cancelled tasks.
• No redelivery. _handle_text_message returns immediately after the
  synchronous enqueue, so python-telegram-bot has already advanced the polling
  offset — Telegram will not resend.
• No backfill. Unlike Discord (_iter_missed_message_backfill_candidates),
  the Telegram adapter has no missed-message recovery.
• No workaround. The batcher cannot be disabled:
  HERMES_TELEGRAM_TEXT_BATCH_DELAY_SECONDS is clamped to a 0.08 floor, and
  lowering the delay only shifts when the window opens.

The log still prints [Telegram] Flushing text batch <key> (N chars) for the
lost message, so it looks delivered.

The same shape exists in two sibling paths in this file:
_flush_photo_batch, and _flush_media_group_event (whose enqueue cancels
without even a .done() check) — for an album the dropped item is the first
one, which is the one carrying the caption.

Fix

Wrap the dispatch in asyncio.shield in all three flushes, mirroring
plugins/platforms/discord/adapter.py, and add the matching
except asyncio.CancelledError arm so a cancel that lands before the pop
still exits cleanly.

Aggregation is unchanged: when the follow-up arrives before the flush fires
(the common case — a Telegram client splitting a long message), the cancel hits
the new CancelledError arm, nothing was popped, and the chunks still merge
into a single dispatch.

Scope

• plugins/platforms/telegram/adapter.py — asyncio.shield + a
  CancelledError arm in _flush_text_batch, _flush_photo_batch,
  _flush_media_group_event.
• tests/gateway/test_telegram_text_batching.py — 3 regression tests.

No signature or config changes.

Testing

Three new tests in TestBatchFlushNotCancelledByFollowUp (text, photo, album).
They give handle_message a real suspension point — the existing tests stub it
with AsyncMock, which never suspends, which is why they miss this.

Verified by running the actual _enqueue_* / _flush_* bodies from main
against the same scenario (first message enqueued, flush fires, follow-up sent
while the dispatch is in flight):

path	on main	with the fix
text batch	completed=['second message'] — first lost	['first message', 'second message']
photo batch	completed=['photo two'] — first lost	['photo one', 'photo two']
media group	completed=['<1 media>'] — caption item lost	['album caption', '<1 media>']

The existing batching scenarios were replayed against the patched code and all
still pass: single_message_dispatched, split_messages_aggregated,
three_way_split_aggregated, different_chats_not_merged,
batch_cleans_up_after_flush.

[teknium1]

Thanks for identifying a real current-main Telegram batch-flush race: the tracked task is cancelled at plugins/platforms/telegram/adapter.py:8825-8831 after the flush has popped its event at line 8865.

Problems

• The new asyncio.shield(...) detaches handle_message from the task that Telegram teardown cancels and awaits at plugins/platforms/telegram/adapter.py:4168-4171. That can leave dispatch running after the adapter has begun teardown, contrary to the delayed-delivery contract at lines 4134-4141.
• Shielding alone does not preserve arrival order. A successor replaces the task immediately at plugins/platforms/telegram/adapter.py:8825-8831, while handle_message first suspends at gateway/platforms/base.py:5552; no predecessor-completion boundary is enforced.
• The new race tests depend on fixed sleeps (for example tests/gateway/test_telegram_text_batching.py:360) rather than deterministic entered/release synchronization.

Suggested changes

• Keep a separately tracked active dispatch, chain successor batches behind it, and ensure disconnect still cancels and awaits active dispatches.
• Use synchronization events in the regressions; cover successor ordering, a third queued follow-up, and shutdown cancellation.

Automated hermes-sweeper review.

[teknium1]

shield leaves this inner dispatch outside _cancel_pending_delivery_tasks: teardown cancels and awaits the tracked outer flush task, then clears its maps. Please retain a cancellable/awaitable dispatch task for shutdown and use shielding only for normal successor supersession.

[teknium1]

Please replace this timing assumption with entered/release asyncio.Event synchronization. Under a delayed event loop the first flush may not have reached the handler yet, so this test can aggregate both messages instead of exercising the in-flight cancellation race.