diff --git a/README.md b/README.md
index 1beb00b..92861e5 100644
--- a/README.md
+++ b/README.md
@@ -94,20 +94,24 @@ Run flags (`up` / `run`):
 
 Global flags: `--bridge-url`, `--name`, `--config-dir`.
 
-## Chat channels (WeChat)
+## Chat channels (WeChat & Telegram)
 
-Besides the browser/Nexior front-end, the daemon can be driven from a **personal
-WeChat account**: message the account and each message runs one Claude Code or
-Codex turn whose reply is sent straight back into the chat — handy for kicking
-off and steering tasks from your phone.
+Besides the browser/Nexior front-end, the daemon can be driven from a **chat
+account** — message it and each message runs one Claude Code or Codex turn whose
+reply is sent straight back into the chat, handy for kicking off and steering
+tasks from your phone. Two channels ship today and run side by side:
 
-A small **WeChat gateway** (a separate service you run, e.g. on a CVM) bridges
-WeChat and this daemon. The daemon connects _outbound_ to the gateway over a
-WebSocket, so — exactly like the relay path — it opens no listening ports and
-your code still only ever runs on your machine.
+- **WeChat** — via a small self-hosted gateway (below).
+- **Telegram** — via a bot you create with [@BotFather](https://t.me/BotFather),
+  talking to the official Bot API directly (no gateway to run).
+
+Both connect **outbound only** — like the relay path they open no listening
+ports, and your code still only ever runs on your machine.
 
 ```
-WeChat  ⇄  WeChat gateway  ⇄ (wss, outbound) ⇄  coding-bridge  ─►  Claude Code / Codex
+WeChat    ⭢  WeChat gateway  ⭢ (wss, outbound) ⭢
+                                                 ├─►  coding-bridge  ─►  Claude Code / Codex
+Telegram  ⭢  Bot API (getUpdates long-poll) ⭢───┘
 ```
 
 ### Setup
@@ -169,13 +173,13 @@ coding-bridge channels start    # connect and serve until Ctrl-C
 | Command  | What it does                                                       |
 | -------- | ------------------------------------------------------------------ |
 | `init`   | Write a skeleton `channels.toml` (refuses to overwrite)            |
-| `doctor` | Validate `channels.toml` and ping every enabled gateway endpoint   |
+| `doctor` | Validate `channels.toml` and ping every enabled channel            |
 | `smoke`  | Run one real provider turn locally to prove your provider works    |
-| `start`  | Connect to each enabled gateway and serve replies until Ctrl-C     |
+| `start`  | Connect to each enabled channel and serve replies until Ctrl-C     |
 | `portal` | Open a local web UI to edit `channels.toml` (pick admins, trigger mode) |
 
 After that one-time setup, day-to-day use is a **single command** —
-`coding-bridge channels start` — and you steer everything from WeChat.
+`coding-bridge channels start` — and you steer everything from WeChat or Telegram.
 
 `channels smoke` flags: `--provider {claude,codex,copilot}` (default `claude`),
 `--prompt` (default `"Reply with the single word: pong"`), `--timeout` seconds
@@ -184,6 +188,44 @@ After that one-time setup, day-to-day use is a **single command** —
 To keep `channels start` running across logout/reboot, install it as a service —
 templates in [docs/deploy/](docs/deploy/README.md).
 
+### Telegram
+
+Telegram needs no gateway — create a bot and point the daemon at it:
+
+1. Message [@BotFather](https://t.me/BotFather), send `/newbot`, and copy the
+   **bot token** it gives you.
+2. Add a `[[channels.telegram]]` block to `~/.ace-bridge/channels.toml`
+   (`coding-bridge channels init` writes a commented example):
+
+```toml
+[[channels.telegram]]
+instance_id = "my-telegram"                 # unique per instance
+token_env = "TELEGRAM_TOKEN_MY_TELEGRAM"    # env var holding the bot token (never the token itself)
+enabled = true                              # explicit opt-in
+require_approval = false                    # true = hold tool use for Approve/Deny in the portal
+
+trigger_prefix = "/ask "                    # prefix to require; "" = reply to every message
+allowed_senders = ["123456789"]             # numeric Telegram user ids; empty = allow all
+allowed_groups = []                         # group/supergroup chat ids (negative); empty = all groups
+rate_limit_per_min = 6                      # per-sender sliding window; 0 disables
+dedup_window_seconds = 300.0                # drop duplicate updates; 0 disables
+```
+
+Export the token, then verify and run — the **same** `doctor` / `start` commands
+drive every channel at once, WeChat and Telegram together:
+
+```bash
+export TELEGRAM_TOKEN_MY_TELEGRAM="123456:ABC-…"
+coding-bridge channels doctor   # getMe confirms the bot token is accepted
+coding-bridge channels start    # long-polls Telegram + serves WeChat, together
+```
+
+Find your own numeric id by messaging [@userinfobot](https://t.me/userinfobot). In
+a group, add the bot and keep a `trigger_prefix` so it only answers on `/ask …`
+(Telegram may also require disabling the bot's privacy mode in BotFather to see
+group messages). Running a self-hosted Bot API server? Set `api_base` on the
+block (defaults to `https://api.telegram.org`).
+
 ### Portal — edit the config in your browser
 
 Hand-editing `channels.toml` means knowing each admin's raw `wxid`. Instead, run
@@ -230,7 +272,7 @@ Bot:  You're on `main` with a clean working tree — nothing staged or modified.
 - **No content in logs.** Only sizes and outcome codes are recorded per turn — the
   message text and the reply body are never written to logs.
 - **Same local trust boundary.** Provider turns run on your machine under your
-  account; the gateway only relays messages.
+  account; the channel (WeChat gateway or Telegram Bot API) only carries messages.
 
 ## Configuration
 
diff --git a/coding_bridge/channels/__init__.py b/coding_bridge/channels/__init__.py
index 033c00b..5951fce 100644
--- a/coding_bridge/channels/__init__.py
+++ b/coding_bridge/channels/__init__.py
@@ -20,6 +20,7 @@
 from .config import (
     ChannelsConfig,
     ConfigError,
+    TelegramInstanceConfig,
     WeChatInstanceConfig,
     load_channels_config,
     parse_channels_config,
@@ -39,6 +40,7 @@
     "PolicyGate",
     "SendResult",
     "SessionDispatcher",
+    "TelegramInstanceConfig",
     "TurnEvent",
     "TurnOutcome",
     "WeChatInstanceConfig",
diff --git a/coding_bridge/channels/config.py b/coding_bridge/channels/config.py
index cc236fc..0fd2fa8 100644
--- a/coding_bridge/channels/config.py
+++ b/coding_bridge/channels/config.py
@@ -19,6 +19,13 @@
     enabled = false                      # default false — explicit opt-in
     default_provider = "claude"          # optional override for this instance
 
+    [[channels.telegram]]
+    instance_id = "my-bot"               # required, unique per file
+    token_env = "TELEGRAM_TOKEN_MY_BOT"  # required unless token_file is set
+    api_base = "https://api.telegram.org"  # optional; override for self-hosted
+    enabled = false                      # default false — explicit opt-in
+    default_provider = "claude"          # optional override for this instance
+
 Design notes:
 
 * The **token itself never lives in the TOML file** — the file references an
@@ -68,7 +75,24 @@
         "dedup_window_seconds",
     }
 )
-_CHANNELS_KEYS = frozenset({"wechat"})
+_CHANNELS_KEYS = frozenset({"wechat", "telegram"})
+
+_TELEGRAM_KEYS = frozenset(
+    {
+        "instance_id",
+        "api_base",
+        "token_env",
+        "token_file",
+        "enabled",
+        "default_provider",
+        "require_approval",
+        "trigger_prefix",
+        "allowed_senders",
+        "allowed_groups",
+        "rate_limit_per_min",
+        "dedup_window_seconds",
+    }
+)
 
 # Cap the TOML file size so a misconfigured `channels.toml` pointing at
 # `/dev/urandom` or similar can't OOM the daemon at startup.
@@ -83,6 +107,69 @@ class ConfigError(ValueError):
     """Raised for any schema violation — never caught silently."""
 
 
+def _resolve_token_source(
+    kind: str,
+    instance_id: str,
+    token_env: str | None,
+    token_file: str | None,
+    environ: dict[str, str] | None = None,
+) -> str:
+    """Load a token from an env var or a secrets file. Never logs the value.
+
+    Shared by every channel type so ``token_env`` / ``token_file`` behave (and
+    fail) identically. Error messages reference the env-var *name* or file
+    *path* only — never the token.
+    """
+    env = environ if environ is not None else dict(os.environ)
+    if token_env:
+        token = env.get(token_env)
+        if not token:
+            raise ConfigError(
+                f"{kind} instance {instance_id!r}: env var "
+                f"{token_env!r} is unset or empty"
+            )
+        return token
+    if token_file:
+        path = Path(token_file).expanduser()
+        try:
+            # Reject anything that isn't a regular file up front so a symlink to
+            # a directory / device raises a clean ConfigError, not a raw OSError.
+            if not path.is_file():
+                raise ConfigError(
+                    f"{kind} instance {instance_id!r}: token_file "
+                    f"{str(path)!r} is not a regular file"
+                )
+            size = path.stat().st_size
+            if size > _MAX_TOKEN_BYTES:
+                raise ConfigError(
+                    f"{kind} instance {instance_id!r}: token_file "
+                    f"{str(path)!r} exceeds {_MAX_TOKEN_BYTES}-byte limit"
+                )
+            raw = path.read_bytes()
+        except OSError as exc:
+            raise ConfigError(
+                f"{kind} instance {instance_id!r}: cannot read "
+                f"token_file {str(path)!r}: {exc.__class__.__name__}"
+            ) from None
+        try:
+            token = raw.decode("utf-8").strip()
+        except UnicodeDecodeError:
+            raise ConfigError(
+                f"{kind} instance {instance_id!r}: token_file "
+                f"{str(path)!r} is not valid UTF-8"
+            ) from None
+        if not token:
+            raise ConfigError(
+                f"{kind} instance {instance_id!r}: token_file "
+                f"{str(path)!r} is empty"
+            )
+        return token
+    raise ConfigError(
+        f"{kind} instance {instance_id!r}: either token_env or "
+        "token_file must be set"
+    )
+
+
 @dataclass(frozen=True)
 class WeChatInstanceConfig:
     """One `[[channels.wechat]]` block."""
@@ -129,60 +216,49 @@ def to_policy(self) -> ChannelPolicy:
         )
 
     def resolve_token(self, environ: dict[str, str] | None = None) -> str:
-        """Load the token from env var or secrets file. Never logs it.
-
-        Raises ``ConfigError`` with a **redacted** message if the token can't
-        be resolved — the message references the env-var *name* or file path,
-        never the value.
-        """
-        env = environ if environ is not None else dict(os.environ)
-        if self.token_env:
-            token = env.get(self.token_env)
-            if not token:
-                raise ConfigError(
-                    f"wechat instance {self.instance_id!r}: env var "
-                    f"{self.token_env!r} is unset or empty"
-                )
-            return token
-        if self.token_file:
-            path = Path(self.token_file).expanduser()
-            try:
-                # Reject anything that isn't a regular file up front so a
-                # symlink pointing at a directory / device raises a clean
-                # ConfigError instead of a raw OSError / UnicodeDecodeError.
-                if not path.is_file():
-                    raise ConfigError(
-                        f"wechat instance {self.instance_id!r}: token_file "
-                        f"{str(path)!r} is not a regular file"
-                    )
-                size = path.stat().st_size
-                if size > _MAX_TOKEN_BYTES:
-                    raise ConfigError(
-                        f"wechat instance {self.instance_id!r}: token_file "
-                        f"{str(path)!r} exceeds {_MAX_TOKEN_BYTES}-byte limit"
-                    )
-                raw = path.read_bytes()
-            except OSError as exc:
-                raise ConfigError(
-                    f"wechat instance {self.instance_id!r}: cannot read "
-                    f"token_file {str(path)!r}: {exc.__class__.__name__}"
-                ) from None
-            try:
-                token = raw.decode("utf-8").strip()
-            except UnicodeDecodeError:
-                raise ConfigError(
-                    f"wechat instance {self.instance_id!r}: token_file "
-                    f"{str(path)!r} is not valid UTF-8"
-                ) from None
-            if not token:
-                raise ConfigError(
-                    f"wechat instance {self.instance_id!r}: token_file "
-                    f"{str(path)!r} is empty"
-                )
-            return token
-        raise ConfigError(
-            f"wechat instance {self.instance_id!r}: either token_env or "
-            "token_file must be set"
+        """Load the token from env var or secrets file. Never logs it."""
+        return _resolve_token_source(
+            "wechat", self.instance_id, self.token_env, self.token_file, environ
+        )
+
+
+@dataclass(frozen=True)
+class TelegramInstanceConfig:
+    """One `[[channels.telegram]]` block (a Telegram bot, long-polling)."""
+
+    instance_id: str
+    token_env: str | None = None
+    token_file: str | None = None
+    #: Bot API host. Default is Telegram's; override for a self-hosted server.
+    api_base: str = "https://api.telegram.org"
+    enabled: bool = False
+    require_approval: bool = False
+    default_provider: str | None = None
+    trigger_prefix: str = "/ask "
+    #: Sender allowlist (Telegram numeric user ids as strings). Empty = allow all.
+    allowed_senders: tuple[str, ...] = ()
+    #: Group/supergroup chat ids (as strings, usually negative) the bot may
+    #: answer in. Empty = every group; never filters private DMs.
+    allowed_groups: tuple[str, ...] = ()
+    rate_limit_per_min: int = 6
+    dedup_window_seconds: float = 300.0
+
+    def to_policy(self) -> ChannelPolicy:
+        """Build the runtime ``ChannelPolicy`` this instance describes."""
+        from .policy import ChannelPolicy
+
+        return ChannelPolicy(
+            trigger_prefix=self.trigger_prefix,
+            allowed_senders=self.allowed_senders,
+            allowed_groups=self.allowed_groups,
+            rate_limit_per_min=self.rate_limit_per_min,
+            dedup_window_seconds=self.dedup_window_seconds,
+        )
+
+    def resolve_token(self, environ: dict[str, str] | None = None) -> str:
+        """Load the bot token from env var or secrets file. Never logs it."""
+        return _resolve_token_source(
+            "telegram", self.instance_id, self.token_env, self.token_file, environ
         )
 
 
@@ -191,11 +267,16 @@ class ChannelsConfig:
     """Root of the channels config file."""
 
     wechat: tuple[WeChatInstanceConfig, ...] = field(default_factory=tuple)
+    telegram: tuple[TelegramInstanceConfig, ...] = field(default_factory=tuple)
 
     @property
     def enabled_wechat(self) -> tuple[WeChatInstanceConfig, ...]:
         return tuple(inst for inst in self.wechat if inst.enabled)
 
+    @property
+    def enabled_telegram(self) -> tuple[TelegramInstanceConfig, ...]:
+        return tuple(inst for inst in self.telegram if inst.enabled)
+
 
 def _require(kind: str, block: dict[str, Any], key: str) -> Any:
     value = block.get(key)
@@ -337,6 +418,95 @@ def _parse_wechat(block: dict[str, Any], index: int) -> WeChatInstanceConfig:
     )
 
 
+def _parse_telegram(block: dict[str, Any], index: int) -> TelegramInstanceConfig:
+    kind = f"[[channels.telegram]] #{index}"
+    unknown = set(block.keys()) - _TELEGRAM_KEYS
+    if unknown:
+        raise ConfigError(f"{kind}: unknown key(s) {sorted(unknown)!r}")
+    instance_id = _require(kind, block, "instance_id")
+    if not isinstance(instance_id, str):
+        raise ConfigError(f"{kind}: instance_id must be a string")
+    tag = f"[[channels.telegram]] {instance_id!r}"
+
+    api_base = block.get("api_base", "https://api.telegram.org")
+    if not isinstance(api_base, str) or not (
+        api_base.startswith("http://") or api_base.startswith("https://")
+    ):
+        raise ConfigError(f"{tag}: api_base must start with http:// or https://")
+    if "@" in urlparse(api_base).netloc:
+        raise ConfigError(f"{tag}: api_base must not contain embedded credentials")
+
+    token_env = block.get("token_env")
+    token_file = block.get("token_file")
+    if token_env is not None and not isinstance(token_env, str):
+        raise ConfigError(f"{tag}: token_env must be a string")
+    if token_file is not None and not isinstance(token_file, str):
+        raise ConfigError(f"{tag}: token_file must be a string")
+    if token_env and token_file:
+        raise ConfigError(f"{tag}: set token_env OR token_file, not both")
+
+    enabled = block.get("enabled", False)
+    if not isinstance(enabled, bool):
+        raise ConfigError(f"{tag}: enabled must be a bool")
+    require_approval = block.get("require_approval", False)
+    if not isinstance(require_approval, bool):
+        raise ConfigError(f"{tag}: require_approval must be a bool")
+
+    default_provider = block.get("default_provider")
+    if default_provider is not None:
+        if not isinstance(default_provider, str):
+            raise ConfigError(f"{tag}: default_provider must be a string")
+        from ..providers import KNOWN_PROVIDERS
+
+        if default_provider not in KNOWN_PROVIDERS:
+            allowed = ", ".join(KNOWN_PROVIDERS)
+            raise ConfigError(
+                f"{tag}: unknown default_provider {default_provider!r} (allowed: {allowed})"
+            )
+
+    trigger_prefix = block.get("trigger_prefix", "/ask ")
+    if not isinstance(trigger_prefix, str):
+        raise ConfigError(f"{tag}: trigger_prefix must be a string")
+
+    allowed_senders_raw = block.get("allowed_senders", [])
+    if not isinstance(allowed_senders_raw, list) or not all(
+        isinstance(x, str) and x for x in allowed_senders_raw
+    ):
+        raise ConfigError(f"{tag}: allowed_senders must be a list of non-empty strings")
+    allowed_groups_raw = block.get("allowed_groups", [])
+    if not isinstance(allowed_groups_raw, list) or not all(
+        isinstance(x, str) and x for x in allowed_groups_raw
+    ):
+        raise ConfigError(f"{tag}: allowed_groups must be a list of non-empty strings")
+
+    rate_limit = block.get("rate_limit_per_min", 6)
+    if isinstance(rate_limit, bool) or not isinstance(rate_limit, int) or rate_limit < 0:
+        raise ConfigError(f"{tag}: rate_limit_per_min must be a non-negative int")
+
+    dedup_window = block.get("dedup_window_seconds", 300.0)
+    if isinstance(dedup_window, bool) or not isinstance(dedup_window, (int, float)):
+        raise ConfigError(f"{tag}: dedup_window_seconds must be a number")
+    if math.isnan(dedup_window) or math.isinf(dedup_window):
+        raise ConfigError(f"{tag}: dedup_window_seconds must be finite")
+    if dedup_window < 0:
+        raise ConfigError(f"{tag}: dedup_window_seconds must be >= 0")
+
+    return TelegramInstanceConfig(
+        instance_id=instance_id,
+        token_env=token_env,
+        token_file=token_file,
+        api_base=api_base.rstrip("/"),
+        enabled=enabled,
+        require_approval=require_approval,
+        default_provider=default_provider,
+        trigger_prefix=trigger_prefix,
+        allowed_senders=tuple(allowed_senders_raw),
+        allowed_groups=tuple(allowed_groups_raw),
+        rate_limit_per_min=rate_limit,
+        dedup_window_seconds=float(dedup_window),
+    )
+
+
 def parse_channels_config(data: dict[str, Any]) -> ChannelsConfig:
     """Parse an already-decoded TOML mapping. Public for test use."""
     channels = data.get("channels")
@@ -364,7 +534,24 @@ def parse_channels_config(data: dict[str, Any]) -> ChannelsConfig:
             )
         seen_ids.add(inst.instance_id)
         parsed.append(inst)
