diff --git a/channels/auth.py b/channels/auth.py
index da34d977..14bdba3f 100644
--- a/channels/auth.py
+++ b/channels/auth.py
@@ -1,156 +1,365 @@
-import hmac
-import json
-import os
-import time
-import urllib.request
-from pathlib import Path
-from config import config_get_by_key
-
-from src.logger import get_logger
-
-logger = get_logger(__name__)
-
-_proxy_url = None
-_auth_enabled = None
-_CHANNEL_DIR_NAME = ".channel"
-_CHANNEL_AUTH_USER_FILE = "authenticated-user.json"
-_REPO_ROOT = Path(__file__).resolve().parents[1]
-_MEMORY_DIRECTORY = str(_REPO_ROOT / "memory")
-_user_ID_processed = False
-
-
-def get_proxy_url():
-    global _proxy_url
-    if _proxy_url is None:
-        _proxy_url = config_get_by_key("GATEWAY_URL", "").rstrip("/")
-    return _proxy_url
-
-
-def _local_auth_secret():
-    return os.environ.get("OMEGA_AUTH_SECRET", "").strip()
-
-
-def is_auth_enabled():
-    global _auth_enabled
-    if _auth_enabled is not None:
-        return _auth_enabled
-    proxy = get_proxy_url()
-    if not proxy:
-        _auth_enabled = bool(_local_auth_secret())
-        return _auth_enabled
-    try:
-        url = f"{proxy}/auth/status"
-        with urllib.request.urlopen(url, timeout=5) as resp:
-            data = json.loads(resp.read())
-            _auth_enabled = data.get("enabled", False)
-    except Exception as e:
-        logger.warning(f"Could not read auth status from proxy, assuming auth is disabled: {e}")
-        _auth_enabled = False
-    return _auth_enabled
-
-
-def verify_token(candidate):
-    proxy = get_proxy_url()
-    if not proxy:
-        secret = _local_auth_secret()
-        return bool(secret) and hmac.compare_digest(
-            str(candidate).encode("utf-8"), secret.encode("utf-8")
-        )
-    url = f"{proxy}/auth/verify"
-    req = urllib.request.Request(url)
-    req.add_header("X-Auth-Token", str(candidate))
-    try:
-        with urllib.request.urlopen(req, timeout=5) as resp:
-            data = json.loads(resp.read())
-            return data.get("match", False)
-    except Exception as e:
-        logger.error(f"Token verification request failed, denying: {e}")
-        return False
-
-
-def _channel_auth_user_path():
-    return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_USER_FILE)
-
-
-def store_channel_authenticated_user_id(channel_identifier, user_id):
-    # For any single run of Omega, allow only a single save of a user-id or verification    
-    global _user_ID_processed
-    if _user_ID_processed:
-        logger.warning(f"[{channel_identifier}] Warning: a user already was validated, ignoring")
-        return False    
-    channel_identifier = str(channel_identifier or "").strip()
-    if not channel_identifier:
-        raise ValueError("channel_identifier is required")
-    user_id = str(user_id or "").strip()
-    if not user_id:
-        raise ValueError("user_id is required")
-    
-    """Record an authenticated channel user ID in the memory directory."""
-    payload = {
-        "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-        "channel_identifier": channel_identifier,
-        "user_id": user_id,
-    }
-    path = _channel_auth_user_path()
-    os.makedirs(os.path.dirname(path), exist_ok=True)
-    try:
-        with open(path, "a", encoding="utf-8") as f:
-            json.dump(payload, f, separators=(",", ":"))
-            f.write("\n")
-    except OSError as e:
-        raise RuntimeError("Failed to write channel authenticated user record") from e
-    _user_ID_processed = True
-    return True
-
-
-def get_channel_authenticated_user_id(channel_identifier):
-    """Return the first owner persisted for a channel, if one exists."""
-    channel_identifier = str(channel_identifier or "").strip()
-    if not channel_identifier:
-        raise ValueError("channel_identifier is required")
-    try:
-        path = _channel_auth_user_path()
-        with open(path, "r", encoding="utf-8") as f:
-            for line in f:
-                try:
-                    record = json.loads(line)
-                    saved_channel_identifier = str(record.get("channel_identifier", "")).strip()
-                    saved_user_id = str(record.get("user_id", "")).strip()
-                except (AttributeError, json.JSONDecodeError) as e:
-                    logger.warning(f"Skipping malformed channel authenticated user record: {e}")
-                    continue
-                if saved_channel_identifier == channel_identifier and saved_user_id:
-                    return saved_user_id
-    except FileNotFoundError:
-        return None
-    except Exception as e:
-        raise RuntimeError("Failed to read channel authenticated user records") from e
-    return None
-
-
-def get_channel_saved_user_id(channel_identifier, user_id):
-    global _user_ID_processed
-    saved_user_id = get_channel_authenticated_user_id(channel_identifier)
-    if saved_user_id != str(user_id or "").strip():
-        return False
-    _user_ID_processed = True
-    return True
-
-
-def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None):
-    # A persisted owner takes precedence over the reusable startup secret,
-    # including after a process restart.
-    saved_user_id = get_channel_authenticated_user_id(channel_identifier)
-    if saved_user_id is not None:
-        return "allow" if saved_user_id == str(user_id or "").strip() else "ignore"
-
-    # A token is accepted only when it came from an explicit auth command.
-    if auth_candidate is not None and verify_token(auth_candidate):
-        if store_channel_authenticated_user_id(channel_identifier, user_id):
-            label = str(channel_identifier).upper()
-            logger.info(f"[{label}] Saved authenticated user ID")
-            return "auth_bound"
-        else:
-            logger.error(f"[{label}] ERROR -- Unable to save user ID")
-            return "ignore"
-    return "ignore"
+import hmac
+import json
+import os
+import time
+import urllib.request
+from pathlib import Path
+from config import config_get_by_key
+
+from src.logger import get_logger
+
+logger = get_logger(__name__)
+
+_proxy_url = None
+_auth_enabled = None
+_CHANNEL_DIR_NAME = ".channel"
+_CHANNEL_AUTH_USER_FILE = "authenticated-user.json"
+_CHANNEL_AUTH_GROUP_FILE = "authenticated-group.json"
+_REPO_ROOT = Path(__file__).resolve().parents[1]
+_MEMORY_DIRECTORY = str(_REPO_ROOT / "memory")
+_user_ID_processed = False
+
+
+def get_proxy_url():
+    global _proxy_url
+    if _proxy_url is None:
+        configured_url = config_get_by_key("GATEWAY_URL", "")
+        _proxy_url = str(configured_url or "").strip().rstrip("/")
+    return _proxy_url
+
+
+def _local_auth_secret():
+    return os.environ.get("OMEGA_AUTH_SECRET", "").strip()
+
+
+def is_auth_enabled():
+    global _auth_enabled
+    if _auth_enabled is not None:
+        return _auth_enabled
+    proxy = get_proxy_url()
+    if not proxy:
+        _auth_enabled = bool(_local_auth_secret())
+        return _auth_enabled
+    try:
+        url = f"{proxy}/auth/status"
+        with urllib.request.urlopen(url, timeout=5) as resp:
+            data = json.loads(resp.read())
+            _auth_enabled = data.get("enabled", False)
+    except Exception as e:
+        logger.warning(f"Could not read auth status from proxy, assuming auth is disabled: {e}")
+        _auth_enabled = False
+    return _auth_enabled
+
+
+def verify_token(candidate):
+    proxy = get_proxy_url()
+    if not proxy:
+        secret = _local_auth_secret()
+        return bool(secret) and hmac.compare_digest(
+            str(candidate).encode("utf-8"), secret.encode("utf-8")
+        )
+    url = f"{proxy}/auth/verify"
+    req = urllib.request.Request(url)
+    req.add_header("X-Auth-Token", str(candidate))
+    try:
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            data = json.loads(resp.read())
+            return data.get("match", False)
+    except Exception as e:
+        logger.error(f"Token verification request failed, denying: {e}")
+        return False
+
+
+def _channel_auth_user_path():
+    return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_USER_FILE)
+
+
+# ---------------------------------------------------------------------------
+# Single-user (owner) authentication.
+
+def store_channel_authenticated_user_id(channel_identifier, user_id):
+    # For any single run of Omega, allow only a single save of a user-id or verification    
+    global _user_ID_processed
+    if _user_ID_processed:
+        logger.warning(f"[{channel_identifier}] Warning: a user already was validated, ignoring")
+        return False
+    channel_identifier = str(channel_identifier or "").strip()
+    if not channel_identifier:
+        raise ValueError("channel_identifier is required")
+    user_id = str(user_id or "").strip()
+    if not user_id:
+        raise ValueError("user_id is required")
+
+    """Record an authenticated channel user ID in the memory directory."""
+    payload = {
+        "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "channel_identifier": channel_identifier,
+        "user_id": user_id,
+    }
+    path = _channel_auth_user_path()
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    try:
+        with open(path, "a", encoding="utf-8") as f:
+            json.dump(payload, f, separators=(",", ":"))
+            f.write("\n")
+    except OSError as e:
+        raise RuntimeError("Failed to write channel authenticated user record") from e
+    _user_ID_processed = True
+    return True
+
+
+def get_channel_saved_user_id(channel_identifier, user_id):
+    # For any single run of OmegaClaw, allow only a single save of a user-id or verification
+    global _user_ID_processed
+    if _user_ID_processed:
+        logger.warning(f"[{channel_identifier}] Warning: a user was already validated, ignoring")
+        return False
+
+    # The first persisted record is the owner.  Do not scan later records
+    saved_user_id = get_channel_authenticated_user_id(channel_identifier)
+    if saved_user_id != str(user_id or "").strip():
+        return False
+
+    _user_ID_processed = True
+    return True
+
+
+def authenticate_channel_user(channel_identifier, user_id, auth_candidate=None):
+    channel_identifier = str(channel_identifier or "").strip()
+    user_id = str(user_id or "").strip()
+
+    # A persisted owner always wins over a reusable secret.
+    saved_user_id = get_channel_authenticated_user_id(channel_identifier)
+    if saved_user_id is not None:
+        return "allow" if saved_user_id == user_id else "ignore"
+
+    # The secret can establish an owner only before an owner has been saved.
+    if auth_candidate is not None and verify_token(auth_candidate):
+        label = channel_identifier.upper()
+        if store_channel_authenticated_user_id(channel_identifier, user_id):
+            logger.info(f"[{label}] Saved authenticated user ID")
+            return "auth_bound"
+        logger.error(f"[{label}] ERROR -- Unable to save user ID")
+
+    return "ignore"
+
+
+def get_channel_authenticated_user_id(channel_identifier):
+    """
+    Read-only owner lookup. Returns the persisted owner user_id for a
+    channel, or None if no owner has authenticated yet.
+    """
+    channel_identifier = str(channel_identifier or "").strip()
+    if not channel_identifier:
+        return None
+    try:
+        path = _channel_auth_user_path()
+        with open(path, "r", encoding="utf-8") as f:
+            for line in f:
+                try:
+                    record = json.loads(line)
+                    saved_channel_identifier = str(record.get("channel_identifier", "")).strip()
+                    saved_user_id = str(record.get("user_id", "")).strip()
+                except (AttributeError, json.JSONDecodeError) as e:
+                    logger.warning(f"Skipping malformed channel authenticated user record: {e}")
+                    continue
+                if saved_channel_identifier == channel_identifier and saved_user_id:
+                    return saved_user_id
+    except FileNotFoundError:
+        return None
+    except Exception as e:
+        raise RuntimeError("Failed to read channel authenticated user records") from e
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Owner-managed group authorization.
+
+def _channel_auth_group_path():
+    return os.path.join(_MEMORY_DIRECTORY, _CHANNEL_DIR_NAME, _CHANNEL_AUTH_GROUP_FILE)
+
+
+def load_channel_auth_state(channel_identifier):
+    """Validate and load one channel's persisted owner and active groups."""
+    channel_identifier = str(channel_identifier or "").strip()
+    if not channel_identifier:
+        raise ValueError("channel_identifier is required")
+
+    def read_records(path, label):
+        try:
+            with open(path, "r", encoding="utf-8") as source:
+                records = []
+                for line_number, line in enumerate(source, 1):
+                    if not line.strip():
+                        continue
+                    try:
+                        record = json.loads(line)
+                    except json.JSONDecodeError as exc:
+                        logger.warning(
+                            f"Skipping malformed {label} record at line "
+                            f"{line_number}: {exc}"
+                        )
+                        continue
+                    if not isinstance(record, dict):
+                        logger.warning(
+                            f"Skipping malformed {label} record at line "
+                            f"{line_number}: expected a JSON object"
+                        )
+                        continue
+                    records.append(record)
+                return records
+        except FileNotFoundError:
+            return []
+        except (OSError, UnicodeError) as exc:
+            raise RuntimeError(f"Failed to read {label} records") from exc
+
+    owner_id = None
+    for record in read_records(_channel_auth_user_path(), "channel authenticated user"):
+        saved_channel = str(record.get("channel_identifier", "")).strip()
+        saved_user = str(record.get("user_id", "")).strip()
+        if saved_channel == channel_identifier and saved_user and owner_id is None:
+            owner_id = saved_user
+
+    group_states = {}
+    for record in read_records(_channel_auth_group_path(), "channel authenticated group"):
+        saved_channel = str(record.get("channel_identifier", "")).strip()
+        saved_group = str(record.get("group_id", "")).strip()
+        authorized_by = str(record.get("authorized_by", "")).strip()
+        if (
+            saved_channel == channel_identifier
+            and saved_group
+            and owner_id is not None
+            and authorized_by == owner_id
+        ):
+            group_states[saved_group] = not bool(record.get("revoked", False))
+        elif saved_channel == channel_identifier and saved_group:
+            logger.warning(
+                f"[{channel_identifier.upper()}] Skipping group authorization "
+                f"for {saved_group}: record is not owned by the authenticated user"
+            )
+
+    authorized_groups = {
+        group_id for group_id, authorized in group_states.items() if authorized
+    }
+    return owner_id, authorized_groups
+
+
+def store_channel_authenticated_group_id(channel_identifier, group_id, authorized_by_user_id):
+    """Persist a trusted group. Never touches the single-user auth file."""
+    channel_identifier = str(channel_identifier or "").strip()
+    group_id = str(group_id or "").strip()
+    authorized_by_user_id = str(authorized_by_user_id or "").strip()
+    if not channel_identifier:
+        raise ValueError("channel_identifier is required")
+    if not group_id:
+        raise ValueError("group_id is required")
+    if not authorized_by_user_id:
+        raise ValueError("authorized_by_user_id is required")
+
+    payload = {
+        "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "channel_identifier": channel_identifier,
+        "group_id": group_id,
+        "authorized_by": authorized_by_user_id,
+    }
+    path = _channel_auth_group_path()
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    try:
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(payload, separators=(",", ":")) + "\n")
+    except OSError as e:
+        raise RuntimeError("Failed to write channel authenticated group record") from e
+    return True
+
+
+def get_channel_saved_group_id(channel_identifier, group_id):
+    """Return whether a group has already been authorized for this channel."""
+    channel_identifier = str(channel_identifier or "").strip()
+    group_id = str(group_id or "").strip()
+    if not channel_identifier or not group_id:
+        return False
+    owner_id = get_channel_authenticated_user_id(channel_identifier)
+    if owner_id is None:
+        return False
+
+    authorized = False
+    try:
+        with open(_channel_auth_group_path(), "r", encoding="utf-8") as f:
+            for line in f:
+                try:
+                    record = json.loads(line)
+                    saved_channel = str(record.get("channel_identifier", "")).strip()
+                    saved_group = str(record.get("group_id", "")).strip()
+                    authorized_by = str(record.get("authorized_by", "")).strip()
+                except (AttributeError, json.JSONDecodeError) as e:
+                    logger.warning(f"Skipping malformed channel authenticated group record: {e}")
+                    continue
+                if (
+                    saved_channel == channel_identifier
+                    and saved_group == group_id
+                    and authorized_by == owner_id
+                ):
+                    authorized = not bool(record.get("revoked", False))
+    except FileNotFoundError:
+        return False
+    except Exception as e:
+        raise RuntimeError("Failed to read channel authenticated group records") from e
+    return authorized
+
+
+def authorize_channel_group(channel_identifier, group_id, requester_user_id):
+    """
+    Open a group chat to all its members -- but ONLY when requester_user_id
+    matches the persisted owner for this channel
+    """
+    if get_channel_saved_group_id(channel_identifier, group_id):
+        return "allow"
+
+    owner_id = get_channel_authenticated_user_id(channel_identifier)
+    if owner_id is None:
+        # No owner has authenticated yet -- nobody can open groups.
+        return "ignore"
+
+    if str(requester_user_id or "").strip() != owner_id:
+        return "ignore"
+
+    if store_channel_authenticated_group_id(channel_identifier, group_id, owner_id):
+        logger.info(f"[{str(channel_identifier).upper()}] Saved authorized group ID")
+        return "group_bound"
+
+    logger.error(f"[{str(channel_identifier).upper()}] ERROR -- Unable to save group ID")
+    return "ignore"
+
+
+def revoke_channel_group(channel_identifier, group_id, requester_user_id):
+    """Remove a trusted group when requested by the persisted channel owner."""
+    channel_identifier = str(channel_identifier or "").strip()
+    group_id = str(group_id or "").strip()
+    requester_user_id = str(requester_user_id or "").strip()
+    if not channel_identifier or not group_id:
+        return "ignore"
+
+    owner_id = get_channel_authenticated_user_id(channel_identifier)
+    if owner_id is None or requester_user_id != owner_id:
+        return "ignore"
+
+    if not get_channel_saved_group_id(channel_identifier, group_id):
+        return "ignore"
+
+    payload = {
+        "time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "channel_identifier": channel_identifier,
+        "group_id": group_id,
+        "authorized_by": owner_id,
+        "revoked": True,
+    }
+    path = _channel_auth_group_path()
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    try:
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(payload, separators=(",", ":")) + "\n")
+    except OSError as e:
+        raise RuntimeError("Failed to write channel group revocation record") from e
+
+    logger.info(f"[{channel_identifier.upper()}] Removed authorized group ID {group_id}")
+    return "group_unbound"
diff --git a/channels/telegram.py b/channels/telegram.py
index b7fa26a5..73876681 100644
--- a/channels/telegram.py
+++ b/channels/telegram.py
@@ -1,298 +1,553 @@
-import json
-import os
-import threading
-import time
-import urllib.parse
-import urllib.request
-import auth
-from src.logger import get_logger
-from delivery_queue import PendingMessages
-import channels
-from config import config_get_by_key
-
-logger = get_logger(__name__)
-
-_running = False
-_last_message = ""
-_msg_lock = threading.Lock()
-_state_lock = threading.Lock()
-
-_bot_token = ""
-_api_base = ""
-_chat_id = ""
-_poll_timeout = 20
-_offset = None
-_connected = False
-
-_authenticated_user_id = None
-_outbox = PendingMessages()
-
-
-def _set_last(msg):
-    global _last_message
-    with _msg_lock:
-        if _last_message == "":
-            _last_message = msg
-        else:
-            _last_message = _last_message + " | " + msg
-
-
-def getLastMessage():
-    global _last_message
-    with _msg_lock:
-        tmp = _last_message
-        _last_message = ""
-        return tmp
-
-
-def _parse_auth_candidate(msg):
-    text = msg.strip()
-    lower = text.lower()
-    if lower.startswith("auth "):
-        return text[5:].strip()
-    if lower.startswith("/auth "):
-        return text[6:].strip()
-    return text
-
-
-def _display_name(user, chat):
-    username = str(user.get("username", "")).strip()
-    if username:
-        return f"@{username}"
-
-    first = str(user.get("first_name", "")).strip()
-    last = str(user.get("last_name", "")).strip()
-    full = f"{first} {last}".strip()
-    if full:
-        return full
-
-    title = str(chat.get("title", "")).strip()
-    if title:
-        return title
-
-    return "telegram_user"
-
-
-def _api_call(method, params=None, timeout=30, use_post=False):
-    if not _api_base:
-        raise RuntimeError("Telegram adapter not initialized")
-
-    params = params or {}
-    encoded = urllib.parse.urlencode(params).encode("utf-8")
-    url = f"{_api_base}/{method}"
-
-    if use_post:
-        req = urllib.request.Request(url, data=encoded)
-    else:
-        if params:
-            url = f"{url}?{urllib.parse.urlencode(params)}"
-        req = urllib.request.Request(url)
-
-    with urllib.request.urlopen(req, timeout=timeout) as response:
-        payload = json.loads(response.read().decode("utf-8", errors="ignore"))
-
-    if not payload.get("ok"):
-        raise RuntimeError(payload.get("description", f"{method} failed"))
-
-    return payload.get("result")
-
-
-def _initialize_offset():
-    global _offset
-    try:
-        updates = _api_call("getUpdates", {"timeout": 0}, timeout=10) or []
-    except Exception as exc:
-        logger.warning(f"Could not read initial offset: {exc}")
-        return
-
-    max_update = -1
-    for update in updates:
-        update_id = update.get("update_id")
-        if isinstance(update_id, int):
-            max_update = max(max_update, update_id)
-
-    if max_update >= 0:
-        with _state_lock:
-            _offset = max_update + 1
-
-
-def _is_auth_command(msg):
-    lower = msg.strip().lower()
-    return lower.startswith("auth ") or lower.startswith("/auth ")
-
-
-def _is_allowed_message(chat_id, user_id, msg):
-    global _chat_id, _authenticated_user_id
-
-    with _state_lock:
-        if _chat_id and chat_id != _chat_id:
-            return "ignore"
-        if not auth.is_auth_enabled():
-            if not _chat_id:
-                _chat_id = chat_id
-            return "allow"
-        if _authenticated_user_id is not None:
-            if chat_id != _chat_id:
-                return "ignore"
-            return "allow" if user_id == _authenticated_user_id else "ignore"
-        auth_candidate = _parse_auth_candidate(msg) if _is_auth_command(msg) else None
-        user_id_check = auth.authenticate_channel_user('TELEGRAM', user_id, auth_candidate)
-        if user_id_check in ["auth_bound", "allow"]:
-            _authenticated_user_id = user_id
-            _chat_id = chat_id
-            return user_id_check
-        else:
-            return "ignore"
-
-
-def _ready_to_send():
-    with _state_lock:
-        return _connected and bool(_chat_id)
-
-
-def _deliver_outbound(chunk):
-    with _state_lock:
-        target_chat = _chat_id
-    if not target_chat:
-        raise RuntimeError("Telegram chat is not bound")
-    _api_call(
-        "sendMessage",
-        {"chat_id": target_chat, "text": chunk},
-        timeout=15,
-        use_post=True,
-    )
-
-
-def _flush_outbox():
-    global _connected
-    try:
-        _outbox.flush(_deliver_outbound, _ready_to_send)
-    except Exception as exc:
-        _connected = False
-        logger.warning(f"Telegram send failed; retaining queued message: {exc}")
-
-
-def _poll_loop():
-    global _connected, _offset
-    logger.info("Polling started")
-
-    while _running:
-        try:
-            params = {"timeout": int(_poll_timeout)}
-            with _state_lock:
-                if _offset is not None:
-                    params["offset"] = _offset
-
-            updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or []
-            _connected = True
-
-            for update in updates:
-                update_id = update.get("update_id")
-                if isinstance(update_id, int):
-                    with _state_lock:
-                        if _offset is None or (update_id + 1) > _offset:
-                            _offset = update_id + 1
-
-                message = update.get("message") or update.get("edited_message")
-                if not isinstance(message, dict):
-                    continue
-
-                text = message.get("text")
-                if not text:
-                    continue
-
-                chat = message.get("chat") or {}
-                user = message.get("from") or {}
-                chat_id = str(chat.get("id", "")).strip()
-                user_id = str(user.get("id", "")).strip()
-                if not chat_id or not user_id:
-                    continue
-
-                state = _is_allowed_message(chat_id, user_id, text)
-                display_name = _display_name(user, chat)
-                if state == "allow":
-                    _set_last(f"{display_name}: {text}")
-                elif state == "auth_bound":
-                    send_message(f"Authentication successful for {display_name}.")
-            _flush_outbox()
-        except Exception as exc:
-            _connected = False
-            logger.warning(f"Poll error: {exc}")
-            time.sleep(2)
-
-    _connected = False
-    logger.info("Polling stopped")
-
-
-def start_telegram(chat_id="", poll_timeout=20):
-    global _running, _bot_token, _api_base, _chat_id, _poll_timeout, _offset, _connected
-
-    proxy = auth.get_proxy_url()
-    if proxy:
-        _bot_token = "proxy"
-        _api_base = f"{proxy}/telegram"
-    else:
-        _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip()
-        if not _bot_token:
-            raise ValueError("TG_BOT_TOKEN is required")
-        _api_base = f"https://api.telegram.org/bot{_bot_token}"
-
-    _chat_id = str(chat_id).strip()
-
-    try:
-        _poll_timeout = max(1, int(poll_timeout))
-    except Exception as e:
-        logger.warning(f"Invalid poll_timeout {poll_timeout!r}, falling back to 20: {e}")
-        _poll_timeout = 20
-
-    _offset = None
-    _running = True
-    _connected = False
-    logger.info(f"Starting adapter with chat target: {_chat_id or 'auto-bind'}")
-    _initialize_offset()
-
-    t = threading.Thread(target=_poll_loop, daemon=True)
-    t.start()
-    return t
-
-
-def stop_telegram():
-    global _running
-    _running = False
-
-
-def send_message(text):
-    text = str(text).replace("\\n", "\n").replace("\r", "")
-    if not text:
-        return
-
-    max_len = 3900
-    chunks = []
-    for i in range(0, len(text), max_len):
-        chunk = text[i:i + max_len]
-        if chunk:
-            chunks.append(chunk)
-    _outbox.extend(chunks)
-    _flush_outbox()
-
-class TelegramChannel(channels.CommChannel):
-
-    def __init__(self):
-        super().__init__()
-
-    def start(self) -> None:
-        chat_id = config_get_by_key("TG_CHAT_ID", "")
-        poll_timeout = int(config_get_by_key("TG_POLL_TIMEOUT", 20))
-        start_telegram(chat_id, poll_timeout)
-
-    def stop(self) -> None:
-        stop_telegram()
-
-    def receive(self) -> str:
-        return getLastMessage()
-
-    def send(self, message: str) -> None:
-        send_message(message)
-
+import json
+import os
+import re
+import threading
+import time
+import urllib.parse
+import urllib.request
+from collections import deque
+import auth
+from src.logger import get_logger
+from delivery_queue import PendingMessages
+import channels
+from config import config_get_by_key
+
+logger = get_logger(__name__)
+
+_running = False
+_msg_lock = threading.Lock()
+_state_lock = threading.Lock()
+_inbox = deque()
+_default_chat_id = ""
+_outbox = PendingMessages()
+_deferred_default_outbox = PendingMessages()
+
+_bot_token = ""
+_api_base = ""
+_bot_username= ""
+_poll_timeout = 20
+_offset = None
+_connected = False
+_admin_allowed_chats = set()
+_owner_id = None
+_authorized_groups = set()
+
+_auto_bound_chat = ""
+
+_BIND_COMMANDS = ("/bind", "/authorize_group")
+_UNBIND_COMMANDS = ("/unbind", "/unauthorize_group")
+_ROUTED_MESSAGE_RE = re.compile(
+    r"^\s*\[(-?\d*)\]\s*\[(\d*)\]\s*(.*)$",
+    re.DOTALL,
+)
+_TARGET_ONLY_MESSAGE_RE = re.compile(r"^\s*\[(-?\d+)\]\s*(.*)$", re.DOTALL)
+
+
+def _enqueue_message(msg, chat_id, reply_to_id=None):
+    with _msg_lock:
+        _inbox.append(
+            (
+                str(chat_id),
+                str(reply_to_id) if reply_to_id is not None else "",
+                str(msg),
+            )
+        )
+
+
+def getLastMessage():
+    with _msg_lock:
+        if not _inbox:
+            return ""
+        chat_id, reply_to_id, message = _inbox.popleft()
+    return f"[{chat_id}] [{reply_to_id}] {message}"
+
+
+def _ready_to_send():
+    return _connected
+
+
+def _deliver_outbound(item):
+    target_chat, reply_to_id, chunk = item
+    params = {"chat_id": target_chat, "text": chunk}
+    if reply_to_id:
+        params["reply_parameters"] = json.dumps(
+            {
+                "message_id": int(reply_to_id),
+                "allow_sending_without_reply": True,
+            },
+            separators=(",", ":"),
+        )
+    _api_call(
+        "sendMessage",
+        params,
+        timeout=15,
+        use_post=True,
+    )
+
+
+def _flush_outbox():
+    try:
+        _outbox.flush(_deliver_outbound, _ready_to_send)
+    except Exception as exc:
+        logger.warning(f"Telegram send failed; retaining queued message: {exc}")
+
+
+def _ready_to_route_deferred_default():
+    with _state_lock:
+        return bool(_default_chat_id)
+
+
+def _route_deferred_default(item):
+    reply_to_id, chunk = item
+    with _state_lock:
+        target_chat = str(_default_chat_id or "").strip()
+    if not target_chat:
+        raise RuntimeError("Telegram owner DM is not available")
+    _outbox.put((target_chat, reply_to_id, chunk))
+
+
+def _flush_deferred_default_outbox():
+    try:
+        _deferred_default_outbox.flush(
+            _route_deferred_default,
+            _ready_to_route_deferred_default,
+        )
+    except Exception as exc:
+        logger.warning(
+            f"Telegram deferred send failed; retaining queued message: {exc}"
+        )
+        return
+    _flush_outbox()
+
+
+def _parse_outbound_message(text):
+    """Return target, reply message ID, body, and whether routing was explicit."""
+    match = _ROUTED_MESSAGE_RE.match(text)
+    if match:
+        return match.group(1), match.group(2), match.group(3), True
+
+    # Backward compatibility with the upstream Telegram form: [chat_id] body.
+    match = _TARGET_ONLY_MESSAGE_RE.match(text)
+    if match:
+        return match.group(1), "", match.group(2), True
+
+    return "", "", text, False
+
+
+def _is_allowed_outbound_target(chat_id):
+    chat_id = str(chat_id or "").strip()
+    if not chat_id:
+        return False
+
+    if auth.is_auth_enabled():
+        return chat_id == str(_owner_id or "") or chat_id in _authorized_groups
+
+    if _admin_allowed_chats:
+        return chat_id in _admin_allowed_chats
+    return chat_id in {str(_default_chat_id or ""), str(_auto_bound_chat or "")}
+
+
+def send_message(text, target_chat=None, reply_to_id=None):
+    text = str(text).replace("\\n", "\n").replace("\r", "")
+    if not text:
+        return False
+
+    trusted_target = str(target_chat or "").strip()
+    if trusted_target:
+        target_chat = trusted_target
+        reply_to_id = str(reply_to_id or "").strip()
+    else:
+        parsed_target, parsed_reply, text, routed = _parse_outbound_message(text)
+        target_chat = parsed_target or str(_default_chat_id or "").strip()
+        reply_to_id = parsed_reply
+        if routed and parsed_target and not _is_allowed_outbound_target(parsed_target):
+            logger.warning(
+                f"Telegram send rejected: target chat {parsed_target} is not authorized"
+            )
+            return False
+
+    if not text:
+        logger.warning("Telegram send skipped: routed message body is empty")
+        return False
+
+    max_len = 3900
+    chunks = [text[i:i + max_len] for i in range(0, len(text), max_len)]
+    if not target_chat:
+        _deferred_default_outbox.extend((reply_to_id, chunk) for chunk in chunks)
+        logger.info("Telegram send deferred until an owner DM is available")
+        return True
+
+    outbound = [(target_chat, reply_to_id, chunk) for chunk in chunks]
+    _outbox.extend(outbound)
+    _flush_outbox()
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Authorization.
+
+def _parse_auth_candidate(msg):
+    text = msg.strip()
+    lower = text.lower()
+    if lower.startswith("auth "):
+        return text[5:].strip()
+    if lower.startswith("/auth "):
+        return text[6:].strip()
+    return text
+
+
+def _is_auth_command(msg):
+    lower = msg.strip().lower()
+    return lower.startswith("auth ") or lower.startswith("/auth ")
+
+
+def _first_token(msg):
+    stripped = msg.strip()
+    if not stripped:
+        return ""
+    return stripped.split(None, 1)[0].lower()
+
+
+def _is_targeted_command(msg, commands):
+    token = _first_token(msg)
+    command, separator, target_username = token.partition("@")
+
+    if command not in commands:
+        return False
+
+    if not separator:
+        return True
+
+    return bool(_bot_username) and target_username == _bot_username
+
+
+def _is_bind_command(msg):
+    return _is_targeted_command(msg, _BIND_COMMANDS)
+
+
+def _is_unbind_command(msg):
+    return _is_targeted_command(msg, _UNBIND_COMMANDS)
+
+
+def _command_argument(msg):
+    parts = str(msg or "").strip().split(None, 1)
+    return parts[1].strip() if len(parts) == 2 else ""
+
+
+def _is_allowed_message(chat_id, user_id, chat_type, msg):
+    global _auto_bound_chat, _owner_id, _default_chat_id
+
+    auth_enabled = auth.is_auth_enabled()
+
+    if not auth_enabled:
+        if _admin_allowed_chats:
+            return "allow" if chat_id in _admin_allowed_chats else "ignore"
+        with _state_lock:
+            if _auto_bound_chat and chat_id != _auto_bound_chat:
+                return "ignore"
+            if not _auto_bound_chat:
+                _auto_bound_chat = chat_id
+                _default_chat_id = chat_id
+        return "allow"
+
+    owner_id = _owner_id
+
+    is_owner_bootstrap = (
+        owner_id is None
+        and chat_type == "private"
+        and _is_auth_command(msg)
+    )
+    is_owner_private_chat = (
+        owner_id is not None
+        and chat_type == "private"
+        and user_id == owner_id
+    )
+    is_owner_group_bind = (
+        owner_id is not None
+        and chat_type != "private"
+        and user_id == owner_id
+        and _is_bind_command(msg)
+    )
+
+    if (
+        _admin_allowed_chats
+        and chat_id not in _admin_allowed_chats
+        and not is_owner_bootstrap
+        and not is_owner_private_chat
+        and not is_owner_group_bind
+    ):
+        return "ignore"
+
+    if owner_id is None:
+        if is_owner_bootstrap:
+            candidate = _parse_auth_candidate(msg)
+            state = auth.authenticate_channel_user("TELEGRAM", user_id, candidate)
+            if state == "auth_bound":
+                with _state_lock:
+                    _owner_id = user_id
+                    _default_chat_id = chat_id
+            return state
+        return "ignore"
+
+    if chat_type == "private":
+        if user_id == owner_id and _is_unbind_command(msg):
+            group_id = _command_argument(msg)
+            if group_id:
+                state = auth.revoke_channel_group("TELEGRAM", group_id, user_id)
+                if state == "group_unbound":
+                    _authorized_groups.discard(group_id)
+                    _admin_allowed_chats.discard(group_id)
+                return state
+            return "ignore"
+        return "allow" if user_id == owner_id else "ignore"
+
+    # Anything that isn't "private" is a group/supergroup chat.
+    if _is_unbind_command(msg):
+        if user_id == owner_id:
+            state = auth.revoke_channel_group("TELEGRAM", chat_id, user_id)
+            if state == "group_unbound":
+                _authorized_groups.discard(chat_id)
+                _admin_allowed_chats.discard(chat_id)
+            return state
+        return "ignore"
+
+    if chat_id in _authorized_groups:
+        return "allow"
+
+    if user_id == owner_id and _is_bind_command(msg):
+        state = auth.authorize_channel_group("TELEGRAM", chat_id, user_id)
+        if state in {"allow", "group_bound"}:
+            _authorized_groups.add(chat_id)
+            _admin_allowed_chats.add(chat_id)
+        return state
+
+    return "ignore"
+
+
+def _display_name(user, chat):
+    username = str(user.get("username", "")).strip()
+    if username:
+        return f"@{username}"
+
+    first = str(user.get("first_name", "")).strip()
+    last = str(user.get("last_name", "")).strip()
+    full = f"{first} {last}".strip()
+    if full:
+        return full
+
+    title = str(chat.get("title", "")).strip()
+    if title:
+        return title
+
+    return "telegram_user"
+
+
+def _api_call(method, params=None, timeout=30, use_post=False):
+    if not _api_base:
+        raise RuntimeError("Telegram adapter not initialized")
+
+    params = params or {}
+    encoded = urllib.parse.urlencode(params).encode("utf-8")
+    url = f"{_api_base}/{method}"
+
+    if use_post:
+        req = urllib.request.Request(url, data=encoded)
+    else:
+        if params:
+            url = f"{url}?{urllib.parse.urlencode(params)}"
+        req = urllib.request.Request(url)
+
+    with urllib.request.urlopen(req, timeout=timeout) as response:
+        payload = json.loads(response.read().decode("utf-8", errors="ignore"))
+
+    if not payload.get("ok"):
+        raise RuntimeError(payload.get("description", f"{method} failed"))
+
+    return payload.get("result")
+
+def _initialize_bot_identity():
+    global _bot_username
+
+    try:
+        result = _api_call("getMe", timeout=10)
+        if not isinstance(result, dict):
+            raise RuntimeError("Telegram getMe returned an invalid response")
+        username = str(result.get("username", "")).strip().lstrip("@").lower()
+        if not username:
+            raise RuntimeError("Telegram getMe did not return the bot username")
+    except Exception as exc:
+        _bot_username = ""
+        logger.warning(f"Could not read Telegram bot identity: {exc}")
+        return False
+
+    _bot_username = username
+    return True
+
+
+def _process_update(update):
+    message = update.get("message") or update.get("edited_message")
+    if not isinstance(message, dict):
+        return
+
+    text = message.get("text")
+    if not text:
+        return
+
+    chat = message.get("chat") or {}
+    user = message.get("from") or {}
+    chat_id = str(chat.get("id", "")).strip()
+    user_id = str(user.get("id", "")).strip()
+    chat_type = str(chat.get("type", "")).strip()
+    message_id = message.get("message_id")
+    if not chat_id or not user_id:
+        return
+
+    state = _is_allowed_message(chat_id, user_id, chat_type, text)
+    display_name = _display_name(user, chat)
+
+    if state == "allow":
+        _flush_deferred_default_outbox()
+        _enqueue_message(f"{display_name}: {text}", chat_id, message_id)
+    elif state == "auth_bound":
+        send_message(
+            f"Authentication successful. {display_name} is now the bot owner. "
+            "Send /bind in a group to open it to everyone there.",
+            chat_id,
+            message_id
+        )
+        _flush_deferred_default_outbox()
+    elif state == "group_bound":
+        send_message(
+            "This group is now authorized. All members can talk to the bot here.",
+            chat_id,
+            message_id
+        )
+    elif state == "group_unbound":
+        send_message("This group is no longer authorized.", chat_id, message_id)
+
+
+
+def _poll_loop():
+    global _connected, _offset
+    logger.info("Polling started")
+
+    while _running:
+        try:
+            params = {"timeout": int(_poll_timeout)}
+            with _state_lock:
+                if _offset is not None:
+                    params["offset"] = _offset
+
+            updates = _api_call("getUpdates", params=params, timeout=int(_poll_timeout) + 10) or []
+            _connected = True
+            _flush_outbox()
+
+            for update in updates:
+                update_id = update.get("update_id")
+                _process_update(update)
+                if isinstance(update_id, int):
+                    with _state_lock:
+                        if _offset is None or (update_id + 1) > _offset:
+                            _offset = update_id + 1
+        except Exception as exc:
+            _connected = False
+            logger.warning(f"Poll error: {exc}")
+            time.sleep(2)
+
+    _connected = False
+    logger.info("Polling stopped")
+
+
+def _parse_admin_allowed_chats(chat_id_config, allowed_config):
+    ids = set()
+    single = str(chat_id_config or "").strip()
+    if single:
+        ids.add(single)
+    for part in str(allowed_config or "").split(","):
+        part = part.strip()
+        if part:
+            ids.add(part)
+    return ids
+
+
+def start_telegram(chat_id="", allowed_chat_ids="", poll_timeout=20):
+    global _running, _bot_token, _api_base, _poll_timeout, _offset, _connected
+    global _default_chat_id
+    global _admin_allowed_chats, _auto_bound_chat, _owner_id, _authorized_groups
+
+    proxy = auth.get_proxy_url()
+    if proxy:
+        _bot_token = "proxy"
+        _api_base = f"{proxy}/telegram"
+    else:
+        _bot_token = os.environ.get("TG_BOT_TOKEN", "").strip()
+        if not _bot_token:
+            raise ValueError("TG_BOT_TOKEN is required")
+        _api_base = f"https://api.telegram.org/bot{_bot_token}"
+
+    _admin_allowed_chats = _parse_admin_allowed_chats(chat_id, allowed_chat_ids)
+    _auto_bound_chat = ""
+    if auth.is_auth_enabled():
+        _owner_id, _authorized_groups = auth.load_channel_auth_state("TELEGRAM")
+        _admin_allowed_chats.update(_authorized_groups)
+        # Telegram private chat IDs are the owner's user ID, so persisted
+        # ownership gives proactive messages a safe destination after restart.
+        _default_chat_id = str(_owner_id or chat_id or "").strip()
+    else:
+        _owner_id, _authorized_groups = None, set()
+        _default_chat_id = str(chat_id or "").strip()
+
+    with _msg_lock:
+        _inbox.clear()
+    _outbox.clear()
+    _deferred_default_outbox.clear()
+
+    try:
+        _poll_timeout = max(1, int(poll_timeout))
+    except Exception as e:
+        logger.warning(f"Invalid poll_timeout {poll_timeout!r}, falling back to 20: {e}")
+        _poll_timeout = 20
+
+    _offset = None
+    _running = True
+    _connected = False
+    if _admin_allowed_chats:
+        logger.info(f"Starting adapter, admin-restricted to chats: {sorted(_admin_allowed_chats)}")
+    else:
+        logger.info("Starting adapter with no admin chat restriction")
+
+    _initialize_bot_identity()
+
+    t = threading.Thread(target=_poll_loop, daemon=True)
+    t.start()
+    return t
+
+
+def stop_telegram():
+    global _running
+    _running = False
+
+
+class TelegramChannel(channels.CommChannel):
+
+    def __init__(self):
+        super().__init__()
+
+    def start(self) -> None:
+        chat_id = config_get_by_key("TG_CHAT_ID", "")
+        allowed_chat_ids = config_get_by_key("TG_ALLOWED_CHAT_IDS", "")
+        poll_timeout = int(config_get_by_key("TG_POLL_TIMEOUT", 20))
+        start_telegram(chat_id, allowed_chat_ids, poll_timeout)
+
+    def stop(self) -> None:
+        stop_telegram()
+
+    def receive(self) -> str:
+        return getLastMessage()
+
+    def send(self, message: str) -> None:
+        send_message(message)
+
+
 def loadOmegaPlugin():
     channels.registerCommChannel("telegram", TelegramChannel())
