#!/usr/bin/env python3
"""Data-minimised Telegram transport gateway MVP.

This gateway authenticates one Telegram group and one owner, receives a small
command set through long polling, and records metadata only. It deliberately
does not call an LLM and never executes Telegram message content.
"""

from __future__ import annotations

import argparse
from collections import deque
import errno
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import socket
import sqlite3
import stat
import struct
import sys
import tempfile
import threading
import time
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


class GatewayError(RuntimeError):
    """Expected configuration, Telegram, or runtime failure."""


class TelegramError(GatewayError):
    """Telegram API request failed without exposing the bot token."""


def emit(event: str, **details: Any) -> None:
    print(json.dumps({"event": event, **details}, ensure_ascii=False, separators=(",", ":")), flush=True)


def atomic_json(path: Path, payload: dict[str, Any], mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    descriptor, temporary = tempfile.mkstemp(prefix=path.name + ".", dir=str(path.parent))
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
            json.dump(payload, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
            os.fchmod(handle.fileno(), mode)
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)


def _is_systemd_runtime_credential(path: Path) -> bool:
    """Return true only for a file inside this unit's credential directory."""
    credentials_directory = os.environ.get("CREDENTIALS_DIRECTORY")
    if not credentials_directory:
        return False
    try:
        expected_parent = Path(credentials_directory).resolve(strict=True)
        actual_parent = path.parent.resolve(strict=True)
    except OSError:
        return False
    return actual_parent == expected_parent


def read_secure_token(path: Path) -> str:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    try:
        descriptor = os.open(path, flags)
    except OSError as error:
        raise GatewayError(f"無法讀取Telegram credential：{type(error).__name__}") from error
    try:
        file_stat = os.fstat(descriptor)
        if not stat.S_ISREG(file_stat.st_mode):
            raise GatewayError("Telegram credential必須係regular file")
        mode = stat.S_IMODE(file_stat.st_mode)
        runtime_credential = _is_systemd_runtime_credential(path)
        allowed_modes = {0o400, 0o600}
        if runtime_credential:
            # systemd exposes a read-only, per-unit credential copy.  Ubuntu
            # 24.04 may present it as root:<service-group> 0440.
            allowed_modes.add(0o440)
        if mode not in allowed_modes:
            raise GatewayError(
                f"Telegram credential權限不符合安全模式：{mode:04o}；"
                "一般檔案只接受0400/0600，systemd runtime credential先接受0440"
            )
        with os.fdopen(descriptor, "rb", closefd=False) as handle:
            raw = handle.read(514)
        if len(raw) > 513:
            raise GatewayError("Telegram credential格式無效")
        try:
            token = raw.decode("utf-8").strip()
        except UnicodeDecodeError as error:
            raise GatewayError("Telegram credential格式無效") from error
    finally:
        os.close(descriptor)
    if not token or len(token) > 512 or any(char.isspace() for char in token):
        raise GatewayError("Telegram credential格式無效")
    return token


def load_config(path: Path) -> dict[str, Any]:
    try:
        config = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise GatewayError(f"設定檔無效：{type(error).__name__}") from error
    if not isinstance(config, dict) or config.get("version") != 1:
        raise GatewayError("設定檔version無效")
    expected_username = config.get("expected_bot_username")
    if not isinstance(expected_username, str) or not expected_username.strip("@"):
        raise GatewayError("expected_bot_username無效")
    config["expected_bot_username"] = expected_username.strip("@")
    for key in ("allowed_chat_id", "owner_user_id", "initial_update_offset"):
        if not isinstance(config.get(key), int):
            raise GatewayError(f"{key}必須係整數")
    if config["owner_user_id"] <= 0 or config["initial_update_offset"] < 0:
        raise GatewayError("owner_user_id或initial_update_offset無效")
    thread_id = config.get("allowed_thread_id")
    if thread_id is not None and (not isinstance(thread_id, int) or thread_id <= 0):
        raise GatewayError("allowed_thread_id必須係null或正整數")
    poll_timeout = config.get("poll_timeout_seconds", 30)
    max_command_chars = config.get("max_command_chars", 4000)  # PATCH-4 default
    rate_limit = config.get("owner_commands_per_minute", 12)
    if not isinstance(poll_timeout, int) or not 5 <= poll_timeout <= 50:
        raise GatewayError("poll_timeout_seconds必須介乎5至50")
    if not isinstance(max_command_chars, int) or not 100 <= max_command_chars <= 4000:
        raise GatewayError("max_command_chars必須介乎100至4000")
    if not isinstance(rate_limit, int) or not 1 <= rate_limit <= 60:
        raise GatewayError("owner_commands_per_minute必須介乎1至60")
    config["poll_timeout_seconds"] = poll_timeout
    config["max_command_chars"] = max_command_chars
    config["owner_commands_per_minute"] = rate_limit
    return config


class TelegramAPI:
    def __init__(self, token: str, api_base: str = "https://api.telegram.org") -> None:
        self.token = token
        self.api_base = api_base.rstrip("/")

    def call(self, method: str, payload: dict[str, Any] | None = None, timeout: float = 20) -> Any:
        endpoint = f"{self.api_base}/bot{self.token}/{method}"
        body = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
        request = Request(
            endpoint,
            data=body,
            headers={
                "Accept": "application/json",
                "Content-Type": "application/json",
                "User-Agent": "AIResearchCouncilTelegramGateway/0.1",
            },
            method="POST",
        )
        try:
            with urlopen(request, timeout=timeout) as response:
                raw = response.read(2_000_001)
        except HTTPError as error:
            raw = error.read(200_000)
            description = "Telegram API HTTP error"
            try:
                parsed = json.loads(raw.decode("utf-8"))
                description = str(parsed.get("description", description))
            except Exception:
                pass
            description = description.replace(self.token, "[REDACTED]")[:300]
            raise TelegramError(f"{method} HTTP {error.code}: {description}") from error
        except URLError as error:
            raise TelegramError(f"{method}連線失敗：{type(error.reason).__name__}") from error
        if len(raw) > 2_000_000:
            raise TelegramError(f"{method}回應過大")
        try:
            result = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as error:
            raise TelegramError(f"{method}回應唔係有效JSON") from error
        if not isinstance(result, dict) or result.get("ok") is not True:
            description = str(result.get("description", "Telegram API回應失敗")) if isinstance(result, dict) else "Telegram API格式錯誤"
            description = description.replace(self.token, "[REDACTED]")[:300]
            raise TelegramError(f"{method}失敗：{description}")
        return result.get("result")


def verify_bot(api: TelegramAPI, expected_username: str) -> dict[str, Any]:
    bot = api.call("getMe")
    if not isinstance(bot, dict) or not isinstance(bot.get("id"), int):
        raise GatewayError("getMe回應欠缺bot ID")
    actual_username = str(bot.get("username", ""))
    if actual_username.casefold() != expected_username.strip("@").casefold():
        raise GatewayError(f"Bot username不符：預期@{expected_username.strip('@')}，實際@{actual_username or '(none)'}")
    webhook = api.call("getWebhookInfo")
    if isinstance(webhook, dict) and webhook.get("url"):
        raise GatewayError("Bot已配置webhook；long-polling MVP拒絕接管。請先明確移除webhook。")
    return bot


def extract_candidates(updates: Any, expected_username: str) -> tuple[list[dict[str, Any]], int]:
    if not isinstance(updates, list):
        raise GatewayError("getUpdates回應格式無效")
    candidates_by_key: dict[tuple[int, int, int | None], dict[str, Any]] = {}
    max_update_id = -1
    for update in updates:
        if not isinstance(update, dict) or not isinstance(update.get("update_id"), int):
            continue
        update_id = int(update["update_id"])
        max_update_id = max(max_update_id, update_id)
        message = update.get("message")
        if not isinstance(message, dict):
            continue
        chat = message.get("chat")
        sender = message.get("from")
        if not isinstance(chat, dict) or not isinstance(sender, dict):
            continue
        if chat.get("type") not in ("group", "supergroup") or sender.get("is_bot") is True:
            continue
        chat_id = chat.get("id")
        user_id = sender.get("id")
        if not isinstance(chat_id, int) or not isinstance(user_id, int):
            continue
        thread_id = message.get("message_thread_id")
        if not isinstance(thread_id, int):
            thread_id = None
        text = message.get("text")
        command_seen = False
        if isinstance(text, str):
            first = text.strip().split(maxsplit=1)[0] if text.strip() else ""
            command_seen = first.casefold() in ("/ask", f"/ask@{expected_username}".casefold(), "/ping", f"/ping@{expected_username}".casefold())
        candidate = {
            "update_id": update_id,
            "chat_id": chat_id,
            "chat_type": str(chat.get("type", "")),
            "chat_title": str(chat.get("title", ""))[:120],
            "owner_user_id": user_id,
            "owner_username": str(sender.get("username", ""))[:80],
            "message_id": int(message.get("message_id", 0)),
            "thread_id": thread_id,
            "command_seen": command_seen,
        }
        key = (chat_id, user_id, thread_id)
        previous = candidates_by_key.get(key)
        if previous is None or update_id > int(previous["update_id"]):
            candidates_by_key[key] = candidate
    candidates = sorted(candidates_by_key.values(), key=lambda item: int(item["update_id"]), reverse=True)
    return candidates, max_update_id


def probe_command(args: argparse.Namespace) -> int:
    token = read_secure_token(args.token_file)
    api = TelegramAPI(token, args.api_base)
    bot = verify_bot(api, args.expected_username)
    updates = api.call(
        "getUpdates",
        {"timeout": args.wait_seconds, "allowed_updates": ["message"]},
        timeout=args.wait_seconds + 15,
    )
    candidates, max_update_id = extract_candidates(updates, args.expected_username.strip("@"))
    output = {
        "version": 1,
        "bot_id": int(bot["id"]),
        "bot_username": str(bot.get("username", "")),
        "privacy_mode_enabled": not bool(bot.get("can_read_all_group_messages", False)),
        "max_update_id": max_update_id,
        "candidates": candidates,
    }
    atomic_json(args.output, output)
    print(f"Bot驗證成功：@{output['bot_username']}（ID {output['bot_id']}）")
    print(f"Privacy Mode：{'enabled' if output['privacy_mode_enabled'] else 'disabled'}")
    if not candidates:
        print("暫時搵唔到group／supergroup候選update。")
    for index, item in enumerate(candidates, start=1):
        thread = item["thread_id"] if item["thread_id"] is not None else "any/no-topic"
        username = f"@{item['owner_username']}" if item["owner_username"] else "(no username)"
        marker = "command-seen" if item["command_seen"] else "message-seen"
        print(
            f"[{index}] chat_id={item['chat_id']} title={item['chat_title']!r} "
            f"owner_user_id={item['owner_user_id']} owner={username} thread={thread} {marker}"
        )
    return 0


def acquire_lock(path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = path.open("a+", encoding="utf-8")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as error:
        handle.close()
        raise GatewayError("已有另一個Telegram gateway運行中") from error
    handle.seek(0)
    handle.truncate()
    handle.write(f"{os.getpid()}\n")
    handle.flush()
    os.fchmod(handle.fileno(), 0o600)
    return handle


def config_fingerprint(config: dict[str, Any]) -> str:
    binding = {
        "expected_bot_username": config["expected_bot_username"].casefold(),
        "allowed_chat_id": config["allowed_chat_id"],
        "owner_user_id": config["owner_user_id"],
        "allowed_thread_id": config.get("allowed_thread_id"),
    }
    encoded = json.dumps(binding, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def open_database(path: Path, initial_offset: int, binding_fingerprint: str) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    connection = sqlite3.connect(path, timeout=10)
    connection.execute("PRAGMA journal_mode=WAL")
    connection.execute("PRAGMA synchronous=FULL")
    connection.execute("PRAGMA busy_timeout=10000")
    connection.executescript(
        """
        CREATE TABLE IF NOT EXISTS gateway_state (
            key TEXT PRIMARY KEY,
            value TEXT NOT NULL
        );
        CREATE TABLE IF NOT EXISTS audit_events (
            update_id INTEGER PRIMARY KEY,
            chat_id INTEGER,
            owner_user_id INTEGER,
            message_id INTEGER,
            thread_id INTEGER,
            command TEXT NOT NULL,
            input_chars INTEGER NOT NULL,
            status TEXT NOT NULL,
            reply_message_id INTEGER,
            processed_at TEXT NOT NULL,
            error_type TEXT
        );
        """
    )
    connection.execute(
        "INSERT OR IGNORE INTO gateway_state(key, value) VALUES('next_update_id', ?)",
        (str(initial_offset),),
    )
    existing_binding = connection.execute(
        "SELECT value FROM gateway_state WHERE key='config_fingerprint'"
    ).fetchone()
    if existing_binding is None:
        connection.execute(
            "INSERT INTO gateway_state(key, value) VALUES('config_fingerprint', ?)",
            (binding_fingerprint,),
        )
    elif str(existing_binding[0]) != binding_fingerprint:
        connection.close()
        raise GatewayError(
            "現有metadata DB屬於另一個chat／owner／thread綁定；拒絕靜默重用"
        )
    connection.commit()
    os.chmod(path, 0o600)
    return connection


def state_int(connection: sqlite3.Connection, key: str) -> int:
    row = connection.execute("SELECT value FROM gateway_state WHERE key = ?", (key,)).fetchone()
    if row is None:
        raise GatewayError(f"gateway state欠缺{key}")
    try:
        return int(row[0])
    except (TypeError, ValueError) as error:
        raise GatewayError(f"gateway state {key}無效") from error


def set_state(connection: sqlite3.Connection, key: str, value: str | int) -> None:
    connection.execute(
        "INSERT INTO gateway_state(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
        (key, str(value)),
    )


def reserve_update(
    connection: sqlite3.Connection,
    update_id: int,
    next_offset: int,
    chat_id: int | None,
    owner_user_id: int | None,
    message_id: int | None,
    thread_id: int | None,
    command: str,
    input_chars: int,
    status: str,
) -> bool:
    try:
        connection.execute("BEGIN IMMEDIATE")
        cursor = connection.execute(
            """
            INSERT OR IGNORE INTO audit_events(
                update_id, chat_id, owner_user_id, message_id, thread_id,
                command, input_chars, status, processed_at
            ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%SZ','now'))
            """,
            (update_id, chat_id, owner_user_id, message_id, thread_id, command, input_chars, status),
        )
        set_state(connection, "next_update_id", next_offset)
        connection.commit()
        return cursor.rowcount == 1
    except Exception:
        connection.rollback()
        raise


def finish_update(
    connection: sqlite3.Connection,
    update_id: int,
    status: str,
    reply_message_id: int | None = None,
    error_type: str | None = None,
) -> None:
    connection.execute(
        "UPDATE audit_events SET status=?, reply_message_id=?, error_type=? WHERE update_id=?",
        (status, reply_message_id, error_type, update_id),
    )
    connection.commit()


def parse_command(text: str, expected_username: str) -> tuple[str | None, str]:
    stripped = text.strip()
    if not stripped.startswith("/"):
        return None, ""
    token, separator, argument = stripped.partition(" ")
    command_target = token[1:]
    command_name, at, target = command_target.partition("@")
    if at and target.casefold() != expected_username.casefold():
        return None, ""
    return command_name.casefold(), argument.strip() if separator else ""


def send_text(
    api: TelegramAPI,
    chat_id: int,
    text: str,
    thread_id: int | None = None,
    reply_to: int | None = None,
) -> int:
    payload: dict[str, Any] = {
        "chat_id": chat_id,
        "text": text,
        "link_preview_options": {"is_disabled": True},
    }
    if thread_id is not None:
        payload["message_thread_id"] = thread_id
    if reply_to is not None:
        payload["reply_parameters"] = {"message_id": reply_to, "allow_sending_without_reply": True}
    result = api.call("sendMessage", payload)
    if not isinstance(result, dict) or not isinstance(result.get("message_id"), int):
        raise TelegramError("sendMessage成功回應欠缺message_id")
    return int(result["message_id"])


COUNCIL_CONTROLLER_IN_SOCKET = Path("/run/ai-council-controller/gateway-in.sock")
COUNCIL_GATEWAY_OUT_SOCKET = Path("/run/ai-research-council/council-out.sock")
COUNCIL_FRAME_MAX_BYTES = 65536
COUNCIL_DATA_CLASS = "SYNTHETIC_PUBLIC_ONLY"
COUNCIL_PREFIXES = {
    "ENTRY": "[ARC/ENTRY/v1]",
    "BENCH": "[ARC/BENCH/v1]",
    "COMMONS": "[ARC/COMMONS/v1]",
    "JOB": "[ARC/JOB/v1]",
    "ENGINEERING": "[ARC/ENG/v1]",
    "REVIEW": "[ARC/REVIEW/v1]",
    "APPROVAL": "[ARC/APPROVAL/v1]",
    "INCIDENT": "[ARC/INCIDENT/v1]",
}
COUNCIL_HEX32 = re.compile(r"[0-9a-f]{32}\Z")
COUNCIL_HEX64 = re.compile(r"[0-9a-f]{64}\Z")
COUNCIL_ID = re.compile(r"[A-Za-z0-9_.:-]{1,64}\Z")
COUNCIL_CALLBACK = re.compile(r"ARC1:[AR]:[A-Za-z0-9_-]{16}:[A-Za-z0-9_-]{8}:[A-Za-z0-9_-]{22}\Z")
COUNCIL_PLAN_ID = re.compile(r"[A-Za-z0-9_-]{16}\Z")
COUNCIL_CALLBACK_QUERY_ID = re.compile(r"[A-Za-z0-9_.:-]{1,128}\Z")


def council_canonical_json(value: Any) -> bytes:
    try:
        return json.dumps(
            value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False,
        ).encode("utf-8")
    except (TypeError, ValueError) as error:
        raise GatewayError("Council frame格式無效") from error


def _strict_json_object(raw: bytes) -> dict[str, Any]:
    def pairs(values: list[tuple[str, Any]]) -> dict[str, Any]:
        output: dict[str, Any] = {}
        for key, value in values:
            if key in output:
                raise ValueError("duplicate key")
            output[key] = value
        return output

    def invalid_constant(_value: str) -> None:
        raise ValueError("invalid constant")

    try:
        parsed = json.loads(
            raw.decode("utf-8"), object_pairs_hook=pairs, parse_constant=invalid_constant,
        )
    except (UnicodeError, json.JSONDecodeError, ValueError) as error:
        raise GatewayError("Council frame格式無效") from error
    if not isinstance(parsed, dict):
        raise GatewayError("Council frame必須係object")
    return parsed


def _recv_exact(connection: socket.socket, size: int) -> bytes:
    output = bytearray()
    while len(output) < size:
        chunk = connection.recv(size - len(output))
        if not chunk:
            raise GatewayError("Council frame提早結束")
        output.extend(chunk)
    return bytes(output)


def council_send_frame(connection: socket.socket, payload: dict[str, Any]) -> None:
    body = council_canonical_json(payload)
    if not 1 <= len(body) <= COUNCIL_FRAME_MAX_BYTES:
        raise GatewayError("Council frame超出上限")
    connection.sendall(struct.pack("!I", len(body)) + body)


def council_receive_frame(connection: socket.socket) -> dict[str, Any]:
    header = _recv_exact(connection, 4)
    length = struct.unpack("!I", header)[0]
    if not 1 <= length <= COUNCIL_FRAME_MAX_BYTES:
        raise GatewayError("Council frame超出上限")
    body = _recv_exact(connection, length)
    parsed = _strict_json_object(body)
    if council_canonical_json(parsed) != body:
        raise GatewayError("Council frame必須係canonical JSON")
    return parsed


def council_peer_uid(connection: socket.socket) -> int:
    if not hasattr(socket, "SO_PEERCRED"):
        raise GatewayError("平台唔支援SO_PEERCRED")
    raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))
    _pid, uid, _gid = struct.unpack("3i", raw)
    return int(uid)


def _safe_council_text(text: Any) -> str:
    if not isinstance(text, str):
        raise GatewayError("Council text無效")
    encoded = text.encode("utf-8")
    if not 1 <= len(text) <= 4000 or len(encoded) > 12000:  # PATCH-5 (Owner approved)
        raise GatewayError("Council text超出上限")
    if any(ord(character) < 32 and character not in "\n\t" for character in text):
        raise GatewayError("Council text包含控制字元")
    return text


def parse_council_argument(argument: str) -> tuple[str, str]:
    for room_id, prefix in COUNCIL_PREFIXES.items():
        marker = prefix + " " + COUNCIL_DATA_CLASS + " "
        if argument.startswith(marker):
            return room_id, _safe_council_text(argument[len(marker):])
    raise GatewayError(
        "Council格式無效；要用固定room prefix同SYNTHETIC_PUBLIC_ONLY標記"
    )


def build_gateway_council_frame(
    update_id: int, chat_id: int, owner_user_id: int, message_id: int,
    thread_id: int | None, room_id: str, text: str,
) -> dict[str, Any]:
    if room_id not in COUNCIL_PREFIXES:
        raise GatewayError("Council room無效")
    text = _safe_council_text(text)
    routing = {
        "update_id": update_id,
        "chat_id": chat_id,
        "owner_user_id": owner_user_id,
        "message_id": message_id,
        "thread_id": thread_id,
        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
    }
    event_digest = hashlib.sha256(council_canonical_json(routing)).hexdigest()
    return {
        "schema": "ARC_GATEWAY_TO_CONTROLLER_V1",
        "op": "owner_council_message",
        "request_id": event_digest[:32],
        "gateway_event_sha256": event_digest,
        "room_id": room_id,
        "fixed_prefix": COUNCIL_PREFIXES[room_id],
        "data_class": COUNCIL_DATA_CLASS,
        "text": text,
        "text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
    }


def forward_council_message(
    frame: dict[str, Any], socket_path: Path = COUNCIL_CONTROLLER_IN_SOCKET,
    expected_peer_uid: int | None = None, timeout: float = 2.0,
) -> dict[str, Any]:
    if expected_peer_uid is None:
        expected_peer_uid = os.geteuid()
    connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        connection.settimeout(timeout)
        connection.connect(str(socket_path))
        if council_peer_uid(connection) != expected_peer_uid:
            raise GatewayError("Council controller peer不符")
        council_send_frame(connection, frame)
        response = council_receive_frame(connection)
    except (OSError, TimeoutError) as error:
        raise GatewayError("Council controller暫時不可用") from error
    finally:
        connection.close()
    if set(response) != {"schema", "request_id", "status", "room_id", "event_sha256"}:
        raise GatewayError("Council controller回應欄位無效")
    if (
        response["schema"] != "ARC_CONTROLLER_ACK_V1"
        or response["request_id"] != frame.get("request_id")
        or response["room_id"] != frame.get("room_id")
        or response["status"] not in {"ACCEPTED", "DUPLICATE"}
        or not isinstance(response["event_sha256"], str)
        or not COUNCIL_HEX64.fullmatch(response["event_sha256"])
    ):
        raise GatewayError("Council controller回應無效")
    return response


def extract_owner_changeplan_callback(update: Any, config: dict[str, Any], approvals_enabled: bool) -> dict[str, Any] | None:
    if not approvals_enabled or not isinstance(update, dict):
        return None
    query = update.get("callback_query")
    if not isinstance(query, dict):
        return None
    query_id, sender, message, data = query.get("id"), query.get("from"), query.get("message"), query.get("data")
    if (not isinstance(query_id, str) or COUNCIL_CALLBACK_QUERY_ID.fullmatch(query_id) is None or
            not isinstance(sender, dict) or sender.get("is_bot") is True or sender.get("id") != config["owner_user_id"] or
            not isinstance(message, dict) or not isinstance(data, str) or COUNCIL_CALLBACK.fullmatch(data) is None):
        return None
    chat = message.get("chat")
    if (not isinstance(chat, dict) or chat.get("type") not in {"group", "supergroup"} or
            chat.get("id") != config["allowed_chat_id"] or not isinstance(message.get("message_id"), int) or message["message_id"] <= 0):
        return None
    thread_id = message.get("message_thread_id") if isinstance(message.get("message_thread_id"), int) else None
    if config.get("allowed_thread_id") is not None and thread_id != config["allowed_thread_id"]:
        return None
    return {"callback_query_id": query_id, "chat_id": chat["id"], "owner_user_id": sender["id"],
            "message_id": message["message_id"], "thread_id": thread_id, "callback_data": data}


def build_gateway_callback_frame(callback: dict[str, Any]) -> dict[str, Any]:
    query_id, data = callback.get("callback_query_id"), callback.get("callback_data")
    if (not isinstance(query_id, str) or COUNCIL_CALLBACK_QUERY_ID.fullmatch(query_id) is None or
            not isinstance(data, str) or COUNCIL_CALLBACK.fullmatch(data) is None):
        raise GatewayError("Council callback格式無效")
    routing = {"callback_query_id_sha256": hashlib.sha256(query_id.encode()).hexdigest(),
               "callback_data_sha256": hashlib.sha256(data.encode()).hexdigest()}
    digest = hashlib.sha256(council_canonical_json(routing)).hexdigest()
    return {"schema": "ARC_GATEWAY_TO_CONTROLLER_V1", "op": "owner_changeplan_callback",
            "request_id": digest[:32], "gateway_event_sha256": digest,
            "callback_query_id_sha256": routing["callback_query_id_sha256"], "room_id": "APPROVAL",
            "data_class": COUNCIL_DATA_CLASS, "callback_data": data,
            "callback_data_sha256": routing["callback_data_sha256"]}


def forward_council_callback(frame: dict[str, Any], socket_path: Path = COUNCIL_CONTROLLER_IN_SOCKET,
                             expected_peer_uid: int | None = None, timeout: float = 2.0) -> dict[str, Any]:
    expected_peer_uid = os.geteuid() if expected_peer_uid is None else expected_peer_uid
    connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        connection.settimeout(timeout); connection.connect(str(socket_path))
        if council_peer_uid(connection) != expected_peer_uid:
            raise GatewayError("Council controller peer不符")
        council_send_frame(connection, frame); response = council_receive_frame(connection)
    except (OSError, TimeoutError) as error:
        raise GatewayError("Council controller暫時不可用") from error
    finally:
        connection.close()
    fields = {"schema", "request_id", "status", "decision", "plan_id", "plan_sha256"}
    if (set(response) != fields or response["schema"] != "ARC_OWNER_DECISION_ACK_V1" or
            response["request_id"] != frame.get("request_id") or
            response["status"] not in {"DECIDED", "STAGED_VERIFIED", "APPLY_FAILED_ROLLED_BACK"} or
            response["decision"] not in {"APPROVE", "REJECT"} or
            (response["decision"] == "REJECT" and response["status"] != "DECIDED") or
            (response["status"] in {"STAGED_VERIFIED", "APPLY_FAILED_ROLLED_BACK"} and response["decision"] != "APPROVE") or
            not isinstance(response["plan_id"], str) or COUNCIL_PLAN_ID.fullmatch(response["plan_id"]) is None or
            not isinstance(response["plan_sha256"], str) or COUNCIL_HEX64.fullmatch(response["plan_sha256"]) is None):
        raise GatewayError("Council callback controller回應無效")
    return response


def _validate_socket_parent(path: Path) -> None:
    parent = path.parent
    if not parent.is_dir() or parent.is_symlink():
        raise GatewayError("Council socket parent無效")
    information = parent.stat()
    if information.st_uid != os.geteuid() or stat.S_IMODE(information.st_mode) not in {0o700, 0o750}:
        raise GatewayError("Council socket parent owner/mode無效")


def _remove_stale_owned_socket(path: Path) -> None:
    try:
        before = path.lstat()
    except FileNotFoundError:
        return
    if (
        not stat.S_ISSOCK(before.st_mode)
        or before.st_uid != os.geteuid()
        or stat.S_IMODE(before.st_mode) != 0o600
    ):
        raise GatewayError("Council socket collision")
    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        probe.settimeout(0.2)
        probe.connect(str(path))
    except OSError as error:
        if error.errno not in {errno.ECONNREFUSED, errno.ENOENT}:
            raise GatewayError("Council socket collision") from error
    else:
        raise GatewayError("另一個Council adapter已運行")
    finally:
        probe.close()
    after = path.lstat()
    if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino):
        raise GatewayError("Council socket喺檢查期間改變")
    path.unlink()