-    return ChannelsConfig(wechat=tuple(parsed))
+
+    tg_blocks = channels.get("telegram", [])
+    if not isinstance(tg_blocks, list):
+        raise ConfigError("[[channels.telegram]]: must be an array of tables")
+    tg_parsed: list[TelegramInstanceConfig] = []
+    tg_seen: set[str] = set()
+    for i, block in enumerate(tg_blocks):
+        if not isinstance(block, dict):
+            raise ConfigError(f"[[channels.telegram]] #{i}: must be a table")
+        tg = _parse_telegram(block, i)
+        if tg.instance_id in tg_seen:
+            raise ConfigError(
+                f"[[channels.telegram]]: duplicate instance_id {tg.instance_id!r}"
+            )
+        tg_seen.add(tg.instance_id)
+        tg_parsed.append(tg)
+
+    return ChannelsConfig(wechat=tuple(parsed), telegram=tuple(tg_parsed))
 
 
 def load_channels_config(path: Path) -> ChannelsConfig:
diff --git a/coding_bridge/channels/telegram/__init__.py b/coding_bridge/channels/telegram/__init__.py
new file mode 100644
index 0000000..443b9b8
--- /dev/null
+++ b/coding_bridge/channels/telegram/__init__.py
@@ -0,0 +1,8 @@
+"""Telegram channel adapter (long-polling Bot API)."""
+
+from __future__ import annotations
+
+from .adapter import TelegramAdapter
+from .client import TelegramClient, TelegramError
+
+__all__ = ["TelegramAdapter", "TelegramClient", "TelegramError"]
diff --git a/coding_bridge/channels/telegram/adapter.py b/coding_bridge/channels/telegram/adapter.py
new file mode 100644
index 0000000..0aeacca
--- /dev/null
+++ b/coding_bridge/channels/telegram/adapter.py
@@ -0,0 +1,200 @@
+"""``TelegramAdapter`` — long-polls the Telegram Bot API and dispatches messages.
+
+Like the WeChat adapter it's a *pure wire*: it runs a ``getUpdates`` long-poll
+loop (outbound HTTPS only — no inbound ports, no webhook), turns each ``message``
+update into an :class:`IncomingMessage`, and hands it to the dispatcher's
+handler. Policy (trigger prefix, sender/group allowlist, rate limit, dedup) lives
+in :class:`coding_bridge.channels.policy.PolicyGate`, shared with WeChat.
+
+Design notes:
+
+* One poll loop per instance — Telegram's ``getUpdates`` is undefined under
+  concurrent callers, so we never run two.
+* ``offset`` advances to ``last update_id + 1`` to acknowledge updates.
+* Recovers from a stray webhook (HTTP 409) by calling ``deleteWebhook``; honors
+  ``retry_after`` on 429; stops on 401 (bad token). Other errors back off
+  (0.5s → 30s).
+* The token only ever lives in the client's URL path and is never logged.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+from typing import Any
+
+from ..base import ChannelTarget, IncomingMessage, MessageHandler, SendResult
+from .client import TelegramClient, TelegramError
+
+logger = logging.getLogger("coding-bridge.channels.telegram")
+
+_INITIAL_BACKOFF_S = 0.5
+_MAX_BACKOFF_S = 30.0
+_DEFAULT_POLL_TIMEOUT_S = 30
+
+
+def _parse_update(update: dict[str, Any]) -> IncomingMessage | None:
+    """Turn one Telegram update into an ``IncomingMessage`` (or ``None`` to skip).
+
+    Only plain-text ``message`` updates are handled — edits, channel posts,
+    callbacks and non-text messages are dropped.
+    """
+    message = update.get("message")
+    if not isinstance(message, dict):
+        return None
+    text = message.get("text")
+    if not isinstance(text, str) or not text:
+        return None
+    chat = message.get("chat") if isinstance(message.get("chat"), dict) else {}
+    chat_id = chat.get("id")
+    if not isinstance(chat_id, int):
+        return None
+    frm = message.get("from") if isinstance(message.get("from"), dict) else {}
+    sender_raw = frm.get("id")
+    sender_id = str(sender_raw) if isinstance(sender_raw, int) else str(chat_id)
+    name = frm.get("username") or frm.get("first_name")
+    # Telegram: private == 1:1 DM; group/supergroup/channel are "group"-like so
+    # the shared allowed_groups gate keys off conversation_type == "group".
+    conv_type = "private" if chat.get("type") == "private" else "group"
+    message_id = message.get("message_id")
+    update_id = update.get("update_id")
+    return IncomingMessage(
+        sender_id=sender_id,
+        sender_name=name if isinstance(name, str) else None,
+        target=ChannelTarget(
+            conversation_id=str(chat_id),
+            conversation_type=conv_type,
+            reply_to_id=str(message_id) if isinstance(message_id, int) else None,
+        ),
+        text=text,
+        msg_type="text",
+        direction="inbound",
+        upstream_id=str(update_id) if isinstance(update_id, int) else None,
+        received_at_ms=(
+            int(message["date"]) * 1000 if isinstance(message.get("date"), int) else None
+        ),
+        raw=dict(message),
+    )
+
+
+class TelegramAdapter:
+    """One Telegram bot token = one instance = one long-poll loop."""
+
+    name = "telegram"
+
+    def __init__(
+        self,
+        *,
+        instance_id: str,
+        token: str,
+        api_base: str = "https://api.telegram.org",
+        client: TelegramClient | None = None,
+        stop_event: asyncio.Event | None = None,
+        poll_timeout: int = _DEFAULT_POLL_TIMEOUT_S,
+    ) -> None:
+        if not instance_id:
+            raise ValueError("instance_id must be a non-empty string")
+        if not token:
+            raise ValueError("token must be a non-empty string")
+        self.instance_id = instance_id
+        self._client = client or TelegramClient(token, api_base=api_base)
+        self._owns_client = client is None
+        self._stop = stop_event or asyncio.Event()
+        self._handler: MessageHandler | None = None
+        self._poll_timeout = poll_timeout
+
+    def set_handler(self, handler: MessageHandler) -> None:
+        self._handler = handler
+
+    async def run(self) -> None:
+        if self._handler is None:
+            raise RuntimeError("TelegramAdapter.run() called before set_handler()")
+        logger.info("telegram: polling instance=%s", self.instance_id)
+        offset = 0
+        backoff = _INITIAL_BACKOFF_S
+        while not self._stop.is_set():
+            try:
+                updates = await self._client.get_updates(offset, self._poll_timeout)
+                backoff = _INITIAL_BACKOFF_S
+                for update in updates:
+                    if not isinstance(update, dict):
+                        continue
+                    uid = update.get("update_id")
+                    if isinstance(uid, int):
+                        offset = uid + 1
+                    if self._stop.is_set():
+                        return
+                    msg = _parse_update(update)
+                    if msg is None:
+                        continue
+                    try:
+                        await self._handler(msg, self)
+                    except Exception:
+                        # A single handler failure must not kill the poll loop.
+                        logger.exception("telegram: handler error instance=%s", self.instance_id)
+            except asyncio.CancelledError:
+                raise
+            except TelegramError as exc:
+                if self._stop.is_set():
+                    return
+                if exc.error_code == 409:
+                    logger.warning(
+                        "telegram: 409 conflict instance=%s (webhook set); deleting it",
+                        self.instance_id,
+                    )
+                    with contextlib.suppress(Exception):
+                        await self._client.delete_webhook()
+                    continue
+                if exc.error_code == 401:
+                    logger.error(
+                        "telegram: unauthorized instance=%s (check the bot token); stopping",
+                        self.instance_id,
+                    )
+                    return
+                wait = exc.retry_after if (exc.error_code == 429 and exc.retry_after) else backoff
+                logger.warning(
+                    "telegram: api error instance=%s code=%s; retry in %.1fs",
+                    self.instance_id,
+                    exc.error_code,
+                    wait,
+                )
+                if await self._sleep_or_stop(wait):
+                    return
+                if exc.error_code != 429:
+                    backoff = min(backoff * 2, _MAX_BACKOFF_S)
+            except Exception as exc:
+                if self._stop.is_set():
+                    return
+                logger.warning(
+                    "telegram: poll error instance=%s err=%s; retry in %.1fs",
+                    self.instance_id,
+                    exc.__class__.__name__,
+                    backoff,
+                )
+                if await self._sleep_or_stop(backoff):
+                    return
+                backoff = min(backoff * 2, _MAX_BACKOFF_S)
+
+    async def _sleep_or_stop(self, seconds: float) -> bool:
+        """Sleep, or return True early if a stop was signaled."""
+        try:
+            await asyncio.wait_for(self._stop.wait(), timeout=seconds)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+    async def send(
+        self, target: ChannelTarget, text: str, *, reply_to: str | None = None
+    ) -> SendResult:
+        return await self._client.send_message(target, text, reply_to=reply_to)
+
+    async def aclose(self) -> None:
+        """Idempotent shutdown. Closing the client interrupts an in-flight poll."""
+        self._stop.set()
+        if self._owns_client:
+            with contextlib.suppress(Exception):
+                await self._client.aclose()
+
+
+__all__ = ["TelegramAdapter"]
diff --git a/coding_bridge/channels/telegram/client.py b/coding_bridge/channels/telegram/client.py
new file mode 100644
index 0000000..2bf9b07
--- /dev/null
+++ b/coding_bridge/channels/telegram/client.py
@@ -0,0 +1,148 @@
+"""Async Telegram Bot API client (long-polling). Mirrors the WeChat client shape.
+
+The bot token lives in the URL **path** (``/bot<token>/<method>``), so it must
+never be logged — this client never echoes the base URL or token in errors, and
+callers should log ``TelegramError.error_code`` (an int), not the message alone.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import time
+from typing import Any
+
+import httpx
+
+from ..base import ChannelTarget, SendResult
+
+# Telegram hard-caps a single message at 4096 chars (after entity parsing).
+_MAX_TEXT = 4096
+
+
+class TelegramError(Exception):
+    """A Telegram API ``{ok: false}`` response or a transport failure.
+
+    ``error_code`` carries Telegram's numeric code (401/403/409/429/…) so the
+    adapter's poll loop can react (delete webhook on 409, honor ``retry_after``
+    on 429, stop on 401). The message is bounded and token-free.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        error_code: int | None = None,
+        retry_after: float | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.error_code = error_code
+        self.retry_after = retry_after
+
+
+class TelegramClient:
+    """Async client for one Telegram bot token."""
+
+    def __init__(
+        self,
+        token: str,
+        *,
+        api_base: str = "https://api.telegram.org",
+        timeout: float = 15.0,
+        transport: httpx.AsyncBaseTransport | None = None,
+    ) -> None:
+        self._token = token
+        self._api_base = api_base.rstrip("/")
+        # The base URL embeds the secret token path segment — never logged.
+        self._client = httpx.AsyncClient(
+            base_url=f"{self._api_base}/bot{token}",
+            timeout=timeout,
+            transport=transport,
+        )
+
+    @property
+    def token(self) -> str:
+        return self._token
+
+    @property
+    def api_base(self) -> str:
+        return self._api_base
+
+    async def aclose(self) -> None:
+        await self._client.aclose()
+
+    async def _call(
+        self, method: str, payload: dict[str, Any], *, timeout: float | None = None
+    ) -> Any:
+        kwargs: dict[str, Any] = {"json": payload}
+        if timeout is not None:
+            kwargs["timeout"] = timeout
+        try:
+            resp = await self._client.post(f"/{method}", **kwargs)
+        except httpx.HTTPError as exc:
+            # Bound + type-only: a library that echoes the request URL must not
+            # leak the token path segment through str(exc).
+            raise TelegramError(f"transport error: {exc.__class__.__name__}") from None
+        try:
+            body = resp.json()
+        except (ValueError, TypeError):
+            raise TelegramError(
+                f"non-JSON response (HTTP {resp.status_code})", error_code=resp.status_code
+            ) from None
+        if isinstance(body, dict) and body.get("ok"):
+            return body.get("result")
+        code = body.get("error_code") if isinstance(body, dict) else resp.status_code
+        desc = body.get("description") if isinstance(body, dict) else ""
+        retry_after: float | None = None
+        params = body.get("parameters") if isinstance(body, dict) else None
+        if isinstance(params, dict) and isinstance(params.get("retry_after"), (int, float)):
+            retry_after = float(params["retry_after"])
+        raise TelegramError(
+            (str(desc)[:200] or f"telegram error {code}"),
+            error_code=int(code) if isinstance(code, int) else None,
+            retry_after=retry_after,
+        )
+
+    async def get_me(self) -> dict[str, Any]:
+        """Return the bot's own account (``id``/``username``/…). Used by doctor."""
+        result = await self._call("getMe", {})
+        return result if isinstance(result, dict) else {}
+
+    async def get_updates(self, offset: int, poll_timeout: int) -> list[dict[str, Any]]:
+        """Long-poll for new updates. Server holds up to ``poll_timeout`` seconds."""
+        result = await self._call(
+            "getUpdates",
+            {"offset": offset, "timeout": poll_timeout, "allowed_updates": ["message"]},
+            # Give httpx headroom over the server-side long-poll window.
+            timeout=float(poll_timeout) + 15.0,
+        )
+        return result if isinstance(result, list) else []
+
+    async def delete_webhook(self) -> None:
+        """Drop any webhook so ``getUpdates`` works (recovery from HTTP 409)."""
+        await self._call("deleteWebhook", {})
+
+    async def send_message(
+        self, target: ChannelTarget, text: str, *, reply_to: str | None = None
+    ) -> SendResult:
+        payload: dict[str, Any] = {"chat_id": target.conversation_id, "text": text[:_MAX_TEXT]}
+        rt = reply_to or target.reply_to_id
+        if rt:
+            with contextlib.suppress(TypeError, ValueError):
+                payload["reply_to_message_id"] = int(rt)
+        started = time.perf_counter()
+        try:
+            result = await self._call("sendMessage", payload)
+        except TelegramError as exc:
+            return SendResult(
+                ok=False,
+                error=f"{exc.__class__.__name__}: {str(exc)[:200]}",
+                latency_ms=int((time.perf_counter() - started) * 1000),
+            )
+        latency_ms = int((time.perf_counter() - started) * 1000)
+        upstream_id: str | None = None
+        if isinstance(result, dict) and isinstance(result.get("message_id"), int):
+            upstream_id = str(result["message_id"])
+        return SendResult(ok=True, upstream_id=upstream_id, latency_ms=latency_ms)
+
+
+__all__ = ["TelegramClient", "TelegramError"]
diff --git a/coding_bridge/channels_cli.py b/coding_bridge/channels_cli.py
index 9c8eea2..3956cfd 100644
--- a/coding_bridge/channels_cli.py
+++ b/coding_bridge/channels_cli.py
@@ -8,7 +8,8 @@
   ``PolicyGate → SessionDispatcher`` per enabled instance, and run the
   adapter loop until Ctrl-C.
 * ``coding-bridge channels doctor`` — validate ``channels.toml``, resolve