diff --git a/config/config.yaml b/config/config.yaml
index 04c2aad7..6a81ae94 100644
--- a/config/config.yaml
+++ b/config/config.yaml
@@ -79,8 +79,12 @@ SL_MAX_FILE_SIZE_MB: 5
 
 # Telegram
 # Enabled with commchannel: telegram
-# Telegram chat ID. If empty, Omega auto-binds after first valid inbound auth/message.
+# Legacy single-chat allowlist. When set, only this chat is accepted.
 TG_CHAT_ID: ""
+# Optional additional comma-separated chat IDs. These are a hard
+# administrator boundary. With authentication enabled, the owner can still
+# bootstrap and use their private DM.
+TG_ALLOWED_CHAT_IDS: ""
 # Telegram polling timeout in seconds.
 TG_POLL_TIMEOUT: 20
 
diff --git a/docs/reference-channels.md b/docs/reference-channels.md
index 8fc1d7e9..3f15fde5 100644
--- a/docs/reference-channels.md
+++ b/docs/reference-channels.md
@@ -50,10 +50,15 @@ Mattermost adapter using a bot token.
 
 Telegram adapter using Bot API long polling.
 
-- `start_telegram(chat_id, poll_timeout)` — starts a poll loop.
-- `TG_CHAT_ID` is optional; if empty, the adapter can auto-bind to the first valid inbound chat.
-- Outbound messages are chunked to Telegram-safe lengths.
-- Uses the same one-time `auth <secret>` ownership gate as the other adapters.
+- `start_telegram(chat_id, allowed_chat_ids, poll_timeout)` — validates saved authorization state and starts the poll loop.
+- With authentication enabled, the owner authenticates with `auth <secret>` in a private DM. That DM becomes the default destination for startup, heartbeat, and other proactive messages and is restored from persisted owner state after restart.
+- The owner uses `/bind` in a group to add its chat ID to the runtime and persisted allowed-group sets. `/unbind` removes the current group; `/unbind <group_id>` performs the same operation from the owner's DM. Targeted forms such as `/bind@BotName` and `/unbind@BotName` are supported.
+- `TG_ALLOWED_CHAT_IDS` supplies initial operator-configured chat IDs. Runtime `/bind` additions and `/unbind` removals are persisted in `memory/.channel/authenticated-group.json`; the YAML file itself is never rewritten.
+- Each inbound message is delivered to the agent as `[chat_id] [message_id] message`. Dequeueing does not depend on the model producing or successfully delivering a reply, so a no-response turn cannot freeze later inbound messages.
+- Outbound replies use the same `[chat_id] [message_id] message` envelope. An empty target falls back to the owner DM, and an empty message ID sends without Telegram reply metadata. Legacy plain outbound text also uses the owner-DM fallback.
+- Explicit LLM-generated targets are accepted only for the owner DM or a currently authorized group. Group chat IDs may be negative.
+- Outbound messages are split into Telegram-safe chunks and retained with their destination and reply ID for retry after transient delivery failures.
+- When `commchannel=telegram`, startup registers the routing instructions from `memory/tg_prompt.txt` through `add-prompt-extension`. Other channels do not receive this prompt section.
 
 ## `channels/slack.py`
 