class CouncilOutboundServer:
    """Peer-checked typed adapter; approval operations require explicit A3 enablement."""

    def __init__(
        self, api: TelegramAPI, config: dict[str, Any],
        socket_path: Path = COUNCIL_GATEWAY_OUT_SOCKET,
        expected_peer_uid: int | None = None,
        approvals_enabled: bool = False,
    ) -> None:
        self.api = api
        self.config = config
        self.socket_path = socket_path
        self.expected_peer_uid = os.geteuid() if expected_peer_uid is None else expected_peer_uid
        self.approvals_enabled = approvals_enabled
        self.listener: socket.socket | None = None
        self.listener_identity: tuple[int, int] | None = None
        self.stop_event = threading.Event()
        self.thread: threading.Thread | None = None

    def start(self) -> None:
        _validate_socket_parent(self.socket_path)
        _remove_stale_owned_socket(self.socket_path)
        listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            listener.bind(str(self.socket_path))
            os.chmod(self.socket_path, 0o600)
            info = self.socket_path.lstat()
            if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o600:
                raise GatewayError("Council outbound socket metadata無效")
            listener.listen(8)
            listener.settimeout(0.2)
        except BaseException:
            listener.close()
            if self.socket_path.exists() or self.socket_path.is_symlink():
                try:
                    self.socket_path.unlink()
                except OSError:
                    pass
            raise
        self.listener = listener
        self.listener_identity = (info.st_dev, info.st_ino)
        self.thread = threading.Thread(target=self._run, name="council-gateway-out", daemon=True)
        self.thread.start()

    def _dispatch(self, request: dict[str, Any]) -> dict[str, Any]:
        if (not isinstance(request, dict) or request.get("schema") != "ARC_CONTROLLER_TO_GATEWAY_V1" or
                not isinstance(request.get("request_id"), str) or COUNCIL_HEX32.fullmatch(request["request_id"]) is None):
            raise GatewayError("Council outbound frame值無效")
        operation = request.get("op")
        if operation == "send_prefixed_text":
            expected = {"schema", "op", "request_id", "room_id", "data_class", "sender_label", "text", "text_sha256"}
            if set(request) != expected: raise GatewayError("Council outbound frame欄位無效")
            room_id, text = request["room_id"], _safe_council_text(request["text"])
            if (room_id not in COUNCIL_PREFIXES or request["data_class"] != COUNCIL_DATA_CLASS or
                    not isinstance(request["sender_label"], str) or COUNCIL_ID.fullmatch(request["sender_label"]) is None or
                    request["text_sha256"] != hashlib.sha256(text.encode()).hexdigest()):
                raise GatewayError("Council outbound frame值無效")
            reply_id = send_text(self.api, self.config["allowed_chat_id"], f"{COUNCIL_PREFIXES[room_id]}\nAgent {request['sender_label']}: {text}", self.config.get("allowed_thread_id")); status = "SENT"
        elif operation == "send_approval_card":
            expected = {"schema", "op", "request_id", "room_id", "data_class", "plan_id", "plan_sha256", "target_scope_sha256", "expires_at", "approve_callback_data", "reject_callback_data"}
            if set(request) != expected or not self.approvals_enabled: raise GatewayError("Council approval operation未啟用")
            approve, reject = request["approve_callback_data"], request["reject_callback_data"]
            if (request["room_id"] != "APPROVAL" or request["data_class"] != COUNCIL_DATA_CLASS or
                    not isinstance(request["plan_id"], str) or COUNCIL_PLAN_ID.fullmatch(request["plan_id"]) is None or
                    not isinstance(request["plan_sha256"], str) or COUNCIL_HEX64.fullmatch(request["plan_sha256"]) is None or
                    not isinstance(request["target_scope_sha256"], str) or COUNCIL_HEX64.fullmatch(request["target_scope_sha256"]) is None or
                    not isinstance(request["expires_at"], int) or isinstance(request["expires_at"], bool) or request["expires_at"] <= 0 or
                    not isinstance(approve, str) or COUNCIL_CALLBACK.fullmatch(approve) is None or
                    not isinstance(reject, str) or COUNCIL_CALLBACK.fullmatch(reject) is None or
                    not approve.startswith("ARC1:A:") or not reject.startswith("ARC1:R:") or
                    approve.split(":")[2] != request["plan_id"] or reject.split(":")[2] != request["plan_id"] or
                    approve.split(":")[3] != reject.split(":")[3]):
                raise GatewayError("Council approval card值無效")
            text = (f"{COUNCIL_PREFIXES['APPROVAL']}\nFrozen synthetic ChangePlan待Owner一次性決定\n"
                    f"plan_id: {request['plan_id']}\nplan_sha256: {request['plan_sha256']}\n"
                    f"target_scope_sha256: {request['target_scope_sha256']}\nexpires_at: {request['expires_at']}\n"
                    "只限SYNTHETIC_PUBLIC_ONLY staging；唔會改target Ubuntu。")
            payload: dict[str, Any] = {"chat_id": self.config["allowed_chat_id"], "text": text, "disable_web_page_preview": True,
                "reply_markup": {"inline_keyboard": [[{"text": "Approve once", "callback_data": approve}, {"text": "Reject", "callback_data": reject}]]}}
            if self.config.get("allowed_thread_id") is not None: payload["message_thread_id"] = self.config["allowed_thread_id"]
            result = self.api.call("sendMessage", payload)
            if not isinstance(result, dict) or not isinstance(result.get("message_id"), int): raise TelegramError("sendMessage成功回應欠缺message_id")
            reply_id, status = int(result["message_id"]), "SENT"
        elif operation == "answer_callback":
            expected = {"schema", "op", "request_id", "callback_query_id", "outcome"}
            if set(request) != expected or not self.approvals_enabled: raise GatewayError("Council callback answer未啟用")
            messages = {
                "APPROVED": "ChangePlan已批准一次。",
                "STAGED_VERIFIED": "ChangePlan已批准並完成synthetic staging verify。",
                "APPLY_FAILED_ROLLED_BACK": "ChangePlan已批准，但staging apply失敗並已rollback。",
                "REJECTED": "ChangePlan已拒絕。",
                "STALE_OR_INVALID": "決定未接受：已過期、已用或無效。",
            }
            query_id, outcome = request["callback_query_id"], request["outcome"]
            if not isinstance(query_id, str) or COUNCIL_CALLBACK_QUERY_ID.fullmatch(query_id) is None or outcome not in messages: raise GatewayError("Council callback answer值無效")
            self.api.call("answerCallbackQuery", {"callback_query_id": query_id, "text": messages[outcome], "show_alert": outcome in {"STALE_OR_INVALID", "APPLY_FAILED_ROLLED_BACK"}})
            reply_id, status = 0, "ANSWERED"
        else:
            raise GatewayError("Council outbound operation無效")
        return {"schema": "ARC_GATEWAY_DELIVERY_ACK_V1", "request_id": request["request_id"], "status": status, "message_id": reply_id}

    def _run(self) -> None:
        assert self.listener is not None
        while not self.stop_event.is_set():
            try:
                connection, _address = self.listener.accept()
            except socket.timeout:
                continue
            except OSError:
                if self.stop_event.is_set():
                    break
                continue
            with connection:
                connection.settimeout(2.0)
                try:
                    if council_peer_uid(connection) != self.expected_peer_uid:
                        raise GatewayError("Council controller peer不符")
                    request = council_receive_frame(connection)
                    response = self._dispatch(request)
                except (GatewayError, TelegramError, OSError, TimeoutError):
                    request_id = "0" * 32
                    try:
                        candidate = request.get("request_id")  # type: ignore[possibly-undefined]
                        if isinstance(candidate, str) and COUNCIL_HEX32.fullmatch(candidate):
                            request_id = candidate
                    except Exception:
                        pass
                    response = {
                        "schema": "ARC_GATEWAY_DELIVERY_ACK_V1",
                        "request_id": request_id,
                        "status": "REJECTED",
                        "message_id": 0,
                    }
                try:
                    council_send_frame(connection, response)
                except (GatewayError, OSError):
                    pass

    def close(self) -> None:
        self.stop_event.set()
        if self.listener is not None:
            self.listener.close()
        if self.thread is not None:
            self.thread.join(timeout=2.0)
        try:
            information = self.socket_path.lstat()
        except FileNotFoundError:
            return
        identity = (information.st_dev, information.st_ino)
        if (
            self.listener_identity == identity
            and stat.S_ISSOCK(information.st_mode)
            and information.st_uid == os.geteuid()
        ):
            self.socket_path.unlink()