-  each token (without ever printing it), and ping each WeChat gateway endpoint.
+  each token (without ever printing it), and ping each channel (WeChat gateway
+  endpoint / Telegram ``getMe``).
 
 The dispatcher-wiring lives ONLY here — the `coding_bridge.channels.*`
 package stays pure library code that can be reused by other CLIs or by tests.
@@ -28,11 +29,12 @@
 import httpx
 
 from .channels import ConfigError, PolicyGate, load_channels_config
+from .channels.telegram import TelegramAdapter, TelegramClient, TelegramError
 from .channels.wechat import WeChatAdapter, WeChatClient
 from .config import Settings
 
 if TYPE_CHECKING:
-    from .channels import ChannelAdapter, WeChatInstanceConfig
+    from .channels import ChannelAdapter, TelegramInstanceConfig, WeChatInstanceConfig
     from .channels.dispatcher import SessionDispatcher
 
 logger = logging.getLogger("coding-bridge.channels.cli")
@@ -65,6 +67,31 @@
 #
 # # Drop repeat msg_id inside this window (upstream retries). 0 disables.
 # dedup_window_seconds = 300.0
+
+# [[channels.telegram]]
+# instance_id = "my-telegram"
+# # Create a bot with @BotFather, then export the token it hands you:
+# token_env = "TELEGRAM_TOKEN_MY_TELEGRAM"
+# enabled = false
+#
+# # Only respond when a message starts with this prefix. In groups, add the bot
+# # and keep a prefix so it doesn't answer every line. Empty answers everything.
+# trigger_prefix = "/ask "
+#
+# # Only accept messages from these Telegram numeric user IDs (as strings).
+# # Empty list means allow all — safe only for a private bot. Message
+# # @userinfobot to find your own id.
+# allowed_senders = []
+#
+# # Group / supergroup chat IDs (usually negative, as strings) the bot may
+# # answer in. Empty = every group it's added to. Never filters private DMs.
+# allowed_groups = []
+#
+# # At most this many messages per sender per 60 s. 0 disables the limit.
+# rate_limit_per_min = 6
+#
+# # Drop duplicate updates inside this window. 0 disables.
+# dedup_window_seconds = 300.0
 """
 
 
@@ -102,9 +129,9 @@ def cmd_channels_init(settings: Settings) -> int:
         return 1
     print(f"Wrote {path}")
     print("Next steps:")
-    print("  1. Uncomment the [[channels.wechat]] block and fill in instance_id + base_url.")
+    print("  1. Uncomment a [[channels.wechat]] or [[channels.telegram]] block.")
     print("  2. Export the token env var referenced by `token_env`.")
-    print("  3. Fill `allowed_senders` with your own WeChat wxid.")
+    print("  3. Fill `allowed_senders` with your own sender id(s).")
     print("  4. Flip `enabled = true`.")
     print("  5. Run `coding-bridge channels doctor` to verify.")
     print("  6. Run `coding-bridge channels start`.")
@@ -177,6 +204,34 @@ async def _doctor_one(inst: WeChatInstanceConfig) -> tuple[bool, str]:
         await client.aclose()
 
 
+async def _doctor_one_telegram(inst: TelegramInstanceConfig) -> tuple[bool, str]:
+    """Return (ok, message) for one Telegram instance. Never logs the token.
+
+    Calls ``getMe`` — Telegram's canonical auth probe. 200 → token good (report
+    the bot's ``@username``); 401 → token rejected; anything else → unreachable
+    or transient server error.
+    """
+    try:
+        token = inst.resolve_token()
+    except ConfigError as exc:
+        return False, f"token: {exc}"
+
+    client = TelegramClient(token, api_base=inst.api_base, timeout=8.0)
+    try:
+        me = await client.get_me()
+        username = me.get("username")
+        who = f"@{username}" if username else f"id={me.get('id')}"
+        return True, f"OK (token accepted, bot {who})"
+    except TelegramError as exc:
+        if exc.error_code == 401:
+            return False, "token rejected (401)"
+        return False, f"api error {exc.error_code}"
+    except Exception as exc:  # noqa: BLE001 - diagnostic; class name only
+        return False, f"unreachable: {exc.__class__.__name__}"
+    finally:
+        await client.aclose()
+
+
 def cmd_channels_doctor(settings: Settings) -> int:
     """Load config + check each instance's token + reachability."""
     path = settings.channels_config_path