diff --git a/docs/reference-configuration.md b/docs/reference-configuration.md
index 1fb5b661..2cc28f2c 100644
--- a/docs/reference-configuration.md
+++ b/docs/reference-configuration.md
@@ -42,7 +42,8 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command
 | `IRC_server` | `irc.quakenet.org` | IRC server hostname. |
 | `IRC_port` | 6667 | IRC port. |
 | `IRC_user` | `omega` | IRC nickname. |
-| `TG_CHAT_ID` | *(empty — auto-bind supported)* | Optional fixed Telegram chat ID. Leave empty to auto-bind on first valid inbound auth/message. |
+| `TG_CHAT_ID` | *(empty)* | Legacy single-chat bootstrap/fallback, primarily for authentication-disabled deployments. With authentication enabled, the authenticated owner's DM becomes the proactive-message default. |
+| `TG_ALLOWED_CHAT_IDS` | *(empty)* | Optional comma-separated initial allowed-chat set. The authenticated owner can add groups with `/bind` and remove them with `/unbind`; those changes are persisted in `memory/.channel/authenticated-group.json`, not written back to YAML. |
 | `TG_POLL_TIMEOUT` | 20 | Telegram long-poll timeout in seconds. |
 | `SL_CHANNEL_ID` | *(empty — auto-bind supported)* | Optional Slack channel ID where Omega reads/writes messages. Leave empty to auto-bind on first valid inbound auth/message. |
 | `SL_POLL_INTERVAL` | 60 | Slack poll interval in seconds (minimum effective value is 60). |
