chore: release v0.2.8

common._format_assets was passing cache_key=<bare asset UUID>, but the
notification dispatcher writes keys as <host>:<uuid> (derived from the
URL by extract_asset_id_from_url). Result: the two paths populated
different keys for the same asset, so neither could hit the other's
cached file_id and the WebUI stats only ever reflected the notification
side.

Drop the explicit cache_key — TelegramClient derives <host>:<uuid> from
the URL, identical to the notification path, so one file_id cached by
any dispatch or /random / /latest reply is reused by every later send.

RELEASE_NOTES.md

@@ -1,19 +1,18 @@
-# v0.2.5 (2026-04-22)
+# v0.2.8 (2026-04-22)
 
-Hotfix release on top of v0.2.4 — the settings page couldn't save numeric
-fields after the cache-TTL / max-entries rework. See v0.2.4 notes for the
-main changes (thumbhash-validated cache, settings UX overhaul, mobile-nav
-parity).
+Follow-up fix to v0.2.7's Telegram-send unification. The shared factory was
+in place, but commands and notifications were still writing into different
+cache-key namespaces — so in practice the caches never actually shared entries.
 
 ## Bug Fixes
 
-- **Accept numeric values in settings update payload** — Svelte's `bind:value` on `<input type="number">` coerces to a JS number, and Pydantic v2 wouldn't auto-coerce `int → str`, producing a 422 on every save that touched a numeric setting (TTL, max entries) after v0.2.4. Widened numeric fields to `int | str | None` in `SettingsUpdate` and normalized to `str` before persisting. ([d7d0a5d](https://git.dolgolyov-family.by/alexei.dolgolyov/notify-bridge/commit/d7d0a5d))
+- **Commands and notifications now share one `file_id` cache namespace** — `common._format_assets` was passing `cache_key=<bare asset UUID>`, while the notification dispatcher writes keys as `<host>:<uuid>` (derived from the URL by `extract_asset_id_from_url`). The two paths populated different keys for the same asset, so neither could hit the other's cached `file_id` and the Settings → cache-stats card only ever reflected the notification side. Dropped the explicit `cache_key` — `TelegramClient` now derives `<host>:<uuid>` from the URL on both paths, so one `file_id` cached by any dispatch or `/random` / `/latest` reply is reused by every later send. ([7dae68f](https://git.dolgolyov-family.by/alexei.dolgolyov/notify-bridge/commit/7dae68f))
 
 ---
 
 <details>
 <summary>All Commits</summary>
 
-- [d7d0a5d](https://git.dolgolyov-family.by/alexei.dolgolyov/notify-bridge/commit/d7d0a5d) — fix(settings): accept numeric values in update payload *(alexei.dolgolyov)*
+- [7dae68f](https://git.dolgolyov-family.by/alexei.dolgolyov/notify-bridge/commit/7dae68f) — fix(commands): match notification cache-key format so writes share one namespace *(alexei.dolgolyov)*
 
 </details>

frontend/package.json

@@ -1,7 +1,7 @@
 {
 	"name": "notify-bridge-frontend",
 	"private": true,
-	"version": "0.2.5",
+	"version": "0.2.8",
 	"type": "module",
 	"scripts": {
 		"dev": "vite dev",

frontend/src/routes/command-template-configs/+page.svelte

@@ -117,6 +117,14 @@
 		return form.slots[slotName]?.[activeLocale] || '';
 	}
 
+	/** Resolve variable reference for a slot, preferring provider-specific over shared. */
+	function getVarsFor(slotName: string) {
+		const providerVars = varsRef[form.provider_type];
+		return providerVars?.[slotName] ?? varsRef[slotName];
+	}
+
+	let modalVars = $derived(showVarsFor ? getVarsFor(showVarsFor) : null);
+
 	/** Set slot template for current locale (immutable update). */
 	function setSlotValue(slotName: string, value: string) {
 		form.slots = {
@@ -369,7 +377,7 @@
 									{t('templateConfig.preview')}
 								</button>
 							{/if}
-							{#if varsRef[slot.name]}
+							{#if getVarsFor(slot.name)}
 								<button type="button" onclick={() => showVarsFor = slot.name}
 									class="text-xs text-[var(--color-muted-foreground)] hover:underline">{t('templateConfig.variables')}</button>
 							{/if}
@@ -385,7 +393,7 @@
 								onchange={(v: string) => { setSlotValue(slot.name, v); validateSlot(slot.name, v); }}
 								rows={3}
 								errorLine={slotErrorLines[slot.name] || null}
-								variables={varsRef[slot.name] || undefined}
+								variables={getVarsFor(slot.name) || undefined}
 							/>
 						{/if}
 
@@ -468,11 +476,11 @@
 
 <!-- Variables reference modal -->
 <Modal open={showVarsFor !== null} title="{t('templateConfig.variables')}: /{showVarsFor || ''}" onclose={() => showVarsFor = null}>
-	{#if showVarsFor && varsRef[showVarsFor]}
-		<p class="text-sm text-[var(--color-muted-foreground)] mb-3">{varsRef[showVarsFor].description}</p>
+	{#if showVarsFor && modalVars}
+		<p class="text-sm text-[var(--color-muted-foreground)] mb-3">{modalVars.description}</p>
 		<div class="space-y-1">
 			<p class="text-xs font-medium mb-1">{t('templateConfig.variables')}:</p>
-			{#each Object.entries(varsRef[showVarsFor].variables || {}) as [name, desc]}
+			{#each Object.entries(modalVars.variables || {}) as [name, desc]}
 				<div class="flex items-start gap-2 text-sm">
 					<code class="text-xs bg-[var(--color-muted)] px-1 py-0.5 rounded font-mono whitespace-nowrap">{'{{ ' + name + ' }}'}</code>
 					<span class="text-xs text-[var(--color-muted-foreground)]">{desc}</span>
@@ -484,11 +492,19 @@
 			['album_fields', 'album', 'Album fields'],
 			['command_fields', 'cmd', 'Command fields'],
 			['event_fields', 'event', 'Event fields'],
+			['repo_fields', 'repo', 'Repository fields'],
+			['issue_fields', 'issue', 'Issue fields'],
+			['pr_fields', 'pr', 'Pull request fields'],
+			['commit_fields', 'c', 'Commit fields'],
+			['board_fields', 'board', 'Board fields'],
+			['card_fields', 'card', 'Card fields'],
+			['list_fields', 'lst', 'List fields'],
+			['device_fields', 'd', 'Device fields'],
 		] as [fieldKey, prefix, title]}
-			{#if varsRef[showVarsFor][fieldKey]}
+			{#if modalVars[fieldKey]}
 				<div class="mt-3 pt-3 border-t border-[var(--color-border)]">
 					<p class="text-xs font-medium mb-1">{title} <span class="font-normal text-[var(--color-muted-foreground)]">(use {prefix}.field)</span>:</p>
-					{#each Object.entries(varsRef[showVarsFor][fieldKey]) as [name, desc]}
+					{#each Object.entries(modalVars[fieldKey]) as [name, desc]}
 						<div class="flex items-start gap-2 text-sm">
 							<code class="text-xs bg-[var(--color-muted)] px-1 py-0.5 rounded font-mono whitespace-nowrap">{'{{ ' + prefix + '.' + name + ' }}'}</code>
 							<span class="text-xs text-[var(--color-muted-foreground)]">{desc}</span>

packages/core/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "notify-bridge-core"
-version = "0.2.5"
+version = "0.2.8"
 description = "Core library for Notify Bridge — service provider abstractions, models, notifications, and templates"
 requires-python = ">=3.12"
 dependencies = [

packages/core/src/notify_bridge_core/notifications/dispatcher.py

@@ -46,6 +46,7 @@ from .receiver import (
 from .telegram.cache import TelegramFileCache
 from .telegram.client import TelegramClient
 from .telegram.media import (
+    build_telegram_asset_entry,
     extract_asset_id_from_url,
     is_asset_cache_key,
     is_asset_id,
@@ -266,23 +267,19 @@ class NotificationDispatcher:
         # Prefer internal URL for fetching (LAN speed vs public internet)
         internal_url = (target.provider_internal_url or "").rstrip("/")
         external_url = (target.provider_external_url or "").rstrip("/")
-        provider_urls = [u for u in (internal_url, external_url) if u]
         assets = []
         media_assets: list[Any] = []  # aligned with `assets` for preload
         for asset in event.added_assets[:max_media]:
             url = asset.preview_url or asset.thumbnail_url or asset.full_url
-            if url:
-                # Rewrite external URL to internal for faster LAN fetching
-                if internal_url and external_url and url.startswith(external_url):
-                    url = internal_url + url[len(external_url):]
-                asset_type = "video" if asset.type.value == "video" else "photo"
-                asset_headers = {}
-                if target.provider_api_key and any(url.startswith(u) for u in provider_urls):
-                    asset_headers["x-api-key"] = target.provider_api_key
-                asset_entry: dict[str, Any] = {"url": url, "type": asset_type, "headers": asset_headers}
-                # Pass explicit cache_key if set by provider (e.g. Google Photos)
-                if asset.extra.get("cache_key"):
-                    asset_entry["cache_key"] = asset.extra["cache_key"]
+            asset_entry = build_telegram_asset_entry(
+                url=url or "",
+                media_type=asset.type.value,
+                api_key=target.provider_api_key,
+                internal_url=internal_url,
+                external_url=external_url,
+                cache_key=asset.extra.get("cache_key"),
+            )
+            if asset_entry is not None:
                 assets.append(asset_entry)
                 media_assets.append(asset)
 

packages/core/src/notify_bridge_core/notifications/telegram/client.py

@@ -89,6 +89,18 @@ class TelegramClient:
         self, url: str | None, cache_key: str | None = None,
     ) -> tuple[TelegramFileCache | None, str | None, str | None]:
         if cache_key:
+            # Route asset-UUID cache keys to the asset cache so single-item
+            # sends hit the same cache the media-group path uses. Without
+            # this, a command returning one photo stored file_ids in the
+            # URL cache and a command returning multiple stored them in
+            # the asset cache — repeated sends never hit.
+            if is_asset_cache_key(cache_key):
+                bare_id = asset_id_from_cache_key(cache_key)
+                thumbhash = (
+                    self._thumbhash_resolver(bare_id)
+                    if self._thumbhash_resolver else None
+                )
+                return self._asset_cache, cache_key, thumbhash
             return self._url_cache, cache_key, None
         if url:
             if is_asset_id(url):

packages/core/src/notify_bridge_core/notifications/telegram/media.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import re
-from typing import Final
+from typing import Any, Final
 from urllib.parse import urlparse
 
 # Telegram constants
@@ -52,6 +52,65 @@ def extract_asset_id_from_url(url: str) -> str | None:
     return None
 
 
+def build_telegram_asset_entry(
+    *,
+    url: str,
+    media_type: str,
+    api_key: str | None = None,
+    internal_url: str = "",
+    external_url: str = "",
+    cache_key: str | None = None,
+) -> dict[str, Any] | None:
+    """Build a ``TelegramClient.send_notification`` asset dict from raw fields.
+
+    Shared by the notification dispatcher and provider command handlers so
+    both paths agree on media typing, URL rewriting, and auth headers. In
+    particular: video assets MUST be typed ``"video"`` and point at a real
+    video endpoint (e.g. Immich ``/video/playback``) — if they are sent as
+    ``"photo"`` pointing at a thumbnail URL, Telegram delivers a still image
+    for every video in a media group and the user sees a dead poster frame
+    instead of a playable clip.
+
+    Args:
+        url: Source URL for the asset bytes. Prefer a transcoded/preview
+            URL for videos (``/video/playback``) and a preview-sized
+            thumbnail for photos.
+        media_type: Case-insensitive type token. Accepts ``"video"``/
+            ``"VIDEO"``/``MediaType.VIDEO`` or any photo-like string.
+        api_key: Optional API key. Attached as ``x-api-key`` iff the URL is
+            served by one of the provider hosts in ``internal_url`` /
+            ``external_url`` (prevents leaking the key to unrelated hosts).
+        internal_url: LAN-facing provider URL. Used to rewrite
+            ``external_url`` prefixes so Docker-host downloads stay on the
+            LAN instead of egressing to the public domain.
+        external_url: Public provider URL the notification URL was built
+            from. Only used for the LAN rewrite and the api-key scope check.
+        cache_key: Optional explicit cache key. Providers whose URLs don't
+            embed a stable asset id (Google Photos) pass one through so the
+            file_id cache still works.
+
+    Returns ``None`` iff ``url`` is empty.
+    """
+    if not url:
+        return None
+
+    if internal_url and external_url and url.startswith(external_url):
+        url = internal_url + url[len(external_url):]
+
+    normalized_type = str(media_type or "").lower()
+    entry_type = "video" if normalized_type == "video" else "photo"
+
+    headers: dict[str, str] = {}
+    provider_urls = [u for u in (internal_url, external_url) if u]
+    if api_key and (not provider_urls or any(url.startswith(u) for u in provider_urls)):
+        headers["x-api-key"] = api_key
+
+    entry: dict[str, Any] = {"url": url, "type": entry_type, "headers": headers}
+    if cache_key:
+        entry["cache_key"] = cache_key
+    return entry
+
+
 def split_media_by_upload_size(
     media_items: list[tuple], max_upload_size: int
 ) -> list[list[tuple]]:

packages/core/src/notify_bridge_core/providers/immich/asset_utils.py

@@ -193,6 +193,27 @@ def get_asset_video_url(
     return None
 
 
+def build_asset_media_urls(
+    external_url: str, asset_id: str, asset_type: str,
+) -> tuple[str, str]:
+    """Return ``(preview_url, full_url)`` for an Immich asset.
+
+    Single source of truth for the photo-vs-video endpoint rule. Used by
+    ``asset_to_media`` (notification path) and the bot command handlers
+    (command path) so both always pick the transcoded ``/video/playback``
+    for videos and the preview-sized thumbnail for photos — if they
+    diverge, Telegram ends up delivering a still JPEG for videos in a
+    media group.
+    """
+    is_video = asset_type == ASSET_TYPE_VIDEO
+    if is_video:
+        preview_url = f"{external_url}/api/assets/{asset_id}/video/playback"
+    else:
+        preview_url = f"{external_url}/api/assets/{asset_id}/thumbnail?size=preview"
+    full_url = f"{external_url}/api/assets/{asset_id}/original"
+    return preview_url, full_url
+
+
 def build_asset_detail(
     asset: ImmichAssetInfo,
     external_url: str,
@@ -246,12 +267,7 @@ def asset_to_media(asset: ImmichAssetInfo, external_url: str) -> MediaAsset:
     # preview_url is what the notification dispatcher feeds to Telegram as the
     # actual media bytes — for videos it must be the transcoded playback (mp4),
     # not the JPEG thumbnail, or Telegram receives a JPEG labeled as video/mp4.
-    if asset.type == ASSET_TYPE_VIDEO:
-        preview_url = f"{external_url}/api/assets/{asset.id}/video/playback"
-        full_url = f"{external_url}/api/assets/{asset.id}/original"
-    else:
-        preview_url = f"{external_url}/api/assets/{asset.id}/thumbnail?size=preview"
-        full_url = f"{external_url}/api/assets/{asset.id}/original"
+    preview_url, full_url = build_asset_media_urls(external_url, asset.id, asset.type)
 
     return MediaAsset(
         id=asset.id,

packages/server/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "notify-bridge-server"
-version = "0.2.5"
+version = "0.2.8"
 description = "Standalone Notify Bridge server — FastAPI REST API with SQLite database"
 requires-python = ">=3.12"
 dependencies = [

packages/server/src/notify_bridge_server/api/app_settings.py

@@ -100,6 +100,13 @@ async def update_settings(
         if value is None:
             continue
         value_str = str(value)
+        # GET masks the webhook secret as "***<last4>" so the real value is
+        # never exposed to the frontend. If the client sends the mask back
+        # (which happens on every save, since bind:value holds whatever GET
+        # returned), treat it as "unchanged" — otherwise we'd overwrite the
+        # real secret with its mask, silently breaking webhook HMAC.
+        if key == "telegram_webhook_secret" and value_str.startswith("***"):
+            continue
         row = await session.get(AppSetting, key)
         if row:
             row.value = value_str

packages/server/src/notify_bridge_server/commands/handler.py

@@ -367,20 +367,23 @@ async def send_reply(
     bot_token: str, chat_id: str, text: str, reply_to_message_id: int | None = None,
     session: aiohttp.ClientSession | None = None,
 ) -> None:
-    """Send a text reply via TelegramClient.
+    """Send a text reply to a chat.
 
-    Command responses are listings (albums, people, events, ...) that embed
-    multiple links; Telegram's default behavior of rendering a preview of
-    the first URL is almost never what the user wants and clashes with the
-    "Disable link previews" toggle operators set on their Telegram target.
-    We always pass ``disable_web_page_preview=True`` here.
+    Thin wrapper that goes through the single ``services.telegram_send``
+    entry point so commands and notifications share one routine — same
+    HTTP session pool, same file_id caches.
+
+    Command responses are listings (albums, people, events, ...) that
+    embed multiple links; Telegram's default behavior of rendering a
+    preview of the first URL is almost never what the user wants and
+    clashes with the "Disable link previews" toggle operators set on
+    their Telegram target. We always pass
+    ``disable_web_page_preview=True`` here.
     """
-    if session is None:
-        from ..services.http_session import get_http_session
-        session = await get_http_session()
-    client = TelegramClient(session, bot_token)
-    result = await client.send_message(
-        chat_id, text,
+    from ..services.telegram_send import send_telegram_message
+
+    result = await send_telegram_message(
+        bot_token, chat_id, text,
         reply_to_message_id=reply_to_message_id,
         disable_web_page_preview=True,
     )
@@ -393,30 +396,28 @@ async def send_media_group(
     reply_to_message_id: int | None = None,
     session: aiohttp.ClientSession | None = None,
 ) -> None:
-    """Send media items via TelegramClient.send_notification."""
+    """Send media items via the shared Telegram routine.
+
+    ``media_items`` must already be in TelegramClient asset format — each
+    entry contains ``type`` (``"photo"``/``"video"``/``"document"``),
+    ``url``, optional ``cache_key``, and optional ``headers``. Provider
+    command handlers build this format via
+    ``build_telegram_asset_entry`` — the same helper the notification
+    dispatcher uses — so videos keep their ``"video"`` type and point at
+    a real video URL instead of a still thumbnail.
+
+    Uses ``services.telegram_send.send_telegram_media`` so the URL cache
+    and asset cache are wired in exactly like the notification path.
+    Repeated ``/latest`` / ``/random`` commands that match previously-sent
+    assets hit the cache and skip the re-upload.
+    """
     if not media_items:
         return
 
-    # Convert command handler media format to TelegramClient asset format
-    assets = []
-    for item in media_items:
-        assets.append({
-            "type": "photo",
-            "url": item.get("thumbnail_url", ""),
-            "cache_key": item.get("asset_id", ""),
-            "headers": {"x-api-key": item.get("api_key", "")},
-        })
+    from ..services.telegram_send import send_telegram_media
 
-    # Build caption from first item
-    captions = [item.get("caption", "") for item in media_items if item.get("caption")]
-    caption = "\n".join(captions) if captions else None
-
-    if session is None:
-        from ..services.http_session import get_http_session
-        session = await get_http_session()
-    client = TelegramClient(session, bot_token)
-    result = await client.send_notification(
-        chat_id, assets=assets, caption=caption,
+    result = await send_telegram_media(
+        bot_token, chat_id, media_items,
         reply_to_message_id=reply_to_message_id,
         chat_action=None,
     )

packages/server/src/notify_bridge_server/commands/immich/common.py

@@ -6,7 +6,11 @@ import asyncio
 import logging
 from typing import Any
 
-from notify_bridge_core.providers.immich.asset_utils import get_public_url
+from notify_bridge_core.notifications.telegram.media import build_telegram_asset_entry
+from notify_bridge_core.providers.immich.asset_utils import (
+    build_asset_media_urls,
+    get_public_url,
+)
 
 from ..handler import _render_cmd_template
 
@@ -74,13 +78,16 @@ def build_asset_dict(
 ) -> dict[str, Any]:
     """Build a rich asset dict for command templates from an ImmichAssetInfo or raw dict."""
     if isinstance(asset, dict):
+        # Immich raw search responses nest geo under exifInfo — pull it out so
+        # templates can use flat asset.city / asset.country.
+        exif = asset.get("exifInfo") or {}
         d = {
             "id": asset.get("id", ""),
             "originalFileName": asset.get("originalFileName", asset.get("filename", "")),
             "type": asset.get("type", "IMAGE"),
             "createdAt": asset.get("createdAt", asset.get("created_at", asset.get("fileCreatedAt", ""))),
-            "city": asset.get("city", ""),
-            "country": asset.get("country", ""),
+            "city": asset.get("city") or exif.get("city") or "",
+            "country": asset.get("country") or exif.get("country") or "",
             "is_favorite": asset.get("is_favorite", asset.get("isFavorite", False)),
             "public_url": asset.get("public_url", public_url),
         }
@@ -123,16 +130,32 @@ def _format_assets(
     })
 
     if response_mode == "media":
+        # Reuse the same URL rule (build_asset_media_urls) and entry builder
+        # (build_telegram_asset_entry) as the notification dispatcher so both
+        # paths agree on video → /video/playback and photo → thumbnail. When
+        # these diverged, Telegram rendered a still JPEG for each video in
+        # the media group instead of the real clip.
+        #
+        # We deliberately do NOT pass ``cache_key`` here. TelegramClient
+        # derives it from the URL as ``<host>:<uuid>`` — identical to what
+        # the notification dispatcher produces via extract_asset_id_from_url.
+        # Passing the bare UUID would put command writes in a separate
+        # namespace from notification writes, so neither path could hit the
+        # other's cached file_ids (which is what made the cache look empty
+        # from the WebUI after running /random).
         media_items: list[dict[str, Any]] = []
         for asset in assets:
             asset_id = asset.get("id", "")
-            media_items.append({
-                "type": "photo",
-                "asset_id": asset_id,
-                "caption": "",
-                "thumbnail_url": f"{client.url}/api/assets/{asset_id}/thumbnail?size=preview",
-                "api_key": client.api_key,
-            })
+            asset_type = (asset.get("type") or "").upper()
+            preview_url, _ = build_asset_media_urls(client.url, asset_id, asset_type)
+            entry = build_telegram_asset_entry(
+                url=preview_url,
+                media_type="video" if asset_type == "VIDEO" else "image",
+                api_key=client.api_key,
+                internal_url=client.url,
+            )
+            if entry is not None:
+                media_items.append(entry)
         # Return text message + media items — text is sent first, media as reply
         return {"text": text, "media": media_items}
 

packages/server/src/notify_bridge_server/commands/immich/search.py

@@ -5,17 +5,14 @@ from __future__ import annotations
 from typing import Any
 
 from ..handler import _render_cmd_template
-from .common import _format_assets
+from .common import _format_assets, build_asset_dict
 
 
 def _enrich_assets(assets: list[dict[str, Any]], asset_public_urls: dict[str, str]) -> list[dict[str, Any]]:
-    """Add public_url to assets from the pre-built map. Returns new list without mutating inputs."""
-    if not asset_public_urls:
-        return assets
+    """Normalize raw Immich assets and attach public_url from the pre-built map."""
+    pub = asset_public_urls or {}
     return [
-        {**asset, "public_url": asset_public_urls.get(asset.get("id", ""), "")}
-        if asset.get("id", "") in asset_public_urls and not asset.get("public_url")
-        else asset
+        build_asset_dict(asset, public_url=pub.get(asset.get("id", ""), ""))
         for asset in assets
     ]
 

packages/server/src/notify_bridge_server/services/telegram_send.py

@@ -0,0 +1,119 @@
+"""Single entry point for all Telegram send operations.
+
+Both the notification dispatcher (event-driven) and the bot command
+handlers (user-driven) funnel their Telegram API calls through this
+module. Keeping construction in one place means:
+
+* The shared aiohttp session is always reused (one TCP pool for the
+  whole process).
+* The Telegram file_id caches (URL cache + asset cache) are always
+  wired in, so repeated sends — whether from a scheduled tracker or
+  a ``/latest`` command — reuse cached file_ids instead of re-uploading
+  the same bytes.
+* Future cross-cutting concerns (rate limiting, telemetry, retries)
+  have exactly one place to live.
+
+The actual Telegram API routine is still ``TelegramClient`` in core —
+this module just guarantees every caller gets a properly-wired client.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+import aiohttp
+
+from notify_bridge_core.notifications.telegram.client import (
+    NotificationResult,
+    TelegramClient,
+)
+
+from .http_session import get_http_session
+from .watcher import _get_telegram_caches
+
+
+async def get_telegram_client(
+    bot_token: str,
+    *,
+    session: aiohttp.ClientSession | None = None,
+    thumbhash_resolver: Callable[[str], str | None] | None = None,
+) -> TelegramClient:
+    """Return a ``TelegramClient`` wired to shared session + shared caches.
+
+    Every Telegram send in the process should acquire its client from
+    here — constructing ``TelegramClient`` directly skips the caches and
+    silently halves cache hit rate.
+
+    Args:
+        bot_token: The bot's API token.
+        session: Optional explicit aiohttp session. Defaults to the
+            process-wide shared session.
+        thumbhash_resolver: Optional asset-id → thumbhash lookup. The
+            notification dispatcher passes one so asset-cache entries
+            invalidate on visual change; the command path doesn't need it
+            (commands always ask for a fresh result).
+    """
+    if session is None:
+        session = await get_http_session()
+    url_cache, asset_cache = await _get_telegram_caches()
+    return TelegramClient(
+        session, bot_token,
+        url_cache=url_cache,
+        asset_cache=asset_cache,
+        thumbhash_resolver=thumbhash_resolver,
+    )
+
+
+async def send_telegram_message(
+    bot_token: str,
+    chat_id: str,
+    text: str,
+    *,
+    reply_to_message_id: int | None = None,
+    disable_web_page_preview: bool = True,
+    parse_mode: str = "HTML",
+) -> NotificationResult:
+    """Send a plain-text Telegram message with caches wired in."""
+    client = await get_telegram_client(bot_token)
+    return await client.send_message(
+        chat_id, text,
+        reply_to_message_id=reply_to_message_id,
+        disable_web_page_preview=disable_web_page_preview,
+        parse_mode=parse_mode,
+    )
+
+
+async def send_telegram_media(
+    bot_token: str,
+    chat_id: str,
+    assets: list[dict[str, Any]],
+    *,
+    caption: str | None = None,
+    reply_to_message_id: int | None = None,
+    max_group_size: int = 10,
+    chunk_delay: int = 0,
+    max_asset_data_size: int | None = None,
+    send_large_photos_as_documents: bool = False,
+    chat_action: str | None = "typing",
+    thumbhash_resolver: Callable[[str], str | None] | None = None,
+) -> NotificationResult:
+    """Send a Telegram media group (or single asset) with caches wired in.
+
+    ``assets`` must be in ``TelegramClient`` format — see
+    ``notify_bridge_core.notifications.telegram.media.build_telegram_asset_entry``
+    for the canonical builder.
+    """
+    client = await get_telegram_client(
+        bot_token, thumbhash_resolver=thumbhash_resolver,
+    )
+    return await client.send_notification(
+        chat_id,
+        assets=assets,
+        caption=caption,
+        reply_to_message_id=reply_to_message_id,
+        max_group_size=max_group_size,
+        chunk_delay=chunk_delay,
+        max_asset_data_size=max_asset_data_size,
+        send_large_photos_as_documents=send_large_photos_as_documents,
+        chat_action=chat_action,
+    )