@@ -186,12 +241,14 @@ def cmd_channels_doctor(settings: Settings) -> int:
         print(f"config error in {path}: {exc}", file=sys.stderr)
         return 2
 
-    if not cfg.wechat:
+    if not cfg.wechat and not cfg.telegram:
         print(f"No channels configured in {path}. Run `channels init` first.")
         return 0
 
+    total = len(cfg.wechat) + len(cfg.telegram)
+    enabled = len(cfg.enabled_wechat) + len(cfg.enabled_telegram)
     print(f"Config file: {path}")
-    print(f"Total instances: {len(cfg.wechat)} (enabled: {len(cfg.enabled_wechat)})")
+    print(f"Total instances: {total} (enabled: {enabled})")
     print()
 
     all_ok = True
@@ -200,7 +257,7 @@ async def _run() -> None:
         nonlocal all_ok
         for inst in cfg.wechat:
             state = "ENABLED" if inst.enabled else "disabled"
-            print(f"[{state}] {inst.instance_id}: {inst.base_url}")
+            print(f"[{state}] wechat/{inst.instance_id}: {inst.base_url}")
             if not inst.enabled:
                 print("  → skipped (enabled=false)")
                 continue
@@ -209,6 +266,17 @@ async def _run() -> None:
             print(f"  {marker} {msg}")
             if not ok:
                 all_ok = False