@@ -54,6 +55,7 @@ This reads a command-line override via `argk` (`name=value` on the MeTTa command
 | Environment variable | Meaning |
 |---|---|
 | `TG_BOT_TOKEN` | Telegram bot token (from BotFather). |
+| `OMEGA_AUTH_SECRET` | Enables the one-time owner-authentication handshake when non-empty. The owner sends `auth <secret>` in a private Telegram DM. |
 | `MM_BOT_TOKEN` | Bot auth token. |
 | `SL_BOT_TOKEN` | Slack bot token (`xoxb-...`). |
 
@@ -65,6 +67,18 @@ Any `configure`d parameter can be overridden at startup:
 metta run.metta provider=Anthropic LLM=claude-opus-4-6 commchannel=mattermost
 ```
 
+Configuration values are resolved in this order: command-line `key=value`,
+`OMEGACLAW_<KEY>` environment variable, `config/config.yaml`, then the caller's
+default. `TG_BOT_TOKEN` and `OMEGA_AUTH_SECRET` are read directly from the
+environment and must be placed before the `metta`/`petta` command.
+
+Telegram example:
+
+```bash
+TG_BOT_TOKEN=... OMEGA_AUTH_SECRET=... \
+  metta run.metta commchannel=telegram
+```
+
 Slack example:
 
 ```bash
diff --git a/memory/tg_prompt.txt b/memory/tg_prompt.txt
new file mode 100644
index 00000000..9336bc65
--- /dev/null
+++ b/memory/tg_prompt.txt
@@ -0,0 +1,19 @@
+TELEGRAM ROUTING:
+
+Incoming Telegram messages use this structure:
+[chat-id] [reply-id] message
+
+The first field is the destination Telegram chat ID. The second field is the
+Telegram message ID to reply to. Group chat IDs may be negative.
+
+When responding to an incoming Telegram message, every send command must begin
+with the exact chat ID and reply ID received with that message:
+send [chat-id] [reply-id] response
+Example:
+send [-3341114] [223] Hey!
+
+Copy both IDs exactly. Never invent, alter, exchange, or omit them when replying.
+Multiple send commands responding to the same message must all use the same IDs.
+
+An empty target routes to the authenticated owner's direct message. An empty
+reply ID sends a normal message without attaching it as a reply.
diff --git a/src/loop.metta b/src/loop.metta
index 9e9e6360..783b2daa 100644
--- a/src/loop.metta
+++ b/src/loop.metta
@@ -44,6 +44,15 @@
                          " HISTORY: " (getHistory) 
                          " TIME: " (get_time_as_string)))))
 