def legacy_command_response(
    command: str, argument: str, event_id: str, expected_username: str,
    uptime_seconds: int, next_update_id: int,
) -> tuple[str, str] | None:
    """Return the frozen transport-only response for the four legacy commands."""
    if command == "ping":
        return (
            f"pong ✅\nevent_id: {event_id}\ntransport: healthy\nllm: not configured",
            "sent_ping",
        )
    if command == "status":
        return (
            "✅ Gateway status\n"
            f"bot: @{expected_username}\n"
            "mode: telegram_transport_mvp_no_llm\n"
            f"uptime_seconds: {uptime_seconds}\n"
            f"next_update_id: {next_update_id}\n"
            "llm_agents: 0\n"
            "message_content_persisted: false",
            "sent_status",
        )
    if command == "help":
        return (
            "AI Research Council Gateway MVP\n"
            "/ping — 測試安全收發\n"
            "/status — 查看transport狀態\n"
            "/ask 內容 — 建立測試event（目前未接LLM）\n"
            "/help — 呢份說明",
            "sent_help",
        )
    if command == "ask":
        if not argument:
            return "用法：/ask 你的測試問題\n目前MVP未接LLM。", "ask_missing_argument"
        return (
            "✅ Gateway已收到測試event\n"
            f"event_id: {event_id}\n"
            f"input_chars: {len(argument)}\n"
            "身份、chat allowlist及去重路徑正常。\n"
            "目前未接LLM，所以唔會假稱呢個係智能答案。",
            "ask_transport_confirmed",
        )
    return None