+        for tg in cfg.telegram:
+            state = "ENABLED" if tg.enabled else "disabled"
+            print(f"[{state}] telegram/{tg.instance_id}: {tg.api_base}")
+            if not tg.enabled:
+                print("  → skipped (enabled=false)")
+                continue
+            ok, msg = await _doctor_one_telegram(tg)
+            marker = "✓" if ok else "✗"
+            print(f"  {marker} {msg}")
+            if not ok:
+                all_ok = False
 
     asyncio.run(_run())
     return 0 if all_ok else 1
@@ -232,27 +300,37 @@ def cmd_channels_start(settings: Settings) -> int:
         print(f"config error in {path}: {exc}", file=sys.stderr)
         return 2
 
-    enabled = cfg.enabled_wechat
-    if not enabled:
+    enabled_wechat = cfg.enabled_wechat
+    enabled_telegram = cfg.enabled_telegram
+    if not enabled_wechat and not enabled_telegram:
         print(
             "No enabled channels — set `enabled = true` on at least one "
-            "[[channels.wechat]] block in " + str(path),
+            "[[channels.wechat]] or [[channels.telegram]] block in " + str(path),
             file=sys.stderr,
         )
         return 1
 
     # Resolve tokens up front so a missing env var kills us before any
     # network connection instead of silently.