+(= (addTelegramPromptExtension)
+   (if (== (commchannel) telegram)
+       (let $path (joinPath ((memoryDirectory) "tg_prompt.txt"))
+            (if (exists-file $path)
+                (add-prompt-extension tg-prompt (read-file $path))
+                (progn (log WARN "loop" (TELEGRAM_PROMPT_NOT_FOUND: $path))
+                       ())))
+       ()))
+
 (= (getPromptExtensions)
    (join (newline) (collapse (prompt-extension $_))))
 
@@ -66,6 +75,7 @@
                                (initKnowledge)
                                (initPlugins)
                                (initChannels)
+                               (addTelegramPromptExtension)
                                (commChannelSend (version))
                                (llmProviderStart (provider)))
                         (change-state! &loops (- (get-state &loops) 1)))
diff --git a/tests/src_skills.metta b/tests/src_skills.metta
index 24ad96ac..18d764df 100644
--- a/tests/src_skills.metta
+++ b/tests/src_skills.metta
@@ -2,6 +2,12 @@
 !(import! &self ../src/helper.py)
 !(import! &self ./src/utils)
 !(import! &self ./src/skills)
+
+; Define the selected channel and memory path used by the Telegram prompt
+; registration test.
+(= (commchannel) telegram)
+(= (memoryDirectory) (joinPath ((projectRootDirectory) "memory")))
+
 !(import! &self ./src/loop)
 
 !(test (progn
@@ -24,6 +30,16 @@
          (contains-text (getPromptExtensions) "TEST PROMPT EXTENSION"))
        False)
 