def write_health(path: Path, **details: Any) -> None:
    atomic_json(path, {"version": 1, **details})


def run_command(args: argparse.Namespace) -> int:
    config = load_config(args.config)
    token = read_secure_token(args.token_file)
    api = TelegramAPI(token, args.api_base)
    lock_handle = acquire_lock(args.lock_file)
    connection: sqlite3.Connection | None = None
    council_outbound: CouncilOutboundServer | None = None
    stop_requested = False

    def request_stop(_signum: int, _frame: object) -> None:
        nonlocal stop_requested
        stop_requested = True

    signal.signal(signal.SIGTERM, request_stop)
    signal.signal(signal.SIGINT, request_stop)

    try:
        bot = verify_bot(api, config["expected_bot_username"])
        connection = open_database(
            args.database,
            config["initial_update_offset"],
            config_fingerprint(config),
        )
        next_update_id = state_int(connection, "next_update_id")
        started_monotonic = time.monotonic()
        owner_command_times: deque[float] = deque()
        failures = 0
        try:
            council_outbound = CouncilOutboundServer(api, config, approvals_enabled=args.enable_council_approvals)
            council_outbound.start()
            emit("council_adapter_ready", mode="A3_SYNTHETIC_APPROVAL" if args.enable_council_approvals else "A2_SYNTHETIC_DISCUSSION_ONLY")
        except (GatewayError, OSError) as error:
            council_outbound = None
            # Council availability must never take down the four legacy commands.
            emit("council_adapter_unavailable", error_type=type(error).__name__)
        write_health(
            args.health_file,
            status="running",
            mode="telegram_transport_mvp_no_llm",
            bot_username=str(bot.get("username", "")),
            allowed_chat_id=config["allowed_chat_id"],
            owner_user_id=config["owner_user_id"],
            allowed_thread_id=config.get("allowed_thread_id"),
            next_update_id=next_update_id,
        )
        emit(
            "gateway_started",
            mode="telegram_transport_mvp_no_llm",
            bot_username=str(bot.get("username", "")),
            allowed_chat_id=config["allowed_chat_id"],
            owner_user_id=config["owner_user_id"],
            allowed_thread_id=config.get("allowed_thread_id"),
            next_update_id=next_update_id,
        )

        announced = connection.execute("SELECT value FROM gateway_state WHERE key='announced_v1'").fetchone()
        if announced is None:
            try:
                announcement_id = send_text(
                    api,
                    config["allowed_chat_id"],
                    "✅ AI Research Council Telegram gateway MVP已上線。\n"
                    "目前只驗證安全收發、身份allowlist、去重及metadata audit；未接駁LLM，唔會假稱智能回答。\n"
                    "可用：/ping、/status、/ask 測試內容、/help",
                    config.get("allowed_thread_id"),
                )
                set_state(connection, "announced_v1", announcement_id)
                connection.commit()
            except TelegramError as error:
                emit("announcement_failed", error_type=type(error).__name__)

        while not stop_requested:
            try:
                updates = api.call(
                    "getUpdates",
                    {
                        "offset": next_update_id,
                        "timeout": config["poll_timeout_seconds"],
                        "allowed_updates": ["message", "callback_query"] if args.enable_council_approvals else ["message"],
                    },
                    timeout=config["poll_timeout_seconds"] + 15,
                )
                if not isinstance(updates, list):
                    raise TelegramError("getUpdates result唔係list")
                for update in updates:
                    if not isinstance(update, dict) or not isinstance(update.get("update_id"), int):
                        continue
                    update_id = int(update["update_id"])
                    if update_id < next_update_id:
                        continue
                    callback = extract_owner_changeplan_callback(update, config, args.enable_council_approvals)
                    if callback is not None:
                        reserved = reserve_update(
                            connection, update_id, update_id + 1, callback["chat_id"],
                            callback["owner_user_id"], callback["message_id"], callback["thread_id"],
                            "council_callback", 0, "reserved",
                        )
                        next_update_id = update_id + 1
                        if not reserved:
                            continue
                        outcome = "STALE_OR_INVALID"
                        status = "council_callback_rejected"
                        try:
                            decision = forward_council_callback(build_gateway_callback_frame(callback))
                            if decision["decision"] == "REJECT":
                                outcome = "REJECTED"
                            elif decision["status"] in {"STAGED_VERIFIED", "APPLY_FAILED_ROLLED_BACK"}:
                                outcome = decision["status"]
                            else:
                                outcome = "APPROVED"
                            status = "council_callback_" + outcome.casefold()
                        except GatewayError:
                            pass
                        try:
                            answer_text = {
                                "APPROVED": "ChangePlan已批准一次。",
                                "STAGED_VERIFIED": "ChangePlan已批准並完成synthetic staging verify。",
                                "APPLY_FAILED_ROLLED_BACK": "ChangePlan已批准，但staging apply失敗並已rollback。",
                                "REJECTED": "ChangePlan已拒絕。",
                                "STALE_OR_INVALID": "決定未接受：已過期、已用或無效。",
                            }[outcome]
                            answer = {"callback_query_id": callback["callback_query_id"],
                                      "text": answer_text,
                                      "show_alert": outcome in {"STALE_OR_INVALID", "APPLY_FAILED_ROLLED_BACK"}}
                            api.call("answerCallbackQuery", answer)
                            finish_update(connection, update_id, status)
                        except TelegramError as error:
                            finish_update(connection, update_id, "send_failed", error_type=type(error).__name__)
                        continue
                    message = update.get("message")
                    chat = message.get("chat") if isinstance(message, dict) else None
                    sender = message.get("from") if isinstance(message, dict) else None
                    chat_id = chat.get("id") if isinstance(chat, dict) and isinstance(chat.get("id"), int) else None
                    owner_user_id = sender.get("id") if isinstance(sender, dict) and isinstance(sender.get("id"), int) else None
                    message_id = message.get("message_id") if isinstance(message, dict) and isinstance(message.get("message_id"), int) else None
                    thread_id = message.get("message_thread_id") if isinstance(message, dict) and isinstance(message.get("message_thread_id"), int) else None
                    text = message.get("text") if isinstance(message, dict) and isinstance(message.get("text"), str) else ""
                    command, argument = parse_command(text, config["expected_bot_username"])
                    plain_eligible = command is None and bool(text.strip()) and not text.strip().startswith("/")
                    audit_command = command if command in ("ping", "status", "ask", "help", "council") else ("council_plain" if plain_eligible else "ignored")
                    authorised = (
                        isinstance(message, dict)
                        and isinstance(chat, dict)
                        and isinstance(sender, dict)
                        and sender.get("is_bot") is not True
                        and chat_id == config["allowed_chat_id"]
                        and owner_user_id == config["owner_user_id"]
                        and (config.get("allowed_thread_id") is None or thread_id == config.get("allowed_thread_id"))
                    )
                    # R05.7.1 plain-text capture: owner non-command text in the allowed
                    # chat becomes a COMMONS SYNTHETIC_PUBLIC_ONLY council event.
                    plain_text = text.strip() if (authorised and plain_eligible) else ""
                    initial_status = "reserved" if authorised and (command is not None or plain_text) else "ignored"
                    reserved = reserve_update(
                        connection,
                        update_id,
                        update_id + 1,
                        chat_id,
                        owner_user_id,
                        message_id,
                        thread_id,
                        audit_command,
                        len(text),
                        initial_status,
                    )
                    next_update_id = update_id + 1
                    if not reserved or not authorised or (command is None and not plain_text):
                        continue
                    now = time.monotonic()
                    while owner_command_times and now - owner_command_times[0] >= 60:
                        owner_command_times.popleft()
                    if len(owner_command_times) >= config["owner_commands_per_minute"]:
                        response_text = "⚠️ Owner command速率已達MVP上限，請稍後再試。"
                        status = "rate_limited"
                    elif len(text) > config["max_command_chars"]:
                        response_text = f"⚠️ 指令太長；MVP上限係{config['max_command_chars']}字。"
                        status = "input_too_long"
                    else:
                        owner_command_times.append(now)
                        event_id = f"tg:{chat_id}:{message_id}"
                        legacy = legacy_command_response(
                            command, argument, event_id, config["expected_bot_username"],
                            int(time.monotonic() - started_monotonic), next_update_id,
                        )
                        if legacy is not None:
                            response_text, status = legacy
                        elif command == "council" or plain_text:
                            try:
                                if not all(isinstance(value, int) for value in (
                                    update_id, chat_id, owner_user_id, message_id,
                                )):
                                    raise GatewayError("Telegram routing metadata無效")
                                if plain_text:
                                    room_id = "COMMONS"
                                    council_text = _safe_council_text(plain_text)
                                else:
                                    room_id, council_text = parse_council_argument(argument)
                                frame = build_gateway_council_frame(
                                    update_id, chat_id, owner_user_id, message_id,
                                    thread_id, room_id, council_text,
                                )
                                receipt = forward_council_message(frame)
                                if plain_text:
                                    response_text = f"📥 {room_id} · {receipt['status']}"
                                else:
                                    response_text = (
                                        "✅ Council synthetic event已接收\n"
                                        f"room: {room_id}\n"
                                        f"status: {receipt['status']}\n"
                                        f"event_sha256: {receipt['event_sha256']}\n"
                                        "只限PROVISIONAL_PILOT；冇host action。"
                                    )
                                status = "council_accepted" if receipt["status"] == "ACCEPTED" else "council_duplicate"
                            except GatewayError:
                                if plain_text:
                                    response_text = "⚠️ plain message未建立event（Council暫停，或訊息超出2000字/8000 bytes/控制字元限制）。"
                                else:
                                    response_text = (
                                        "⚠️ Council暫停或格式無效；event未執行任何host action。\n"
                                        "格式：/council [ARC/COMMONS/v1] SYNTHETIC_PUBLIC_ONLY 內容"
                                    )
                                status = "council_unavailable_or_rejected"
                        else:
                            response_text = "未知指令。請用 /help。"
                            status = "unknown_command"
                    try:
                        reply_id = send_text(api, chat_id, response_text, thread_id, message_id)
                        finish_update(connection, update_id, status, reply_id)
                        emit("command_handled", update_id=update_id, command=audit_command, status=status, reply_message_id=reply_id)
                    except TelegramError as error:
                        finish_update(connection, update_id, "send_failed", error_type=type(error).__name__)
                        emit("command_send_failed", update_id=update_id, command=audit_command, error_type=type(error).__name__)
                failures = 0
                write_health(
                    args.health_file,
                    status="running",
                    mode="telegram_transport_mvp_no_llm",
                    bot_username=config["expected_bot_username"],
                    allowed_chat_id=config["allowed_chat_id"],
                    owner_user_id=config["owner_user_id"],
                    allowed_thread_id=config.get("allowed_thread_id"),
                    next_update_id=next_update_id,
                )
            except TelegramError as error:
                failures += 1
                delay = min(60, 2 ** min(failures, 6))
                emit("poll_error", failures=failures, retry_seconds=delay, error_type=type(error).__name__)
                time.sleep(delay)
            except sqlite3.Error as error:
                emit("fatal_error", stage="database", error_type=type(error).__name__)
                return 1
        emit("gateway_stopped", next_update_id=next_update_id)
        write_health(
            args.health_file,
            status="stopped",
            mode="telegram_transport_mvp_no_llm",
            next_update_id=next_update_id,
        )
        return 0
    finally:
        if council_outbound is not None:
            council_outbound.close()
        if connection is not None:
            connection.close()
        lock_handle.close()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="AI Research Council Telegram transport gateway MVP")
    subparsers = parser.add_subparsers(dest="command", required=True)

    probe = subparsers.add_parser("probe", help="安全驗證bot並列出group／owner候選")
    probe.add_argument("--token-file", type=Path, required=True)
    probe.add_argument("--expected-username", required=True)
    probe.add_argument("--output", type=Path, required=True)
    probe.add_argument("--wait-seconds", type=int, default=3)
    probe.add_argument("--api-base", default="https://api.telegram.org")
    probe.set_defaults(handler=probe_command)

    run = subparsers.add_parser("run", help="運行long-polling transport gateway")
    run.add_argument("--config", type=Path, required=True)
    run.add_argument("--token-file", type=Path, required=True)
    run.add_argument("--database", type=Path, required=True)
    run.add_argument("--lock-file", type=Path, required=True)
    run.add_argument("--health-file", type=Path, required=True)
    run.add_argument("--api-base", default="https://api.telegram.org")
    run.add_argument("--enable-council-approvals", action="store_true", help="explicitly enable isolated A3 synthetic ChangePlan approvals")
    run.set_defaults(handler=run_command)
    return parser


def main() -> int:
    os.umask(0o077)
    parser = build_parser()
    args = parser.parse_args()
    if getattr(args, "wait_seconds", 0) < 0 or getattr(args, "wait_seconds", 0) > 30:
        parser.error("--wait-seconds必須介乎0至30")
    try:
        return int(args.handler(args))
    except GatewayError as error:
        emit("fatal_error", error=str(error))
        return 1
    except KeyboardInterrupt:
        emit("interrupted")
        return 130
    except Exception as error:
        emit("unexpected_fatal_error", error_type=type(error).__name__)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())