fix(telegram): retain loop cleanup across bounded stops

[FanouZeng-TT]

Why

TelegramChannel.stop() runs under the Gateway's bounded five-second
shutdown hook (asyncio.wait_for(channel.stop(), ...)), and it schedules its
real teardown — cancelling in-flight inbound-file bridge tasks and calling
Bot.shutdown() to close the download Bot's HTTPX clients — onto the Telegram
polling loop via run_coroutine_threadsafe, because python-telegram-bot's HTTPX
clients may only be closed on the loop that created them.

The previous implementation awaited that cleanup with a bare
asyncio.wait_for(asyncio.wrap_future(drain_future), ...) and, on timeout or
caller cancellation, called drain_future.cancel() and then unconditionally
telegram_loop.stop() in a finally block. Both the deadline and a cancelled
stop() therefore tore down the Telegram-loop-owned cleanup mid-flight:

• the download Bot's HTTPX connection pool could be left open (connection/FD
  leak), and in-flight bridge tasks could be abandoned rather than cancelled;
• there was no single cleanup owner a retried stop() could join — a second
  stop() would schedule a fresh _shutdown_telegram_resources() racing the
  abandoned one.

This is the same "an outer deadline/cancellation severs owner-held cleanup"
class fixed for the subagent batch service in #6552, here adapted to Telegram's
cross-thread event loop.

What changed

TelegramChannel.stop() now owns a single, retryable cleanup generation:

• The first stop() schedules _shutdown_telegram_resources() on the
  Telegram loop once and records the resulting future
  (_shutdown_resources_future, guarded by a lock so concurrent stops cannot
  schedule two).
• The caller awaits it through asyncio.wait_for(asyncio.shield(...)), so the
  Gateway deadline or a cancelled stop() interrupts only the wait, never the
  cleanup the Telegram loop is running.
• On the deadline, stop() raises ChannelStopTimeout while leaving the loop,
  worker thread, application, and cleanup future owned by the instance — exactly
  the retain-and-retry contract ChannelService already applies to Feishu,
  DingTalk, Slack, and Buzz.
• The loop is stopped, the worker joined, and _thread / _application /
  _shutdown_resources_future cleared only after the cleanup future has
  resolved. A retried stop() joins that same future and converges, with
  Bot.shutdown() running exactly once.

No behavior change for a channel that stops within budget.

Surface area

• Backend API — IM channel lifecycle under backend/app/channels
• Docs / tests / CI only — regression tests only (no change to the happy path)

Bug fix verification

• Test path: backend/tests/test_channels.py::TestTelegramInboundMessages
  • test_shutdown_timeout_keeps_download_bot_cleanup_running
  • test_cancelled_stop_keeps_owned_cleanup_and_retry_converges
  • test_concurrent_stops_share_one_telegram_cleanup_generation
• Did they go red on main and green on this branch? yes. On main the
  deadline/cancellation cancels the owned cleanup (the download Bot's
  shutdown() never completes and the retry cannot join it); on this branch the
  cleanup survives, a retried stop() joins the same generation, and
  Bot.shutdown() runs exactly once.

Validation

Run from backend/:

• PYTHONPATH=.:packages/harness:packages/extension-api python -m pytest tests/test_channels.py::TestTelegramInboundMessages tests/test_telegram_channel_connections.py -q → 42 passed
• make lint → ruff check + ruff format --check clean
• make test-blocking-io → 245 passed, 1 skipped
• make test (4 offline shards) → all Telegram/channel suites green. Two
  failures surfaced that are unrelated to this change and reproduce independently
  of it: test_agent_guidance_check flags frontend/AGENTS.md as over its size
  budget on origin/main already (this PR touches no frontend file), and
  test_skills_custom_multiprocess is a multi-process worker-startup timeout
  under full-suite load that passes when run in isolation.

AI assistance

Tool(s) used: Claude Code

How you used it: AI assisted with diagnosis, the implementation, and the
regression tests; I reviewed every line.

• I've read and understand every line of this change and take responsibility for it — it's not unreviewed AI output.

🤖 Generated with Claude Code

[willem-bd]

Verified against head 6cec431: the shield/generation design matches the ChannelService retain-and-retry contract (instance retained on ChannelStopTimeout/CancelledError, popped only after stop() completes), and I confirmed the regression tests are doing real work — all three new tests fail against the base revision's telegram.py and pass at head, the Telegram suites pass (42 passed), and ruff check/format are clean. One minor, non-blocking finding inline about the dead-loop branch never releasing an abandoned cleanup generation; happy path and retry convergence look correct.

[willem-bd]

Minor/nit, non-blocking: this branch clears only the local cleanup_future, so a retained cleanup generation whose owner loop died before the cleanup could run is never released. The final guard at the bottom (if self._shutdown_resources_future is cleanup_future:) compares the stored future against None and never matches, leaving the abandoned future — and its never-started coroutine, which references the channel — attached to the retained instance, and every retried stop() re-walks this warning path. The instance is retained and heading for discard anyway, so impact is a stale object reference plus repeated warnings, but releasing the generation here keeps the 'exactly one cleanup generation' invariant airtight:

if cleanup_future is not None and not telegram_loop_running:
    # Its owner loop is gone, so nothing can finish this cleanup.
    logger.warning("[Telegram] Telegram loop stopped before its resource cleanup finished; the download bot's HTTPX clients may not have been closed")
    with self._shutdown_resources_lock:
        if self._shutdown_resources_future is cleanup_future:
            self._shutdown_resources_future = None
    cleanup_future = None