+!(test (progn
+         (addTelegramPromptExtension)
+         (contains-text (getPromptExtensions) "TELEGRAM ROUTING:"))
+       True)
+
+!(test (progn
+         (remove-prompt-extension tg-prompt)
+         (contains-text (getPromptExtensions) "TELEGRAM ROUTING:"))
+       False)
+
 (= (on-heartbeat $iter)
    (let $count (get-state &test_heartbeat_counter)
      (change-state! &test_heartbeat_counter (+ $count $iter))))
diff --git a/tests/test_auth_standalone.py b/tests/test_auth_standalone.py
index ac594d66..011687bf 100644
--- a/tests/test_auth_standalone.py
+++ b/tests/test_auth_standalone.py
@@ -87,6 +87,23 @@ def test_saved_owner_blocks_auth_secret_reuse_after_restart(monkeypatch, tmp_pat
     assert owner_process.authenticate_channel_user("IRC", "owner") == "allow"
 
 
+def test_saved_owner_cannot_be_replaced_by_a_reused_secret(monkeypatch, tmp_path):
+    monkeypatch.setenv("OMEGA_AUTH_SECRET", "one-time-secret")
+
+    first_process = load_auth_module(monkeypatch)
+    monkeypatch.setattr(first_process, "_MEMORY_DIRECTORY", str(tmp_path))
+    assert first_process.authenticate_channel_user(
+        "TELEGRAM", "owner", "one-time-secret"
+    ) == "auth_bound"
+
+    restarted_process = load_auth_module(monkeypatch)
+    monkeypatch.setattr(restarted_process, "_MEMORY_DIRECTORY", str(tmp_path))
+    assert restarted_process.authenticate_channel_user(
+        "TELEGRAM", "attacker", "one-time-secret"
+    ) == "ignore"
+    assert restarted_process.get_channel_authenticated_user_id("TELEGRAM") == "owner"
+
+
 def test_plain_message_does_not_verify_a_token(monkeypatch):
     auth = load_auth_module(monkeypatch)
     monkeypatch.setattr(
@@ -95,3 +112,81 @@ def test_plain_message_does_not_verify_a_token(monkeypatch):
     monkeypatch.setattr(auth, "get_channel_authenticated_user_id", lambda *args: None)
 
     assert auth.authenticate_channel_user("IRC", "alice") == "ignore"
+
+
+def test_owner_can_revoke_an_authorized_group(monkeypatch, tmp_path):
+    auth = load_auth_module(monkeypatch)
+    monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path))
+
+    monkeypatch.setattr(auth, "get_channel_authenticated_user_id", lambda _channel: "owner")
+    assert auth.store_channel_authenticated_group_id("TELEGRAM", "group", "owner") is True
+    assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True
+
+    assert auth.revoke_channel_group("TELEGRAM", "group", "attacker") == "ignore"
+    assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True
+
+    assert auth.revoke_channel_group("TELEGRAM", "group", "owner") == "group_unbound"
+    assert auth.get_channel_saved_group_id("TELEGRAM", "group") is False
+
+    assert auth.store_channel_authenticated_group_id("TELEGRAM", "group", "owner") is True
+    assert auth.get_channel_saved_group_id("TELEGRAM", "group") is True
+
+
+def test_group_record_requires_authorizing_owner(monkeypatch, tmp_path):
+    auth = load_auth_module(monkeypatch)
+    monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path))
+
+    with pytest.raises(ValueError, match="authorized_by_user_id is required"):
+        auth.store_channel_authenticated_group_id("TELEGRAM", "group", "")
+
+
+def test_group_records_from_non_owner_are_not_loaded(monkeypatch, tmp_path):
+    auth = load_auth_module(monkeypatch)
+    monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path))
+
+    assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True
+    assert auth.store_channel_authenticated_group_id(
+        "TELEGRAM", "forged-group", "attacker"
+    ) is True
+
+    assert auth.get_channel_saved_group_id("TELEGRAM", "forged-group") is False
+    assert auth.load_channel_auth_state("TELEGRAM") == ("owner", set())
+
+
+def test_load_channel_auth_state_validates_and_loads_records(monkeypatch, tmp_path):
+    auth = load_auth_module(monkeypatch)
+    monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path))
+
+    assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True
+    assert auth.store_channel_authenticated_group_id("TELEGRAM", "active", "owner") is True
+    assert auth.store_channel_authenticated_group_id("TELEGRAM", "revoked", "owner") is True
+    assert auth.revoke_channel_group("TELEGRAM", "revoked", "owner") == "group_unbound"
+
+    owner, groups = auth.load_channel_auth_state("TELEGRAM")
+
+    assert owner == "owner"
+    assert groups == {"active"}
+
+
+@pytest.mark.parametrize("damaged_record", ["not-json\n", '{"time":'])
+def test_load_channel_auth_state_skips_malformed_records(
+    monkeypatch, tmp_path, damaged_record
+):
+    auth = load_auth_module(monkeypatch)
+    monkeypatch.setattr(auth, "_MEMORY_DIRECTORY", str(tmp_path))
+    warnings = []
+    monkeypatch.setattr(auth.logger, "warning", warnings.append)
+
+    assert auth.store_channel_authenticated_user_id("TELEGRAM", "owner") is True
+    assert auth.store_channel_authenticated_group_id(
+        "TELEGRAM", "active", "owner"
+    ) is True
+    path = tmp_path / ".channel" / "authenticated-group.json"
+    with path.open("a", encoding="utf-8") as target:
+        target.write(damaged_record)
+
+    assert auth.load_channel_auth_state("TELEGRAM") == ("owner", {"active"})
+    assert any(
+        "Skipping malformed channel authenticated group record at line 2" in warning
+        for warning in warnings
+    )
diff --git a/tests/test_channel_auth_gating.py b/tests/test_channel_auth_gating.py
index 91c7c301..1e1788af 100644
--- a/tests/test_channel_auth_gating.py
+++ b/tests/test_channel_auth_gating.py
@@ -14,7 +14,7 @@
     ("module_name", "arguments"),
     [
         ("irc", ("alice", "hello 🌍")),
-        ("telegram", ("chat", "alice", "hello 🌍")),
+        ("telegram", ("chat", "alice", "private", "hello 🌍")),
         ("slack", ("channel", "alice", "hello 🌍")),
         ("mattermost", ("alice", "hello 🌍")),
     ],