-    resolved: list[tuple[WeChatInstanceConfig, str]] = []
-    for inst in enabled:
+    wechat_resolved: list[tuple[WeChatInstanceConfig, str]] = []
+    telegram_resolved: list[tuple[TelegramInstanceConfig, str]] = []
+    for inst in enabled_wechat:
+        try:
+            token = inst.resolve_token()
+        except ConfigError as exc:
+            print(f"config error: {exc}", file=sys.stderr)
+            return 2
+        wechat_resolved.append((inst, token))
+    for inst in enabled_telegram:
         try:
             token = inst.resolve_token()
         except ConfigError as exc:
             print(f"config error: {exc}", file=sys.stderr)
             return 2
-        resolved.append((inst, token))
+        telegram_resolved.append((inst, token))
 
     provider_factory = default_provider_factory(settings)
+    total = len(wechat_resolved) + len(telegram_resolved)
 
     async def _go() -> None:
         from .channels.approvals import ApprovalStore
@@ -264,16 +342,19 @@ async def _go() -> None:
         # Shared across instances; only used by instances with require_approval.
         approval_store = ApprovalStore(settings.config_dir / "approvals")
 
-        for inst, token in resolved:
-            dispatcher = SessionDispatcher(
+        def _make_dispatcher(inst: object) -> SessionDispatcher:
+            return SessionDispatcher(
                 settings,
                 provider_factory,
-                default_provider=inst.default_provider or "claude",
+                default_provider=getattr(inst, "default_provider", None) or "claude",
                 approval_store=approval_store,
-                require_approval=inst.require_approval,
+                require_approval=getattr(inst, "require_approval", False),
             )
+
+        for inst, token in wechat_resolved:
+            dispatcher = _make_dispatcher(inst)
             gate = PolicyGate(inst.to_policy(), dispatcher.handle_message)
-            adapter = WeChatAdapter(
+            adapter: ChannelAdapter = WeChatAdapter(
                 instance_id=inst.instance_id,
                 base_url=inst.base_url,
                 token=token,
@@ -283,7 +364,22 @@ async def _go() -> None:
             dispatchers.append(dispatcher)
             adapters.append(adapter)
             runners.append(asyncio.create_task(adapter.run()))
-            print(f"channel started: {inst.instance_id} → {inst.base_url}")
+            print(f"channel started: wechat/{inst.instance_id} → {inst.base_url}")
+
+        for inst, token in telegram_resolved:
+            dispatcher = _make_dispatcher(inst)
+            gate = PolicyGate(inst.to_policy(), dispatcher.handle_message)
+            tg_adapter: ChannelAdapter = TelegramAdapter(
+                instance_id=inst.instance_id,
+                token=token,
+                api_base=inst.api_base,
+                stop_event=stop,
+            )
+            tg_adapter.set_handler(gate.handle)
+            dispatchers.append(dispatcher)
+            adapters.append(tg_adapter)
+            runners.append(asyncio.create_task(tg_adapter.run()))
+            print(f"channel started: telegram/{inst.instance_id} → {inst.api_base}")
 
         try:
             # Wait for any runner to exit (auth failure, etc.) or Ctrl-C.
@@ -307,7 +403,7 @@ async def _go() -> None:
                 with contextlib.suppress(BaseException):
                     await t
 
-    print(f"Starting {len(resolved)} channel(s). Ctrl-C to stop.")
+    print(f"Starting {total} channel(s). Ctrl-C to stop.")
     try:
         asyncio.run(_go())
     except KeyboardInterrupt:
diff --git a/pyproject.toml b/pyproject.toml
index 8dac349..193506f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -32,6 +32,7 @@ qr = ["qrcode>=7.4"]
 # these expose intent (`pip install coding-bridge[wechat]`) without pulling
 # additional wheels.
 wechat = []
+telegram = []
 dev = [
     "pytest>=8",
     "pytest-asyncio>=0.23",
diff --git a/tests/test_channels_telegram.py b/tests/test_channels_telegram.py
new file mode 100644
index 0000000..9512ece
--- /dev/null
+++ b/tests/test_channels_telegram.py
@@ -0,0 +1,680 @@
+"""Telegram channel: update parsing, client (mock transport), adapter poll loop, config."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+import httpx
+import pytest
+
+from coding_bridge.channels import ChannelTarget, IncomingMessage, SendResult
+from coding_bridge.channels.config import (
+    ConfigError,
+    TelegramInstanceConfig,
+    load_channels_config,
+    parse_channels_config,
+)
+from coding_bridge.channels.telegram import TelegramAdapter, TelegramClient, TelegramError
+from coding_bridge.channels.telegram.adapter import _parse_update
+
+# ---------------------------------------------------------------------------
+# _parse_update — pure, no network
+# ---------------------------------------------------------------------------
+
+
+def _update(
+    update_id: int = 1,
+    *,
+    text: str = "/ask hi",
+    chat_id: int = 100,
+    chat_type: str = "private",
+    from_id: int | None = 555,
+    username: str | None = "alice",
+    message_id: int = 7,
+    date: int = 1_700_000_000,
+) -> dict[str, Any]:
+    msg: dict[str, Any] = {
+        "message_id": message_id,
+        "chat": {"id": chat_id, "type": chat_type},
+        "text": text,
+        "date": date,
+    }
+    if from_id is not None:
+        msg["from"] = {"id": from_id, "username": username}
+    return {"update_id": update_id, "message": msg}
+
+
+def test_parse_update_private_happy() -> None:
+    m = _parse_update(_update())
+    assert isinstance(m, IncomingMessage)
+    assert m.sender_id == "555"
+    assert m.sender_name == "alice"
+    assert m.target.conversation_id == "100"
+    assert m.target.conversation_type == "private"
+    assert m.target.reply_to_id == "7"
+    assert m.text == "/ask hi"
+    assert m.upstream_id == "1"
+    assert m.received_at_ms == 1_700_000_000 * 1000
+
+
+def test_parse_update_group_is_group_type() -> None:
+    m = _parse_update(_update(chat_id=-1001, chat_type="supergroup"))
+    assert m is not None
+    assert m.target.conversation_type == "group"
+    assert m.target.conversation_id == "-1001"
+
+
+def test_parse_update_sender_falls_back_to_chat_id() -> None:
+    m = _parse_update(_update(from_id=None))
+    assert m is not None
+    assert m.sender_id == "100"
+    assert m.sender_name is None
+
+
+def test_parse_update_uses_first_name_when_no_username() -> None:
+    upd = _update()
+    upd["message"]["from"] = {"id": 9, "first_name": "Bob"}
+    m = _parse_update(upd)
+    assert m is not None
+    assert m.sender_name == "Bob"
+
+
+@pytest.mark.parametrize(
+    "upd",
+    [
+        pytest.param({"update_id": 1}, id="no-message"),
+        pytest.param({"update_id": 1, "message": "x"}, id="message-not-dict"),
+        pytest.param(
+            {"update_id": 1, "message": {"chat": {"id": 1, "type": "private"}}},
+            id="no-text",
+        ),
+        pytest.param(
+            {"update_id": 1, "message": {"text": "", "chat": {"id": 1, "type": "private"}}},
+            id="empty-text",
+        ),
+        pytest.param({"update_id": 1, "message": {"text": "hi"}}, id="no-chat"),
+        pytest.param(
+            {"update_id": 1, "message": {"text": "hi", "chat": {"id": "x", "type": "private"}}},
+            id="chat-id-not-int",
+        ),
+        pytest.param(
+            {
+                "update_id": 1,
+                "edited_message": {"text": "hi", "chat": {"id": 1, "type": "private"}},
+            },
+            id="edited-not-message",
+        ),
+    ],
+)
+def test_parse_update_drops_bad(upd: dict[str, Any]) -> None:
+    assert _parse_update(upd) is None
+
+
+# ---------------------------------------------------------------------------
+# TelegramClient — httpx.MockTransport (token lives in URL path, never asserted)
+# ---------------------------------------------------------------------------
+
+
+def _transport(routes: dict[str, Any]) -> httpx.MockTransport:
+    def _handler(request: httpx.Request) -> httpx.Response:
+        method = request.url.path.rsplit("/", 1)[-1]
+        entry = routes[method]
+        return entry(request) if callable(entry) else entry
+
+    return httpx.MockTransport(_handler)
+
+
+@pytest.mark.asyncio
+async def test_client_get_me_returns_result() -> None:
+    tr = _transport(
+        {"getMe": httpx.Response(200, json={"ok": True, "result": {"id": 42, "username": "mybot"}})}
+    )
+    c = TelegramClient("SEKRIT", transport=tr)
+    try:
+        me = await c.get_me()
+    finally:
+        await c.aclose()
+    assert me == {"id": 42, "username": "mybot"}
+
+
+@pytest.mark.asyncio
+async def test_client_send_message_posts_chat_text_and_reply() -> None:
+    seen: dict[str, Any] = {}
+
+    def _send(request: httpx.Request) -> httpx.Response:
+        seen["body"] = json.loads(request.content)
+        seen["path"] = request.url.path
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 99}})
+
+    tr = _transport({"sendMessage": _send})
+    c = TelegramClient("t", transport=tr)
+    try:
+        r = await c.send_message(ChannelTarget(conversation_id="100", reply_to_id="7"), "hello")
+    finally:
+        await c.aclose()
+    assert r.ok is True
+    assert r.upstream_id == "99"
+    assert seen["body"] == {"chat_id": "100", "text": "hello", "reply_to_message_id": 7}
+    assert seen["path"].endswith("/sendMessage")
+
+
+@pytest.mark.asyncio
+async def test_client_send_truncates_to_4096() -> None:
+    seen: dict[str, Any] = {}
+
+    def _send(request: httpx.Request) -> httpx.Response:
+        seen["body"] = json.loads(request.content)
+        return httpx.Response(200, json={"ok": True, "result": {"message_id": 1}})
+
+    tr = _transport({"sendMessage": _send})
+    c = TelegramClient("t", transport=tr)
+    try:
+        await c.send_message(ChannelTarget(conversation_id="1"), "x" * 5000)
+    finally:
+        await c.aclose()
+    assert len(seen["body"]["text"]) == 4096
+
+
+@pytest.mark.asyncio
+async def test_client_get_updates_sends_offset_and_returns_list() -> None:
+    seen: dict[str, Any] = {}
+
+    def _upd(request: httpx.Request) -> httpx.Response:
+        seen["body"] = json.loads(request.content)
+        return httpx.Response(200, json={"ok": True, "result": [{"update_id": 5}]})
+
+    tr = _transport({"getUpdates": _upd})
+    c = TelegramClient("t", transport=tr)
+    try:
+        res = await c.get_updates(5, 30)
+    finally:
+        await c.aclose()
+    assert res == [{"update_id": 5}]
+    assert seen["body"]["offset"] == 5
+    assert seen["body"]["allowed_updates"] == ["message"]
+
+
+@pytest.mark.asyncio
+async def test_client_ok_false_raises_with_code_and_no_token() -> None:
+    tr = _transport(
+        {
+            "getMe": httpx.Response(
+                200, json={"ok": False, "error_code": 401, "description": "Unauthorized"}
+            )
+        }
+    )
+    c = TelegramClient("supersecrettoken", transport=tr)
+    try:
+        with pytest.raises(TelegramError) as ei:
+            await c.get_me()
+    finally:
+        await c.aclose()
+    assert ei.value.error_code == 401
+    assert "supersecrettoken" not in str(ei.value)
+
+
+@pytest.mark.asyncio
+async def test_client_429_carries_retry_after() -> None:
+    tr = _transport(
+        {
+            "getUpdates": httpx.Response(
+                200,
+                json={
+                    "ok": False,
+                    "error_code": 429,
+                    "description": "Too Many Requests",
+                    "parameters": {"retry_after": 3},
+                },
+            )
+        }
+    )
+    c = TelegramClient("t", transport=tr)
+    try:
+        with pytest.raises(TelegramError) as ei:
+            await c.get_updates(0, 30)
+    finally:
+        await c.aclose()
+    assert ei.value.error_code == 429
+    assert ei.value.retry_after == 3.0
+
+
+@pytest.mark.asyncio
+async def test_client_send_transport_error_returns_result_no_token() -> None:
+    def _boom(_request: httpx.Request) -> httpx.Response:
+        raise httpx.ConnectError("no route")
+
+    tr = _transport({"sendMessage": _boom})
+    c = TelegramClient("supersecrettoken", transport=tr)
+    try:
+        r = await c.send_message(ChannelTarget(conversation_id="1"), "hi")
+    finally:
+        await c.aclose()
+    assert r.ok is False
+    assert "supersecrettoken" not in (r.error or "")
+
+
+@pytest.mark.asyncio
+async def test_client_delete_webhook_calls_endpoint() -> None:
+    seen: dict[str, Any] = {}
+
+    def _dw(_request: httpx.Request) -> httpx.Response:
+        seen["called"] = True
+        return httpx.Response(200, json={"ok": True, "result": True})
+
+    tr = _transport({"deleteWebhook": _dw})
+    c = TelegramClient("t", transport=tr)
+    try:
+        await c.delete_webhook()
+    finally:
+        await c.aclose()
+    assert seen.get("called") is True
+
+
+# ---------------------------------------------------------------------------
+# TelegramAdapter — poll loop with an injected fake client
+# ---------------------------------------------------------------------------
+
+
+class _FakeClient:
+    """Scripts ``get_updates`` batches, then blocks until stop (mimics long-poll)."""
+
+    def __init__(self, stop: asyncio.Event, script: list[Any]) -> None:
+        self._stop = stop
+        self._script = list(script)
+        self.offsets: list[int] = []
+        self.deleted = 0
+        self.sent: list[tuple[str, str, str | None]] = []
+        self.closed = False
+
+    async def get_updates(self, offset: int, poll_timeout: int) -> list[dict[str, Any]]:
+        self.offsets.append(offset)
+        if self._script:
+            item = self._script.pop(0)
+            if isinstance(item, BaseException):
+                raise item
+            return item
+        await self._stop.wait()
+        return []
+
+    async def delete_webhook(self) -> None:
+        self.deleted += 1
+
+    async def send_message(
+        self, target: ChannelTarget, text: str, *, reply_to: str | None = None
+    ) -> SendResult:
+        self.sent.append((target.conversation_id, text, reply_to))
+        return SendResult(ok=True, upstream_id="1", latency_ms=0)
+
+    async def aclose(self) -> None:
+        self.closed = True
+
+
+def _mk_adapter(stop: asyncio.Event, script: list[Any], **kw: Any) -> TelegramAdapter:
+    return TelegramAdapter(
+        instance_id="tg1", token="t", client=_FakeClient(stop, script), stop_event=stop, **kw
+    )
+
+
+async def _run_briefly(adapter: TelegramAdapter, stop: asyncio.Event, delay: float = 0.05) -> None:
+    async def _drive() -> None:
+        await asyncio.sleep(delay)
+        stop.set()
+
+    await asyncio.gather(adapter.run(), _drive())
+    await adapter.aclose()
+
+
+@pytest.mark.asyncio
+async def test_adapter_dispatches_one_update() -> None:
+    stop = asyncio.Event()
+    seen: list[IncomingMessage] = []
+
+    async def handler(msg: IncomingMessage, _a: object) -> None:
+        seen.append(msg)
+
+    adapter = _mk_adapter(stop, [[_update(text="ping")]])
+    adapter.set_handler(handler)
+    await _run_briefly(adapter, stop)
+    assert [m.text for m in seen] == ["ping"]
+    assert seen[0].sender_id == "555"
+
+
+@pytest.mark.asyncio
+async def test_adapter_skips_bad_updates() -> None:
+    stop = asyncio.Event()
+    seen: list[IncomingMessage] = []
+
+    async def handler(msg: IncomingMessage, _a: object) -> None:
+        seen.append(msg)
+
+    script = [
+        [
+            {"update_id": 1, "edited_message": {"text": "x", "chat": {"id": 1, "type": "private"}}},
+            {"update_id": 2, "message": {"chat": {"id": 1, "type": "private"}}},
+            _update(update_id=3, text="real"),
+        ]
+    ]
+    adapter = _mk_adapter(stop, script)
+    adapter.set_handler(handler)
+    await _run_briefly(adapter, stop)
+    assert [m.text for m in seen] == ["real"]
+
+
+@pytest.mark.asyncio
+async def test_adapter_advances_offset() -> None:
+    stop = asyncio.Event()
+
+    async def handler(_m: IncomingMessage, _a: object) -> None:
+        return None
+
+    fake = _FakeClient(stop, [[_update(update_id=41)]])
+    adapter = TelegramAdapter(instance_id="tg", token="t", client=fake, stop_event=stop)
+    adapter.set_handler(handler)
+    await _run_briefly(adapter, stop)
+    assert 42 in fake.offsets
+
+
+@pytest.mark.asyncio
+async def test_adapter_stops_on_401() -> None:
+    stop = asyncio.Event()
+
+    async def handler(_m: IncomingMessage, _a: object) -> None:
+        return None
+
+    adapter = _mk_adapter(stop, [TelegramError("unauthorized", error_code=401)])
+    adapter.set_handler(handler)
+    # Returns on its own — no stop.set() needed. Would hang (→ TimeoutError) if not.
+    await asyncio.wait_for(adapter.run(), timeout=1.0)
+    await adapter.aclose()
+
+
+@pytest.mark.asyncio
+async def test_adapter_409_deletes_webhook_then_continues() -> None:
+    stop = asyncio.Event()
+    seen: list[IncomingMessage] = []
+
+    async def handler(msg: IncomingMessage, _a: object) -> None:
+        seen.append(msg)
+
+    fake = _FakeClient(
+        stop, [TelegramError("conflict", error_code=409), [_update(text="after")]]
+    )
+    adapter = TelegramAdapter(instance_id="tg", token="t", client=fake, stop_event=stop)
+    adapter.set_handler(handler)
+    await _run_briefly(adapter, stop, delay=0.1)
+    assert fake.deleted == 1
+    assert [m.text for m in seen] == ["after"]
+
+
+@pytest.mark.asyncio
+async def test_adapter_handler_error_does_not_kill_loop() -> None:
+    stop = asyncio.Event()
+    seen: list[IncomingMessage] = []
+    calls = 0
+
+    async def handler(msg: IncomingMessage, _a: object) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("boom")
+        seen.append(msg)
+
+    adapter = _mk_adapter(
+        stop, [[_update(update_id=1, text="one"), _update(update_id=2, text="two")]]
+    )
+    adapter.set_handler(handler)
+    await _run_briefly(adapter, stop)
+    assert [m.text for m in seen] == ["two"]
+
+
+@pytest.mark.asyncio
+async def test_adapter_run_without_handler_raises() -> None:
+    stop = asyncio.Event()
+    adapter = _mk_adapter(stop, [])
+    with pytest.raises(RuntimeError):
+        await adapter.run()
+
+
+@pytest.mark.asyncio
+async def test_adapter_send_delegates_to_client() -> None:
+    stop = asyncio.Event()
+    fake = _FakeClient(stop, [])
+    adapter = TelegramAdapter(instance_id="tg", token="t", client=fake, stop_event=stop)
+    r = await adapter.send(ChannelTarget(conversation_id="100"), "hi", reply_to="7")
+    assert r.ok is True
+    assert fake.sent == [("100", "hi", "7")]
+
+
+@pytest.mark.asyncio
+async def test_adapter_aclose_does_not_close_injected_client() -> None:
+    stop = asyncio.Event()
+    fake = _FakeClient(stop, [])
+    adapter = TelegramAdapter(instance_id="tg", token="t", client=fake, stop_event=stop)
+    await adapter.aclose()
+    # Injected client is caller-owned — the adapter must not close it.
+    assert fake.closed is False
+
+
+def test_adapter_rejects_empty_instance_id() -> None:
+    with pytest.raises(ValueError):
+        TelegramAdapter(instance_id="", token="t")
+
+
+def test_adapter_rejects_empty_token() -> None:
+    with pytest.raises(ValueError):
+        TelegramAdapter(instance_id="x", token="")
+
+
+# ---------------------------------------------------------------------------
+# Config — [[channels.telegram]] parsing + validation
+# ---------------------------------------------------------------------------
+
+
+def _write(tmp_path: Path, body: str) -> Path:
+    p = tmp_path / "channels.toml"
+    p.write_text(body, encoding="utf-8")
+    return p
+
+
+class TestTelegramConfig:
+    def test_single_instance_token_env(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "bot1"
+token_env = "TG_TOKEN"
+enabled = true
+""",
+        )
+        cfg = load_channels_config(p)
+        assert len(cfg.telegram) == 1
+        inst = cfg.telegram[0]
+        assert inst.instance_id == "bot1"
+        assert inst.api_base == "https://api.telegram.org"
+        assert inst.token_env == "TG_TOKEN"
+        assert inst.enabled is True
+        assert inst.trigger_prefix == "/ask "
+        assert cfg.enabled_telegram == (inst,)
+
+    def test_api_base_override_trailing_slash_stripped(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+api_base = "https://tg.example.com/"
+""",
+        )
+        assert load_channels_config(p).telegram[0].api_base == "https://tg.example.com"
+
+    def test_defaults(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+""",
+        )
+        inst = load_channels_config(p).telegram[0]
+        assert inst.enabled is False
+        assert inst.require_approval is False
+        assert inst.rate_limit_per_min == 6
+        assert inst.dedup_window_seconds == 300.0
+        assert inst.allowed_senders == ()
+        assert inst.allowed_groups == ()
+
+    def test_to_policy_maps_fields(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+trigger_prefix = "!go "
+allowed_senders = ["555", "666"]
+allowed_groups = ["-100"]
+rate_limit_per_min = 10
+dedup_window_seconds = 120.0
+""",
+        )
+        pol = load_channels_config(p).telegram[0].to_policy()
+        assert pol.trigger_prefix == "!go "
+        assert pol.allowed_senders == ("555", "666")
+        assert pol.allowed_groups == ("-100",)
+        assert pol.rate_limit_per_min == 10
+        assert pol.dedup_window_seconds == 120.0
+
+    def test_resolve_token_from_env(self) -> None:
+        inst = TelegramInstanceConfig(instance_id="b", token_env="TG_X")
+        assert inst.resolve_token({"TG_X": "abc"}) == "abc"
+
+    def test_resolve_token_missing_env_raises_without_leaking(self) -> None:
+        inst = TelegramInstanceConfig(instance_id="b", token_env="TG_X")
+        with pytest.raises(ConfigError, match="unset or empty"):
+            inst.resolve_token({})
+
+    def test_duplicate_instance_id_raises(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "dup"
+token_env = "A"
+
+[[channels.telegram]]
+instance_id = "dup"
+token_env = "B"
+""",
+        )
+        with pytest.raises(ConfigError, match="duplicate instance_id"):
+            load_channels_config(p)
+
+    def test_unknown_key_raises(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+webhook_url = "https://x"
+""",
+        )
+        with pytest.raises(ConfigError, match="unknown key"):
+            load_channels_config(p)
+
+    def test_token_env_and_file_conflict(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+token_file = "/tmp/x"
+""",
+        )
+        with pytest.raises(ConfigError, match="not both"):
+            load_channels_config(p)
+
+    def test_unknown_provider_raises(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+default_provider = "not-real"
+""",
+        )
+        with pytest.raises(ConfigError, match="unknown default_provider"):
+            load_channels_config(p)
+
+    def test_bad_api_base_scheme_raises(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+api_base = "ftp://x"
+""",
+        )
+        with pytest.raises(ConfigError, match="api_base must start with"):
+            load_channels_config(p)
+
+    def test_api_base_with_userinfo_raises(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+api_base = "https://user:pass@tg.example.com"
+""",
+        )
+        with pytest.raises(ConfigError, match="embedded credentials"):
+            load_channels_config(p)
+
+    def test_allowed_senders_must_be_strings(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.telegram]]
+instance_id = "b"
+token_env = "T"
+allowed_senders = [555]
+""",
+        )
+        with pytest.raises(ConfigError, match="allowed_senders"):
+            load_channels_config(p)
+
+    def test_telegram_not_array_raises(self) -> None:
+        with pytest.raises(ConfigError, match="array of tables"):
+            parse_channels_config({"channels": {"telegram": {"instance_id": "x"}}})
+
+    def test_wechat_and_telegram_coexist(self, tmp_path: Path) -> None:
+        p = _write(
+            tmp_path,
+            """
+[[channels.wechat]]
+instance_id = "wx"
+base_url = "http://host"
+token_env = "WT"
+enabled = true
+
+[[channels.telegram]]
+instance_id = "tg"
+token_env = "TT"
+enabled = true
+""",
+        )
+        cfg = load_channels_config(p)
+        assert len(cfg.wechat) == 1
+        assert len(cfg.telegram) == 1
+        assert len(cfg.enabled_wechat) == 1
+        assert len(cfg.enabled_telegram) == 1