@@ -23,6 +23,9 @@ def test_unbound_plain_message_is_not_used_as_auth_token(monkeypatch, module_nam
     auth = types.ModuleType("auth")
     auth.is_auth_enabled = lambda: True
     auth.get_channel_saved_user_id = lambda *args: False
+    auth.get_channel_saved_group_id = lambda *args: False
+    auth.get_channel_authenticated_user_id = lambda *args: None
+    auth.authorize_channel_group = lambda *args: "ignore"
     calls = []
 
     def authenticate_channel_user(*args):
@@ -46,4 +49,7 @@ def authenticate_channel_user(*args):
     spec.loader.exec_module(module)
 
     assert module._is_allowed_message(*arguments) == "ignore"
-    assert calls == [(module_name.upper(), "alice", None)]
+    if module_name == "telegram":
+        assert calls == []
+    else:
+        assert calls == [(module_name.upper(), "alice", None)]
diff --git a/tests/test_telegram_multichat.py b/tests/test_telegram_multichat.py
new file mode 100644
index 00000000..37ca3376
--- /dev/null
+++ b/tests/test_telegram_multichat.py
@@ -0,0 +1,401 @@
+import importlib.util
+import json
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+CHANNELS_DIRECTORY = REPO_ROOT / "channels"
+
+
+def load_telegram(monkeypatch, auth_enabled=True):
+    state = {"owner": None, "groups": set()}
+    auth = types.ModuleType("auth")
+    auth.is_auth_enabled = lambda: auth_enabled
+    auth.get_proxy_url = lambda: ""
+    auth.load_channel_auth_state = lambda _channel: (
+        state["owner"],
+        {group for channel, group in state["groups"] if channel == "TELEGRAM"},
+    )
+    auth.get_channel_authenticated_user_id = lambda channel: state["owner"]
+    auth.get_channel_saved_group_id = lambda channel, group: (
+        channel, str(group)
+    ) in state["groups"]
+
+    def authenticate_channel_user(channel, user, candidate=None):
+        if candidate != "secret" or state["owner"] is not None:
+            return "ignore"
+        state["owner"] = str(user)
+        return "auth_bound"
+
+    def authorize_channel_group(channel, group, requester):
+        if str(requester) != state["owner"]:
+            return "ignore"
+        state["groups"].add((channel, str(group)))
+        return "group_bound"
+
+    def revoke_channel_group(channel, group, requester):
+        key = (channel, str(group))
+        if str(requester) != state["owner"] or key not in state["groups"]:
+            return "ignore"
+        state["groups"].remove(key)
+        return "group_unbound"
+
+    auth.authenticate_channel_user = authenticate_channel_user
+    auth.authorize_channel_group = authorize_channel_group
+    auth.revoke_channel_group = revoke_channel_group
+    monkeypatch.setitem(sys.modules, "auth", auth)
+
+    config = types.ModuleType("config")
+    config.config_get_by_key = lambda _key, default=None: default
+    monkeypatch.setitem(sys.modules, "config", config)
+    channels = types.ModuleType("channels")
+    channels.CommChannel = type("CommChannel", (), {"__init__": lambda self: None})
+    channels.registerCommChannel = lambda *_args: None
+    monkeypatch.setitem(sys.modules, "channels", channels)
+    monkeypatch.syspath_prepend(str(REPO_ROOT))
+    monkeypatch.syspath_prepend(str(CHANNELS_DIRECTORY))
+
+    path = CHANNELS_DIRECTORY / "telegram.py"
+    spec = importlib.util.spec_from_file_location("telegram_multichat_under_test", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_exposes_omega_plugin_entrypoint(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+
+    assert callable(telegram.loadOmegaPlugin)
+
+
+def test_reply_uses_the_chat_that_supplied_the_message(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._connected = True
+    telegram._admin_allowed_chats = {"101", "-202"}
+    sent = []
+    telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params))
+
+    telegram._enqueue_message("dm message", "101", 11)
+    telegram._enqueue_message("group message", "-202", 12)
+    assert telegram.getLastMessage() == "[101] [11] dm message"
+    telegram.send_message("[101] [11] dm reply")
+    assert telegram.getLastMessage() == "[-202] [12] group message"
+    telegram.send_message("[-202] [12] group reply")
+
+    assert [params["chat_id"] for _, params in sent] == ["101", "-202"]
+    assert [
+        json.loads(params["reply_parameters"])["message_id"]
+        for _, params in sent
+    ] == [11, 12]
+
+
+def test_identical_messages_from_different_chats_are_processed(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._connected = True
+    telegram._enqueue_message("same message", "101", 21)
+    telegram._enqueue_message("same message", "-202", 22)
+
+    assert telegram.getLastMessage() == "[101] [21] same message"
+    assert telegram.getLastMessage() == "[-202] [22] same message"
+    assert telegram.getLastMessage() == ""
+
+
+def test_failed_delivery_retains_the_original_chat(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._connected = True
+    telegram._admin_allowed_chats = {"101", "-202"}
+    attempts = []
+
+    def flaky_api(method, params, **_kwargs):
+        attempts.append((method, params.copy()))
+        if len(attempts) == 1:
+            raise RuntimeError("temporary failure")
+
+    telegram._api_call = flaky_api
+    telegram._enqueue_message("dm message", "101", 31)
+    assert telegram.getLastMessage() == "[101] [31] dm message"
+
+    telegram.send_message("[101] [31] dm reply")
+    telegram._enqueue_message("group message", "-202", 32)
+    # A failed outbound delivery does not block the inbound queue.
+    assert telegram.getLastMessage() == "[-202] [32] group message"
+
+    telegram._flush_outbox()
+    assert [params["chat_id"] for _, params in attempts] == ["101", "101"]
+
+
+def test_proactive_message_uses_authenticated_owner_dm(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._connected = True
+    telegram._owner_id = "101"
+    telegram._default_chat_id = "101"
+    sent = []
+    telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params))
+
+    telegram.send_message("startup message")
+
+    assert sent[0][1]["chat_id"] == "101"
+    assert "reply_parameters" not in sent[0][1]
+
+
+def test_proactive_message_before_auth_is_sent_after_owner_binding(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._connected = True
+    sent = []
+    telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params))
+
+    assert telegram.send_message("OmegaClaw version=test") is True
+    assert sent == []
+
+    telegram._process_update(
+        {
+            "message": {
+                "message_id": 42,
+                "text": "auth secret",
+                "chat": {"id": 101, "type": "private"},
+                "from": {"id": 101, "username": "owner"},
+            }
+        }
+    )
+
+    assert [params["text"] for _, params in sent] == [
+        "Authentication successful. @owner is now the bot owner. "
+        "Send /bind in a group to open it to everyone there.",
+        "OmegaClaw version=test",
+    ]
+    assert all(params["chat_id"] == "101" for _, params in sent)
+    assert json.loads(sent[0][1]["reply_parameters"]) == {
+        "message_id": 42,
+        "allow_sending_without_reply": True,
+    }
+    assert "reply_parameters" not in sent[1][1]
+
+def test_missing_route_fields_fall_back_to_owner_without_reply(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._connected = True
+    telegram._owner_id = "101"
+    telegram._default_chat_id = "101"
+    sent = []
+    telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params))
+
+    assert telegram.send_message("[] [] proactive message") is True
+
+    assert sent[0][1] == {"chat_id": "101", "text": "proactive message"}
+
+
+def test_generated_target_must_be_owner_or_authorized_group(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._connected = True
+    telegram._owner_id = "101"
+    telegram._default_chat_id = "101"
+    telegram._authorized_groups = {"-202"}
+    sent = []
+    telegram._api_call = lambda method, params, **_kwargs: sent.append((method, params))
+
+    assert telegram.send_message("[-202] [] allowed") is True
+    assert telegram.send_message("[-999] [] rejected") is False
+
+    assert [params["chat_id"] for _, params in sent] == ["-202"]
+
+
+def test_cached_authorization_avoids_disk_reads_per_message(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._owner_id = "owner"
+    telegram._authorized_groups = {"group"}
+    telegram.auth.get_channel_authenticated_user_id = lambda *_args: (_ for _ in ()).throw(
+        AssertionError("unexpected owner file read")
+    )
+    telegram.auth.get_channel_saved_group_id = lambda *_args: (_ for _ in ()).throw(
+        AssertionError("unexpected group file read")
+    )
+
+    assert telegram._is_allowed_message("owner-dm", "owner", "private", "hello") == "allow"
+    assert telegram._is_allowed_message("group", "member", "group", "hello") == "allow"
+
+
+def test_get_me_failure_does_not_abort_initialization(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._bot_username = "stale-name"
+    telegram._api_call = lambda *_args, **_kwargs: (_ for _ in ()).throw(
+        RuntimeError("temporary Telegram failure")
+    )
+
+    assert telegram._initialize_bot_identity() is False
+    assert telegram._bot_username == ""
+
+
+def test_invalid_auth_state_stops_startup_before_telegram_is_polled(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    monkeypatch.setenv("TG_BOT_TOKEN", "token")
+    telegram.auth.load_channel_auth_state = lambda _channel: (_ for _ in ()).throw(
+        RuntimeError("malformed authorization state")
+    )
+    telegram._api_call = lambda *_args, **_kwargs: (_ for _ in ()).throw(
+        AssertionError("Telegram API called before authorization validation")
+    )
+
+    with pytest.raises(RuntimeError, match="malformed authorization state"):
+        telegram.start_telegram()
+
+
+def test_startup_restores_owner_default_and_bound_groups(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    monkeypatch.setenv("TG_BOT_TOKEN", "token")
+    telegram.auth.load_channel_auth_state = lambda _channel: ("101", {"-202"})
+    telegram._initialize_bot_identity = lambda: True
+
+    class FakeThread:
+        def start(self):
+            return None
+
+    telegram.threading.Thread = lambda **_kwargs: FakeThread()
+
+    telegram.start_telegram(allowed_chat_ids="-303")
+
+    assert telegram._default_chat_id == "101"
+    assert telegram._authorized_groups == {"-202"}
+    assert telegram._admin_allowed_chats == {"-202", "-303"}
+
+
+def test_offset_advances_only_after_update_processing(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    update = {
+        "update_id": 7,
+        "message": {
+            "message_id": 41,
+            "text": "hello",
+            "chat": {"id": 101, "type": "private"},
+            "from": {"id": "user", "username": "alice"},
+        },
+    }
+    telegram._running = True
+    telegram._offset = None
+    telegram._flush_outbox = lambda: None
+
+    def get_updates(*_args, **_kwargs):
+        telegram._running = False
+        return [update]
+
+    telegram._api_call = get_updates
+    telegram._poll_loop()
+
+    assert telegram._offset == 8
+    assert telegram.getLastMessage() == "[101] [41] @alice: hello"
+
+
+def test_failed_update_processing_retains_offset(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._running = True
+    telegram._offset = None
+    telegram._flush_outbox = lambda: None
+    telegram.time.sleep = lambda _seconds: None
+
+    def get_updates(*_args, **_kwargs):
+        telegram._running = False
+        return [{"update_id": 7, "message": {"text": "hello"}}]
+
+    telegram._api_call = get_updates
+    telegram._process_update = lambda _update: (_ for _ in ()).throw(
+        RuntimeError("authorization state failure")
+    )
+    telegram._poll_loop()
+
+    assert telegram._offset is None
+
+
+def test_owner_binds_group_without_exposing_secret(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._bot_username = "examplebot"
+
+    # A secret sent in a group cannot establish an owner.
+    assert telegram._is_allowed_message("group", "1", "group", "auth secret") == "ignore"
+    # The owner authenticates once in a private DM.
+    assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound"
+    assert telegram._default_chat_id == "dm"
+    assert telegram._is_allowed_message("dm", "2", "private", "hello") == "ignore"
+    # Only that owner can open the group; then every group member is allowed.
+    assert telegram._is_allowed_message("group", "2", "group", "/bind") == "ignore"
+    assert telegram._is_allowed_message("group", "1", "group", "/bind@ExampleBot") == "group_bound"
+    assert "group" in telegram._authorized_groups
+    assert "group" in telegram._admin_allowed_chats
+    assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow"
+    assert telegram._is_allowed_message("other-group", "2", "group", "hello") == "ignore"
+
+
+def test_bind_command_targets_this_bot_only(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._bot_username = "examplebot"
+
+    assert telegram._is_bind_command("/bind")
+    assert telegram._is_bind_command("/authorize_group")
+
+    assert telegram._is_bind_command("/bind@ExampleBot")
+    assert telegram._is_bind_command("/BIND@examplebot")
+    assert telegram._is_bind_command("/authorize_group@ExampleBot")
+
+    assert not telegram._is_bind_command("/bind@AnotherBot")
+    assert not telegram._is_bind_command("/authorize_group@AnotherBot")
+    assert not telegram._is_bind_command("/binder@ExampleBot")
+
+    assert telegram._is_unbind_command("/unbind")
+    assert telegram._is_unbind_command("/UNBIND@examplebot")
+    assert not telegram._is_unbind_command("/unbind@AnotherBot")
+
+
+def test_only_owner_can_unbind_group(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._bot_username = "examplebot"
+
+    assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound"
+    assert telegram._is_allowed_message("group", "1", "group", "/bind") == "group_bound"
+    assert telegram._is_allowed_message("group", "2", "group", "/unbind") == "ignore"
+    assert telegram._is_allowed_message("group", "2", "group", "hello") == "allow"
+    assert telegram._is_allowed_message("group", "1", "group", "/unbind@ExampleBot") == "group_unbound"
+    assert "group" not in telegram._authorized_groups
+    assert "group" not in telegram._admin_allowed_chats
+    assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore"
+
+
+def test_owner_can_unbind_group_from_dm(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+
+    assert telegram._is_allowed_message("dm", "1", "private", "auth secret") == "auth_bound"
+    assert telegram._is_allowed_message("group", "1", "group", "/bind") == "group_bound"
+    assert telegram._is_allowed_message("dm", "1", "private", "/unbind group") == "group_unbound"
+    assert "group" not in telegram._authorized_groups
+    assert "group" not in telegram._admin_allowed_chats
+    assert telegram._is_allowed_message("group", "2", "group", "hello") == "ignore"
+
+
+def test_configured_chats_are_a_hard_boundary_when_auth_is_disabled(monkeypatch):
+    telegram = load_telegram(monkeypatch, auth_enabled=False)
+    telegram._admin_allowed_chats = telegram._parse_admin_allowed_chats(
+        "legacy-chat", "group-a, group-b"
+    )
+
+    assert telegram._is_allowed_message("legacy-chat", "1", "private", "hello") == "allow"
+    assert telegram._is_allowed_message("group-a", "2", "group", "hello") == "allow"
+    assert telegram._is_allowed_message("unlisted", "3", "group", "hello") == "ignore"
+
+
+def test_allowlist_still_permits_owner_dm_bootstrap(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._admin_allowed_chats = {"approved-group"}
+
+    assert telegram._is_allowed_message("owner-dm", "1", "private", "auth secret") == "auth_bound"
+    assert telegram._is_allowed_message("owner-dm", "1", "private", "hello") == "allow"
+    assert telegram._is_allowed_message("unlisted-group", "2", "group", "hello") == "ignore"
+
+
+def test_owner_bind_can_expand_configured_runtime_allowlist(monkeypatch):
+    telegram = load_telegram(monkeypatch)
+    telegram._admin_allowed_chats = {"configured-group"}
+
+    assert telegram._is_allowed_message("owner-dm", "1", "private", "auth secret") == "auth_bound"
+    assert telegram._is_allowed_message("new-group", "1", "group", "/bind") == "group_bound"
+    assert "new-group" in telegram._admin_allowed_chats
+    assert "new-group" in telegram._authorized_groups