openclaw/scripts/e2e/telegram-user-driver.py

#!/usr/bin/env -S uv run --script

import argparse
import atexit
import base64
import ctypes
import json
import os
import secrets
import shutil
import signal
import socket
import stat
import subprocess
import sys
import time
import urllib.request
from pathlib import Path

SKILL_DIR = Path(__file__).resolve().parents[1]
STATE_DIR = Path(os.environ.get("TELEGRAM_USER_DRIVER_STATE_DIR") or (SKILL_DIR / "user-driver")).expanduser()
CONFIG_PATH = STATE_DIR / "config.local.json"
BOT_CREDENTIALS_PATH = SKILL_DIR / "credentials.local.json"


class DriverError(RuntimeError):
    pass


def read_json(path):
    try:
        return json.loads(path.read_text())
    except FileNotFoundError:
        return {}


def open_contained_file(root, relative):
    """Opens a file under root, descending one component at a time from an open directory
    descriptor so no symlink can be traversed and no directory can be swapped mid-walk.
    Anchoring on descriptors is what keeps containment true without Linux-only /proc paths."""
    parent = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
    try:
        *directories, name = relative.parts
        for directory in directories:
            child = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
            os.close(parent)
            parent = child
        return os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=parent)
    finally:
        os.close(parent)


def write_json_private(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.parent.chmod(stat.S_IRWXU)
    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
    path.chmod(stat.S_IRUSR | stat.S_IWUSR)


def env_or_config(env_name, config, key, default=""):
    value = os.environ.get(env_name)
    if value:
        return value
    value = config.get(key)
    if value is None:
        return default
    return str(value)


def load_config():
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    STATE_DIR.chmod(stat.S_IRWXU)
    config = read_json(CONFIG_PATH)
    bot_config = read_json(BOT_CREDENTIALS_PATH)
    if not valid_base64(config.get("databaseEncryptionKey", "")):
        config["databaseEncryptionKey"] = base64.b64encode(secrets.token_bytes(32)).decode()
        write_json_private(CONFIG_PATH, config)
    return config, bot_config


def valid_base64(value):
    if not isinstance(value, str) or not value:
        return False
    try:
        base64.b64decode(value.encode(), validate=True)
        return True
    except ValueError:
        return False


def find_tdjson(config):
    explicit = env_or_config("TELEGRAM_USER_DRIVER_TDLIB_PATH", config, "tdlibPath")
    candidates = []
    if explicit:
        candidates.append(Path(explicit).expanduser())
    candidates.extend(
        Path(path)
        for path in [
            "/opt/homebrew/lib/libtdjson.dylib",
            "/opt/homebrew/opt/tdlib/lib/libtdjson.dylib",
            "/usr/local/lib/libtdjson.dylib",
            "/usr/local/opt/tdlib/lib/libtdjson.dylib",
            "/usr/lib/libtdjson.so",
            "/usr/local/lib/libtdjson.so",
        ]
    )
    for path in candidates:
        if path.exists():
            return path
    found = shutil.which("tdjson")
    if found:
        return Path(found)
    return None


def telegram_bot(token, method, payload=None):
    data = json.dumps(payload or {}).encode()
    request = urllib.request.Request(
        f"https://api.telegram.org/bot{token}/{method}",
        data=data,
        headers={"content-type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=15) as response:
        body = json.loads(response.read().decode())
    if not body.get("ok"):
        raise DriverError(body.get("description") or f"{method} failed")
    return body["result"]


def resolve_sut(config, bot_config):
    username = env_or_config("TELEGRAM_USER_DRIVER_SUT_USERNAME", config, "sutUsername")
    user_id = env_or_config("TELEGRAM_USER_DRIVER_SUT_ID", config, "sutId")
    if username and user_id:
        return {"username": username.lstrip("@"), "id": int(user_id)}
    token = (
        os.environ.get("TELEGRAM_E2E_SUT_BOT_TOKEN")
        or os.environ.get("OPENCLAW_QA_TELEGRAM_SUT_BOT_TOKEN")
        or bot_config.get("sutBotToken")
        or bot_config.get("botAToken")
        or bot_config.get("BOTA")
    )
    if token:
        me = telegram_bot(token, "getMe")
        return {"username": me.get("username", "").lstrip("@"), "id": int(me["id"])}
    if username:
        return {"username": username.lstrip("@"), "id": None}
    return {"username": "", "id": None}


def default_chat(config, bot_config):
    return (
        os.environ.get("TELEGRAM_USER_DRIVER_CHAT_ID")
        or os.environ.get("TELEGRAM_E2E_GROUP_ID")
        or os.environ.get("OPENCLAW_QA_TELEGRAM_GROUP_ID")
        or str(config.get("defaultChatId") or "")
        or str(bot_config.get("groupId") or "")
    )


class TdClient:
    def __init__(self, config):
        lib_path = find_tdjson(config)
        if not lib_path:
            raise DriverError("Missing libtdjson. Install with: brew install tdlib")
        self.lib_path = lib_path
        self.lib = ctypes.CDLL(str(lib_path))
        self.lib.td_execute.argtypes = [ctypes.c_char_p]
        self.lib.td_execute.restype = ctypes.c_char_p
        self.lib.td_json_client_create.restype = ctypes.c_void_p
        self.lib.td_json_client_send.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
        self.lib.td_json_client_receive.argtypes = [ctypes.c_void_p, ctypes.c_double]
        self.lib.td_json_client_receive.restype = ctypes.c_char_p
        self.lib.td_json_client_execute.argtypes = [ctypes.c_void_p, ctypes.c_char_p]
        self.lib.td_json_client_execute.restype = ctypes.c_char_p
        self.lib.td_json_client_destroy.argtypes = [ctypes.c_void_p]
        self.lib.td_execute(json.dumps({"@type": "setLogStream", "log_stream": {"@type": "logStreamEmpty"}}).encode())
        self.lib.td_execute(json.dumps({"@type": "setLogVerbosityLevel", "new_verbosity_level": 0}).encode())
        self.client = self.lib.td_json_client_create()
        self.extra = 0
        self.pending = {}
        self.users = {}
        self.updates = []
        atexit.register(self.destroy)

    def execute(self, payload):
        raw = self.lib.td_json_client_execute(self.client, json.dumps(payload).encode())
        return json.loads(raw.decode()) if raw else None

    def send(self, payload):
        self.extra += 1
        extra = str(self.extra)
        payload = dict(payload)
        payload["@extra"] = extra
        self.pending[extra] = payload["@type"]
        self.lib.td_json_client_send(self.client, json.dumps(payload).encode())
        return extra

    def receive(self, timeout=1.0):
        raw = self.lib.td_json_client_receive(self.client, ctypes.c_double(timeout))
        if not raw:
            return None
        item = json.loads(raw.decode())
        if item.get("@type") == "updateUser":
            user = item.get("user") or {}
            self.users[int(user["id"])] = user
        return item

    def destroy(self):
        if not self.client:
            return
        client = self.client
        self.client = None
        self.lib.td_json_client_destroy(client)

    def request(self, payload, timeout=20):
        extra = self.send(payload)
        deadline = time.time() + timeout
        while time.time() < deadline:
            item = self.receive(1)
            if not item:
                continue
            if item.get("@extra") == extra:
                if item.get("@type") == "error":
                    raise DriverError(f"{payload['@type']} failed: {item.get('message')}")
                return item
            self.handle_update(item)
        raise DriverError(f"Timed out waiting for {payload['@type']}")

    def handle_update(self, item):
        self.updates.append(item)
        return item

    def next_update(self, timeout=1.0):
        if self.updates:
            return self.updates.pop(0)
        return self.receive(timeout)


class UserDriver:
    def __init__(self, config, bot_config):
        self.config = config
        self.bot_config = bot_config
        self.client = TdClient(config)
        self.auth_state = None
        self.printed_qr_link = ""

    def td_params(self):
        api_id = env_or_config("TELEGRAM_USER_DRIVER_API_ID", self.config, "apiId")
        api_hash = env_or_config("TELEGRAM_USER_DRIVER_API_HASH", self.config, "apiHash")
        if not api_id or not api_hash:
            raise DriverError("Missing Telegram API app credentials. Run configure or set TELEGRAM_USER_DRIVER_API_ID/API_HASH.")
        params = {
            "use_test_dc": False,
            "database_directory": str(STATE_DIR / "db"),
            "files_directory": str(STATE_DIR / "files"),
            "use_file_database": True,
            "use_chat_info_database": True,
            "use_message_database": True,
            "use_secret_chats": False,
            "api_id": int(api_id),
            "api_hash": api_hash,
            "system_language_code": "en",
            "device_model": "OpenClaw Telegram User Driver",
            "system_version": sys.platform,
            "application_version": "1",
            "enable_storage_optimizer": True,
            "ignore_file_names": False,
        }
        return {
            "@type": "setTdlibParameters",
            "parameters": {"@type": "tdlibParameters", **params},
        }

    def td_params_current(self):
        payload = self.td_params()
        params = dict(payload["parameters"])
        params.pop("@type", None)
        return {"@type": "setTdlibParameters", **params}

    def encryption_key(self):
        return env_or_config(
            "TELEGRAM_USER_DRIVER_DB_ENCRYPTION_KEY",
            self.config,
            "databaseEncryptionKey",
        )

    def encryption_key_for_current_tdlib(self):
        return base64.b64encode(self.encryption_key().encode()).decode()

    def authorize(self, args=None, need_ready=True):
        args = args or argparse.Namespace()
        self.client.execute({"@type": "setLogStream", "log_stream": {"@type": "logStreamEmpty"}})
        self.client.execute({"@type": "setLogVerbosityLevel", "new_verbosity_level": 1})
        self.client.send({"@type": "getOption", "name": "version"})
        retried_current_params = False
        retried_current_encryption_key = False
        deadline = time.time() + getattr(args, "timeout_ms", 120000) / 1000
        while time.time() < deadline:
            item = self.client.receive(1)
            if not item:
                continue
            if item.get("@type") == "updateAuthorizationState":
                self.auth_state = item["authorization_state"]
                state = self.auth_state["@type"]
                if state == "authorizationStateWaitTdlibParameters":
                    self.client.send(self.td_params())
                elif state == "authorizationStateWaitEncryptionKey":
                    self.client.send(
                        {
                            "@type": "checkDatabaseEncryptionKey",
                            "encryption_key": self.encryption_key(),
                        }
                    )
                elif state == "authorizationStateWaitPhoneNumber":
                    if getattr(args, "phone", ""):
                        self.client.send(
                            {
                                "@type": "setAuthenticationPhoneNumber",
                                "phone_number": args.phone,
                                "settings": None,
                            }
                        )
                    elif getattr(args, "qr", False):
                        self.client.send(
                            {
                                "@type": "requestQrCodeAuthentication",
                                "other_user_ids": [],
                            }
                        )
                    elif need_ready:
                        raise DriverError("Not logged in. Run: user-driver.py login --qr")
                    else:
                        return False
                elif state == "authorizationStateWaitOtherDeviceConfirmation":
                    self.show_qr_link(self.auth_state["link"])
                elif state == "authorizationStateWaitCode":
                    code = getattr(args, "code", "") or prompt_secret("Telegram login code: ")
                    self.client.send({"@type": "checkAuthenticationCode", "code": code})
                elif state == "authorizationStateWaitPassword":
                    password = getattr(args, "password", "") or prompt_secret("Telegram 2FA password: ")
                    self.client.send({"@type": "checkAuthenticationPassword", "password": password})
                elif state == "authorizationStateReady":
                    return True
                elif state in {
                    "authorizationStateClosing",
                    "authorizationStateClosed",
                    "authorizationStateLoggingOut",
                }:
                    raise DriverError(f"TDLib auth state is {state}")
            elif item.get("@type") == "error":
                message = item.get("message") or "TDLib error"
                if not retried_current_params and "Valid api_id must be provided" in message:
                    retried_current_params = True
                    self.client.send(
                        {
                            "@type": "setDatabaseEncryptionKey",
                            "new_encryption_key": self.encryption_key_for_current_tdlib(),
                        }
                    )
                    self.client.send(self.td_params_current())
                    continue
                if not retried_current_encryption_key and "Wrong database encryption key" in message:
                    retried_current_encryption_key = True
                    self.client.send(
                        {
                            "@type": "setDatabaseEncryptionKey",
                            "new_encryption_key": self.encryption_key_for_current_tdlib(),
                        }
                    )
                    self.client.send(self.td_params_current())
                    continue
                if retried_current_params and "Initialization parameters are needed" in message:
                    continue
                raise DriverError(message)
        raise DriverError("Timed out waiting for Telegram authorization")

    def show_qr_link(self, link):
        if link == self.printed_qr_link:
            return
        self.printed_qr_link = link
        print("\nScan this with Telegram > Settings > Devices > Link Desktop Device\n")
        qrencode = shutil.which("qrencode")
        if qrencode:
            subprocess.run([qrencode, "-t", "UTF8", link], check=False)
        print(link)
        print()

    def resolve_chat(self, chat):
        chat = chat or default_chat(self.config, self.bot_config)
        if not chat:
            raise DriverError(
                "Missing chat. Pass --chat or configure defaultChatId. Run `user-driver.py chats --json` to list chats visible to the tester account."
            )
        if chat.startswith(("https://t.me/+", "tg://join")) or "joinchat" in chat:
            return self.client.request({"@type": "joinChatByInviteLink", "invite_link": chat})["id"]
        if chat.startswith("@"):
            return self.client.request({"@type": "searchPublicChat", "username": chat[1:]})["id"]
        if chat.startswith("https://t.me/") and "/" not in chat.removeprefix("https://t.me/"):
            return self.client.request(
                {
                    "@type": "searchPublicChat",
                    "username": chat.removeprefix("https://t.me/"),
                }
            )["id"]
        try:
            return self.client.request({"@type": "getChat", "chat_id": int(chat)}, timeout=10)["id"]
        except DriverError as error:
            raise DriverError(
                f"Chat not found for tester account: {chat}. Add the QA user to the group, or configure the TDLib chat id from `user-driver.py chats --json`."
            ) from error

    def text_content(self, text):
        return {
            "@type": "inputMessageText",
            "text": {"@type": "formattedText", "text": text, "entities": []},
            "disable_web_page_preview": False,
            "clear_draft": True,
        }

    def document_content(self, file_path, caption):
        return {
            "@type": "inputMessageDocument",
            "document": {"@type": "inputFileLocal", "path": file_path},
            "thumbnail": None,
            "disable_content_type_detection": False,
            "caption": {
                "@type": "formattedText",
                "text": caption or "",
                "entities": [],
            },
        }

    def send_text(self, chat_id, text, reply_to=None, thread_id=0, file_path=None):
        content = self.document_content(file_path, text) if file_path else self.text_content(text)
        return self.settle_sent_message(
            self.client.request(
                {
                    "@type": "sendMessage",
                    "chat_id": chat_id,
                    "message_thread_id": int(thread_id or 0),
                    "reply_to_message_id": int(reply_to or 0),
                    "options": {
                        "@type": "messageSendOptions",
                        "disable_notification": True,
                        "from_background": False,
                        "scheduling_state": None,
                    },
                    "reply_markup": None,
                    "input_message_content": content,
                },
                timeout=60,
            )
        )

    def settle_sent_message(self, message, timeout=30):
        if not message.get("sending_state"):
            return message
        pending_id = message["id"]
        deadline = time.time() + timeout
        deferred = []
        while time.time() < deadline:
            item = self.client.next_update(1)
            if not item:
                continue
            if item.get("@type") == "updateMessageSendSucceeded" and item.get("old_message_id") == pending_id:
                self.client.updates = deferred + self.client.updates
                return item["message"]
            if item.get("@type") == "updateMessageSendFailed" and item.get("old_message_id") == pending_id:
                self.client.updates = deferred + self.client.updates
                raise DriverError(item.get("error_message") or "Telegram message send failed")
            deferred.append(item)
        self.client.updates = deferred + self.client.updates
        raise DriverError("Timed out waiting for Telegram message send confirmation")

    def wait_for_message(self, chat_id, args, after_message_id=0):
        sut = resolve_sut(self.config, self.bot_config)
        expect = args.expect or []
        from_bot = (args.from_bot or sut["username"] or "").lstrip("@")
        from_bot_id = int(from_bot) if from_bot.isdigit() else sut["id"]
        from_bot_username = "" if from_bot.isdigit() else from_bot
        deadline = time.time() + args.timeout_ms / 1000
        observed = []
        while time.time() < deadline:
            item = self.client.next_update(1)
            if not item or item.get("@type") != "updateNewMessage":
                continue
            message = item["message"]
            if int(message.get("chat_id", 0)) != int(chat_id):
                continue
            normalized = normalize_message(message, self.client.users)
            observed.append(normalized)
            if int(normalized["messageId"]) <= int(after_message_id):
                continue
            if getattr(args, "thread_id", None) and str(normalized.get("threadId") or "") != str(args.thread_id):
                continue
            if from_bot_id and int(normalized.get("senderId") or 0) != int(from_bot_id):
                continue
            if not from_bot_id and from_bot_username and normalized["senderUsername"].lower() != from_bot_username.lower():
                continue
            if args.reply_to and str(normalized.get("replyToMessageId") or "") != str(args.reply_to):
                continue
            if any(entry not in normalized["text"] for entry in expect):
                continue
            return normalized, observed
        return None, observed


def prompt_secret(label):
    if not sys.stdin.isatty():
        raise DriverError(f"{label.strip()} required; rerun with explicit flag in an interactive terminal.")
    import getpass

    return getpass.getpass(label)


def normalize_message(message, users=None):
    users = users or {}
    sender = message.get("sender_id") or {}
    sender_id = sender.get("user_id") or sender.get("chat_id")
    sender_user = users.get(int(sender_id or 0), {}) if sender_id else {}
    content = message.get("content") or {}
    text = ""
    if content.get("@type") == "messageText":
        text = (content.get("text") or {}).get("text", "")
    elif "caption" in content:
        text = (content.get("caption") or {}).get("text", "")
    reply_to_message_id = message.get("reply_to_message_id") or (message.get("reply_to") or {}).get("message_id")
    message_id = message.get("id")
    return {
        "messageId": message_id,
        # TDLib reserves the low 20 bits; Telegram private-post links use the
        # server/Bot API id. Keep both identities at the boundary that owns them.
        "botApiMessageId": (int(message_id) >> 20) if message_id else None,
        "chatId": message.get("chat_id"),
        "senderId": sender_id,
        "senderUsername": sender_user.get("username") or "",
        "date": message.get("date"),
        "replyToMessageId": reply_to_message_id,
        "replyToBotApiMessageId": (int(reply_to_message_id) >> 20) if reply_to_message_id else None,
        "threadId": message.get("message_thread_id"),
        "text": text,
        "contentType": content.get("@type"),
        "raw": message,
    }


def rich_text(value):
    if not isinstance(value, dict):
        return ""
    kind = value.get("@type", "")
    if kind == "richTextPlain":
        return value.get("text", "")
    if kind == "richTextCustomEmoji":
        return value.get("alternative_text", "")
    if kind == "richTextMathematicalExpression":
        return value.get("expression", "")
    if kind == "richTexts":
        return "".join(rich_text(item) for item in value.get("texts") or [])
    return rich_text(value.get("text"))


def rich_message_text(value):
    if not isinstance(value, dict):
        return ""
    parts = []

    def visit(node):
        if isinstance(node, list):
            for item in node:
                visit(item)
        elif isinstance(node, dict):
            if str(node.get("@type", "")).startswith("richText"):
                text = rich_text(node)
                if text:
                    parts.append(text)
            else:
                for child in node.values():
                    visit(child)

    visit(value.get("blocks") or [])
    return "\n".join(parts)


def content_text(content):
    for key in ("text", "caption"):
        value = content.get(key)
        if isinstance(value, dict) and isinstance(value.get("text"), str):
            return value["text"]
    if content.get("@type") == "messageRichMessage":
        return rich_message_text(content.get("message") or {})
    return ""


def server_message_id(tdlib_message_id):
    return str(int(tdlib_message_id) >> 20) if tdlib_message_id else None


class UserObserver:
    MAX_EVENTS = 500

    def __init__(self, user_driver, chat_id, sut_user_id, sut_username, journal_path, media_root):
        self.driver = user_driver
        self.client = user_driver.client
        self.chat_id = int(chat_id)
        self.sut_user_id = int(sut_user_id)
        self.sut = {"id": self.sut_user_id, "username": sut_username.lstrip("@")}
        self.media_root = Path(media_root).resolve()
        self.media_staging = Path(journal_path).parent / "media"
        self.media_staging.mkdir(mode=0o700)
        self.started_at = time.monotonic()
        self.events = []
        self.truncated = False
        self.message_fields = {}
        self.message_ids = {}
        self.recorded_message_ids = set()
        self.buttons = {}
        self.sent_message_ids = set()
        # One line-buffered handle keeps the live journal readable without reopening per event.
        self.journal = Path(journal_path).open("w", buffering=1)  # noqa: SIM115

    def close(self):
        self.journal.close()
        shutil.rmtree(self.media_staging)

    def actor(self, sender_id, is_outgoing=False):
        if is_outgoing:
            return "user"
        if sender_id == self.sut_user_id:
            return "bot"
        return "other"

    def append(self, kind, message_id=None, **fields):
        if len(self.events) >= self.MAX_EVENTS:
            self.truncated = True
            return None
        event = {
            "seq": len(self.events) + 1,
            "elapsedMs": int((time.monotonic() - self.started_at) * 1000),
            "kind": kind,
            "messageId": server_message_id(message_id),
            **fields,
        }
        self.events.append(event)
        self.journal.write(json.dumps(event, separators=(",", ":")) + "\n")
        return event

    def sender_id(self, message):
        sender = message.get("sender_id") or {}
        return sender.get("user_id") or sender.get("chat_id")

    def reply_fields(self, message):
        reply = message.get("reply_to") or {}
        return {"replyToMessageId": server_message_id(reply.get("message_id") or message.get("reply_to_message_id"))}

    def remember_buttons(self, message_id, markup):
        flattened = []
        public_buttons = []
        for row in markup.get("rows") or markup.get("inline_keyboard") or []:
            row_buttons = row.get("buttons") if isinstance(row, dict) else row
            for button in row_buttons or []:
                index = len(flattened)
                button_type = button.get("type") or {}
                flattened.append(button_type)
                public_buttons.append(
                    {
                        "index": index,
                        "text": button.get("text", ""),
                        "type": str(button_type.get("@type", "")).removeprefix("inlineKeyboardButtonType"),
                    }
                )
        if flattened:
            self.buttons[message_id] = flattened
        else:
            self.buttons.pop(message_id, None)
        return public_buttons

    def remember_message(self, message):
        message_id = message.get("id")
        if not isinstance(message_id, int):
            return {}
        sender_id = self.sender_id(message)
        fields = {
            "actor": self.actor(sender_id, bool(message.get("is_outgoing"))),
            "isOutgoing": bool(message.get("is_outgoing")),
            **self.reply_fields(message),
        }
        self.message_fields[message_id] = fields
        public_buttons = []
        if fields["actor"] != "other":
            self.message_ids[server_message_id(message_id)] = message_id
            if fields["isOutgoing"]:
                self.sent_message_ids.add(message_id)
            public_buttons = self.remember_buttons(message_id, message.get("reply_markup") or {})
        return {**fields, "buttons": public_buttons}

    def ingest(self, update):
        message = update.get("message") or {}
        update_chat_id = message.get("chat_id") if message else update.get("chat_id")
        if int(update_chat_id or 0) != self.chat_id:
            return
        kind = update.get("@type")
        if kind == "updateNewMessage":
            if message.get("sending_state"):
                return
            if message.get("id") in self.recorded_message_ids:
                return
            content = message.get("content") or {}
            self.recorded_message_ids.add(message.get("id"))
            fields = self.remember_message(message)
            if fields.get("actor") == "other":
                return
            self.append(
                "message",
                message.get("id"),
                **fields,
                contentType=content.get("@type", ""),
                text=content_text(content),
            )
        elif kind == "updateMessageContent":
            message_id = update.get("message_id")
            content = update.get("new_content") or {}
            fields = self.message_fields.get(message_id)
            if not fields or fields.get("actor") == "other":
                return
            self.append(
                "edit",
                message_id,
                **fields,
                contentType=content.get("@type", ""),
                text=content_text(content),
            )
        elif kind == "updateMessageEdited":
            message_id = update.get("message_id")
            fields = self.message_fields.get(message_id)
            if not fields or fields.get("actor") == "other":
                return
            buttons = self.remember_buttons(message_id, update.get("reply_markup") or {})
            self.append(
                "edit-meta",
                message_id,
                **fields,
                buttons=buttons,
                editDate=update.get("edit_date"),
            )
        elif kind == "updateDeleteMessages" and not update.get("from_cache"):
            for message_id in update.get("message_ids") or []:
                fields = self.message_fields.get(message_id)
                if not fields or fields.get("actor") == "other":
                    continue
                self.append(
                    "delete",
                    message_id,
                    **fields,
                    isPermanent=bool(update.get("is_permanent")),
                )
        elif kind == "updateChatAction":
            sender_id = self.sender_id({"sender_id": update.get("sender_id") or {}})
            actor = self.actor(sender_id)
            if actor == "other":
                return
            self.append(
                "typing",
                actor=actor,
                action=str((update.get("action") or {}).get("@type", "")).removeprefix("chatAction"),
            )

    def pump(self, seconds):
        deadline = time.monotonic() + seconds
        while time.monotonic() < deadline:
            update = self.client.next_update(timeout=min(0.2, max(0.0, deadline - time.monotonic())))
            if update:
                self.ingest(update)

    def resolve_message_id(self, value):
        message_id = self.message_ids.get(str(value))
        if message_id is None:
            raise DriverError(f"Message {value} was not observed in this session.")
        return message_id

    def resolve_media(self, value):
        relative = Path(value)
        if relative.is_absolute() or not relative.parts or ".." in relative.parts:
            raise DriverError("Media must be inside the Mantis output directory.")
        try:
            descriptor = open_contained_file(self.media_root, relative)
        except OSError as error:
            raise DriverError("Media must be a regular file inside the Mantis output directory.") from error
        try:
            if not stat.S_ISREG(os.fstat(descriptor).st_mode):
                raise DriverError("Media must be a regular file no larger than 20 MiB.")
            staging_dir = self.media_staging / f"upload-{secrets.token_hex(8)}"
            staging_dir.mkdir(mode=0o700)
            target = staging_dir / relative.name
            target_descriptor = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            try:
                copied = 0
                with os.fdopen(os.dup(descriptor), "rb") as source, os.fdopen(target_descriptor, "wb") as destination:
                    while chunk := source.read(1024 * 1024):
                        copied += len(chunk)
                        if copied > 20 * 1024 * 1024:
                            raise DriverError("Media must be a regular file no larger than 20 MiB.")
                        destination.write(chunk)
            except Exception:
                shutil.rmtree(staging_dir)
                raise
            return str(target)
        finally:
            os.close(descriptor)

    def call(self, request):
        command = request.get("command")
        if command == "ping":
            return {"ok": True, "cursor": len(self.events)}
        if command == "events":
            seconds = float(request.get("seconds") or 0)
            if seconds < 0 or seconds > 60:
                raise DriverError("Observation must be between 0 and 60 seconds.")
            self.pump(seconds)
            since = int(request.get("since") or 0)
            if since < 0 or since > len(self.events):
                raise DriverError("Observation cursor is outside this session's timeline.")
            return {
                "ok": True,
                "cursor": len(self.events),
                "events": self.events[since:],
            }
        if command == "send":
            text, _ = apply_template(str(request.get("text") or ""), self.sut)
            media = request.get("media")
            if not text and not media:
                raise DriverError("A message needs text or media.")
            if len(text) > 4000:
                raise DriverError("Message text exceeds 4000 characters.")
            reply_to = request.get("replyTo")
            reply_id = self.resolve_message_id(reply_to) if reply_to else None
            staged_media = self.resolve_media(str(media)) if media else None
            event_start = len(self.events)
            try:
                sent = self.driver.send_text(
                    self.chat_id,
                    text,
                    reply_to=reply_id,
                    file_path=staged_media,
                )
            finally:
                if staged_media:
                    shutil.rmtree(Path(staged_media).parent)
            fields = self.remember_message(sent)
            self.recorded_message_ids.add(sent.get("id"))
            sent_event = self.append(
                "message",
                sent.get("id"),
                **fields,
                contentType=(sent.get("content") or {}).get("@type", ""),
                text=content_text(sent.get("content") or {}),
            )
            self.pump(0.2)
            return {
                "ok": True,
                "cursor": len(self.events),
                "events": self.events[event_start:],
                "sent": sent_event,
            }
        if command == "delete":
            message_id = self.resolve_message_id(request.get("messageId"))
            if message_id not in self.sent_message_ids:
                raise DriverError("Only user messages sent in this session can be deleted.")
            self.client.request(
                {
                    "@type": "deleteMessages",
                    "chat_id": self.chat_id,
                    "message_ids": [message_id],
                    "revoke": True,
                }
            )
            self.pump(0.2)
            return {"ok": True, "cursor": len(self.events)}
        if command == "press":
            message_id = self.resolve_message_id(request.get("messageId"))
            index = int(request.get("button"))
            buttons = self.buttons.get(message_id) or []
            if index < 0 or index >= len(buttons):
                raise DriverError("Button index was not observed on that message.")
            button = buttons[index]
            if button.get("@type") != "inlineKeyboardButtonTypeCallback":
                raise DriverError("Only callback buttons can be pressed by this harness.")
            self.client.request(
                {
                    "@type": "getCallbackQueryAnswer",
                    "chat_id": self.chat_id,
                    "message_id": message_id,
                    "payload": {
                        "@type": "callbackQueryPayloadData",
                        "data": button.get("data", ""),
                    },
                }
            )
            self.pump(0.2)
            return {"ok": True, "cursor": len(self.events)}
        if command == "shutdown":
            self.pump(float(request.get("settleSeconds") or 0))
            return {"ok": True, "cursor": len(self.events), "shutdown": True}
        raise DriverError("Unknown observer command.")


def command_serve_session(args):
    config, bot_config = load_config()
    user_driver = UserDriver(config, bot_config)
    user_driver.authorize(need_ready=True)
    chat_id = user_driver.resolve_chat(args.chat)
    # Authorization can replay updates received while this QA account was offline.
    # The proof timeline begins only after this observer is ready for scenario actions.
    for _ in range(UserObserver.MAX_EVENTS):
        if not user_driver.client.next_update(timeout=0):
            break
    else:
        raise DriverError("Telegram observer startup backlog exceeded its event budget.")
    observer = UserObserver(
        user_driver,
        chat_id,
        args.sut_user_id,
        args.sut_username,
        args.journal,
        args.media_root,
    )
    socket_path = Path(args.socket)
    socket_path.unlink(missing_ok=True)
    server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    server.bind(str(socket_path))
    # The session root's default ACL grants mantis-sut access; group rw keeps
    # that named ACL entry effective when the runner tightens this socket.
    socket_path.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IWGRP)
    server.listen(4)
    server.settimeout(0.2)
    shutdown = False
    try:
        while not shutdown:
            update = observer.client.next_update(timeout=0)
            if update:
                observer.ingest(update)
                continue
            try:
                connection, _ = server.accept()
            except TimeoutError:
                continue
            with connection:
                payload = b""
                while b"\n" not in payload and len(payload) <= 65536:
                    chunk = connection.recv(65536 - len(payload) + 1)
                    if not chunk:
                        break
                    payload += chunk
                try:
                    request = json.loads(payload.split(b"\n", 1)[0])
                    response = observer.call(request)
                    shutdown = bool(response.pop("shutdown", False))
                    response["truncated"] = observer.truncated
                except (
                    DriverError,
                    ValueError,
                    TypeError,
                    json.JSONDecodeError,
                ) as error:
                    response = {"ok": False, "error": str(error)}
                connection.sendall(json.dumps(response, separators=(",", ":")).encode() + b"\n")
    finally:
        observer.close()
        server.close()
        socket_path.unlink(missing_ok=True)


def command_serve(args):
    pid_path = Path(args.pid_file)
    pid_fd = os.open(pid_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
    with os.fdopen(pid_fd, "w") as handle:
        json.dump({"pid": os.getpid(), "pgid": os.getpgrp(), "socket": args.socket}, handle)
        handle.write("\n")
    try:
        command_serve_session(args)
    finally:
        pid_path.unlink(missing_ok=True)


def running_process_command_line(pid):
    """Returns the argv pid is running under, or None once it is gone, so pid reuse cannot
    redirect cleanup at an unrelated process group. Zombies hold their table entry until
    they are reaped, so their pid cannot be reused and they count as already gone."""
    result = subprocess.run(
        ["ps", "-p", str(pid), "-o", "state=", "-o", "args="],
        capture_output=True,
        check=False,
        text=True,
    )
    state, _, command_line = result.stdout.strip().partition(" ")
    if result.returncode != 0 or not state or state.startswith("Z"):
        return None
    return command_line.strip()


def command_terminate_observer(args):
    pid_path = Path(args.pid_file)
    try:
        metadata = pid_path.lstat()
    except FileNotFoundError:
        return
    if (
        not stat.S_ISREG(metadata.st_mode)
        or stat.S_IMODE(metadata.st_mode) != 0o600
        or metadata.st_uid != os.getuid()
        or metadata.st_nlink != 1
    ):
        raise DriverError("Telegram observer pid file is not a private runner-owned file.")
    value = json.loads(pid_path.read_text())
    pid = int(value.get("pid") or 0)
    pgid = int(value.get("pgid") or 0)
    if value.get("socket") != args.socket or pid <= 0 or pgid <= 0 or pgid == os.getpgrp():
        raise DriverError("Telegram observer pid file is invalid.")
    command_line = running_process_command_line(pid)
    if command_line is None:
        pid_path.unlink(missing_ok=True)
        return
    if "telegram-user-driver" not in command_line or args.socket not in command_line or "serve" not in command_line:
        raise DriverError("Telegram observer process identity changed before cleanup.")
    try:
        os.killpg(pgid, signal.SIGTERM)
    except ProcessLookupError:
        pass
    deadline = time.monotonic() + 2
    while running_process_command_line(pid) is not None and time.monotonic() < deadline:
        time.sleep(0.05)
    if running_process_command_line(pid) is not None:
        try:
            os.killpg(pgid, signal.SIGKILL)
        except ProcessLookupError:
            pass
    pid_path.unlink(missing_ok=True)


def apply_template(text, sut):
    run = f"USER-E2E-{int(time.time())}"
    username = sut.get("username", "")
    return text.replace("{run}", run).replace("{sut}", username), run


def print_result(payload, as_json=False, output=""):
    if output:
        output_path = Path(output).expanduser()
        output_path.parent.mkdir(parents=True, exist_ok=True)
        output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
    if as_json:
        print(json.dumps(payload, indent=2, sort_keys=True))
    else:
        print(json.dumps(payload, indent=2, sort_keys=True))


def command_configure(args):
    config, _ = load_config()
    changed = False
    for arg_name, key in [
        ("api_id", "apiId"),
        ("api_hash", "apiHash"),
        ("tdlib_path", "tdlibPath"),
        ("chat", "defaultChatId"),
        ("sut_username", "sutUsername"),
        ("sut_id", "sutId"),
    ]:
        value = getattr(args, arg_name)
        if value:
            config[key] = int(value) if key in {"apiId", "sutId"} else value
            changed = True
    if not changed:
        raise DriverError("No config values passed.")
    write_json_private(CONFIG_PATH, config)
    print(f"Wrote {CONFIG_PATH}")


def command_doctor(args):
    config, bot_config = load_config()
    lib_path = find_tdjson(config)
    tdjson_loads = False
    tdjson_error = ""
    if lib_path:
        try:
            ctypes.CDLL(str(lib_path))
            tdjson_loads = True
        except OSError as error:
            tdjson_error = str(error)
    checks = {
        "configPath": str(CONFIG_PATH),
        "tdjson": str(lib_path) if lib_path else "",
        "tdjsonLoads": tdjson_loads,
        "tdjsonError": tdjson_error,
        "hasApiId": bool(env_or_config("TELEGRAM_USER_DRIVER_API_ID", config, "apiId")),
        "hasApiHash": bool(env_or_config("TELEGRAM_USER_DRIVER_API_HASH", config, "apiHash")),
        "hasDefaultChat": bool(default_chat(config, bot_config)),
        "hasSutIdentity": bool(resolve_sut(config, bot_config).get("username")),
        "hasTesterIdentity": bool(config.get("testerUserId")),
        "testerUsername": config.get("testerUsername", ""),
        "hasQrencode": bool(shutil.which("qrencode")),
    }
    checks["ok"] = bool(checks["tdjsonLoads"] and checks["hasApiId"] and checks["hasApiHash"])
    if checks["ok"]:
        checks["next"] = "run login --qr"
    elif not checks["tdjsonLoads"]:
        checks["next"] = "brew install tdlib"
    else:
        checks["next"] = "configure --api-id ... --api-hash ..."
    print_result(checks, args.json, getattr(args, "output", ""))
    if not checks["ok"]:
        sys.exit(1)


def command_login(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(args)
    me = driver.client.request({"@type": "getMe"})
    save_tester_identity(config, me)
    print_result({"ok": True, "user": public_user(me)}, args.json, getattr(args, "output", ""))


def command_status(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    ready = driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms), need_ready=False)
    if not ready:
        print_result(
            {"ok": False, "authorized": False, "next": "login --qr"},
            args.json,
            getattr(args, "output", ""),
        )
        sys.exit(1)
    me = driver.client.request({"@type": "getMe"})
    save_tester_identity(config, me)
    print_result(
        {"ok": True, "authorized": True, "user": public_user(me)},
        args.json,
        getattr(args, "output", ""),
    )


def command_confirm_qr(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    result = driver.client.request({"@type": "confirmQrCodeAuthentication", "link": args.link}, timeout=30)
    print_result(
        {
            "ok": True,
            "session": {
                "id": result.get("id"),
                "isCurrent": result.get("is_current"),
                "isPasswordPending": result.get("is_password_pending"),
                "applicationName": result.get("application_name"),
                "deviceModel": result.get("device_model"),
                "platform": result.get("platform"),
                "systemVersion": result.get("system_version"),
            },
        },
        args.json,
        getattr(args, "output", ""),
    )


def command_terminate_session(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    driver.client.request({"@type": "terminateSession", "session_id": int(args.session_id)}, timeout=30)
    print_result(
        {"ok": True, "sessionId": args.session_id},
        args.json,
        getattr(args, "output", ""),
    )


def command_terminate_desktop_sessions(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    result = driver.client.request({"@type": "getActiveSessions"}, timeout=30)
    terminated = []
    for session in result.get("sessions", []):
        if session.get("is_current"):
            continue
        if session.get("application_name") != "Telegram Desktop":
            continue
        session_id = session.get("id")
        if session_id is None:
            continue
        driver.client.request({"@type": "terminateSession", "session_id": int(session_id)}, timeout=30)
        terminated.append({"id": session_id, "applicationName": session.get("application_name")})
    print_result({"ok": True, "terminated": terminated}, args.json, getattr(args, "output", ""))


def public_user(user):
    return {
        "id": user.get("id"),
        "firstName": user.get("first_name"),
        "lastName": user.get("last_name"),
        "username": user.get("username"),
        "isBot": user.get("type", {}).get("@type") == "userTypeBot",
    }


def save_tester_identity(config, user):
    config["testerUserId"] = user.get("id")
    config["testerUsername"] = user.get("username") or ""
    config["testerFirstName"] = user.get("first_name") or ""
    config["testerLastName"] = user.get("last_name") or ""
    write_json_private(CONFIG_PATH, config)


def command_send(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    chat_id = driver.resolve_chat(args.chat)
    text, _run = apply_template(args.text, resolve_sut(config, bot_config))
    sent = driver.send_text(chat_id, text, args.reply_to, args.thread_id)
    print_result(
        {"ok": True, "sent": normalize_message(sent)},
        args.json,
        getattr(args, "output", ""),
    )


def command_wait(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    chat_id = driver.resolve_chat(args.chat)
    message, observed = driver.wait_for_message(chat_id, args, args.after_message_id)
    print_result(
        {
            "ok": bool(message),
            "message": message,
            "observedCount": len(observed),
            "observed": [] if message else observed[-10:],
        },
        args.json,
        getattr(args, "output", ""),
    )
    if not message:
        sys.exit(1)


def command_probe(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    sut = resolve_sut(config, bot_config)
    chat_id = driver.resolve_chat(args.chat)
    text, run = apply_template(args.text, sut)
    sent = driver.send_text(chat_id, text, args.reply_to)
    wait_args = argparse.Namespace(
        expect=args.expect,
        from_bot=args.from_bot,
        reply_to=sent["id"] if args.require_reply else None,
        thread_id=args.thread_id,
        timeout_ms=args.timeout_ms,
    )
    message, observed = driver.wait_for_message(chat_id, wait_args, sent["id"])
    result = {
        "ok": bool(message),
        "chatId": chat_id,
        "run": run,
        "sent": normalize_message(sent),
        "reply": message,
        "observedCount": len(observed),
        "observed": [] if message else observed[-10:],
    }
    print_result(result, args.json, getattr(args, "output", ""))
    if not message:
        sys.exit(1)


def command_transcript(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    chat_id = driver.resolve_chat(args.chat)
    history = driver.client.request(
        {
            "@type": "getChatHistory",
            "chat_id": chat_id,
            "from_message_id": 0,
            "offset": 0,
            "limit": args.limit,
            "only_local": False,
        }
    )
    messages = [normalize_message(message) for message in history.get("messages", [])]
    print_result(
        {"ok": True, "chatId": chat_id, "messages": messages},
        args.json,
        getattr(args, "output", ""),
    )


def command_chats(args):
    config, bot_config = load_config()
    driver = UserDriver(config, bot_config)
    driver.authorize(argparse.Namespace(timeout_ms=args.timeout_ms))
    chats = driver.client.request(
        {
            "@type": "getChats",
            "chat_list": {"@type": "chatListMain"},
            "limit": args.limit,
        },
        timeout=20,
    )
    seen = set()
    rows = []
    for chat_id in chats.get("chat_ids", []):
        chat = driver.client.request({"@type": "getChat", "chat_id": chat_id}, timeout=10)
        rows.append(public_chat(chat, "main"))
        seen.add(int(chat["id"]))
    configured = default_chat(config, bot_config)
    if configured and configured.lstrip("-").isdigit():
        try:
            chat = driver.client.request({"@type": "getChat", "chat_id": int(configured)}, timeout=10)
            if int(chat["id"]) not in seen:
                rows.append(public_chat(chat, "configured"))
        except DriverError:
            pass
    print_result(
        {"ok": True, "configuredChat": configured, "chats": rows},
        args.json,
        getattr(args, "output", ""),
    )


def public_chat(chat, source):
    return {
        "id": chat.get("id"),
        "title": chat.get("title"),
        "type": (chat.get("type") or {}).get("@type"),
        "unreadCount": chat.get("unread_count"),
        "source": source,
    }


def add_common(parser):
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--output", default="")
    parser.add_argument("--timeout-ms", type=int, default=120000)


def main():
    parser = argparse.ArgumentParser(description="Telegram real-user E2E driver backed by TDLib.")
    sub = parser.add_subparsers(dest="command", required=True)

    configure = sub.add_parser("configure")
    configure.add_argument("--api-id")
    configure.add_argument("--api-hash")
    configure.add_argument("--tdlib-path")
    configure.add_argument("--chat")
    configure.add_argument("--sut-username")
    configure.add_argument("--sut-id")
    configure.set_defaults(func=command_configure)

    doctor = sub.add_parser("doctor")
    doctor.add_argument("--json", action="store_true")
    doctor.add_argument("--output", default="")
    doctor.set_defaults(func=command_doctor)

    login = sub.add_parser("login")
    add_common(login)
    login.add_argument("--qr", action="store_true", default=True)
    login.add_argument("--phone", default="")
    login.add_argument("--code", default="")
    login.add_argument("--password", default="")
    login.set_defaults(func=command_login)

    status = sub.add_parser("status")
    add_common(status)
    status.set_defaults(func=command_status)

    confirm_qr = sub.add_parser("confirm-qr")
    add_common(confirm_qr)
    confirm_qr.add_argument("--link", required=True)
    confirm_qr.set_defaults(func=command_confirm_qr)

    terminate_session = sub.add_parser("terminate-session")
    add_common(terminate_session)
    terminate_session.add_argument("--session-id", required=True)
    terminate_session.set_defaults(func=command_terminate_session)

    terminate_desktop_sessions = sub.add_parser("terminate-desktop-sessions")
    add_common(terminate_desktop_sessions)
    terminate_desktop_sessions.set_defaults(func=command_terminate_desktop_sessions)

    send = sub.add_parser("send")
    add_common(send)
    send.add_argument("--chat", default="")
    send.add_argument("--text", required=True)
    send.add_argument("--reply-to")
    send.add_argument("--thread-id", type=int, default=0)
    send.set_defaults(func=command_send)

    wait = sub.add_parser("wait")
    add_common(wait)
    wait.add_argument("--chat", default="")
    wait.add_argument("--expect", action="append", default=[])
    wait.add_argument("--from-bot", default="")
    wait.add_argument("--reply-to")
    wait.add_argument("--thread-id", type=int, default=0)
    wait.add_argument("--after-message-id", type=int, default=0)
    wait.set_defaults(func=command_wait)

    probe = sub.add_parser("probe")
    add_common(probe)
    probe.add_argument("--chat", default="")
    probe.add_argument("--text", default="@{sut} Reply exactly: USER-E2E-{run}")
    probe.add_argument("--expect", action="append", default=[])
    probe.add_argument("--from-bot", default="")
    probe.add_argument("--reply-to")
    probe.add_argument("--thread-id", type=int, default=0)
    probe.add_argument("--require-reply", action="store_true", default=True)
    probe.add_argument("--any-sut-reply", dest="require_reply", action="store_false")
    probe.set_defaults(func=command_probe)

    transcript = sub.add_parser("transcript")
    add_common(transcript)
    transcript.add_argument("--chat", default="")
    transcript.add_argument("--limit", type=int, default=20)
    transcript.set_defaults(func=command_transcript)

    chats = sub.add_parser("chats")
    add_common(chats)
    chats.add_argument("--limit", type=int, default=50)
    chats.set_defaults(func=command_chats)

    serve = sub.add_parser("serve", help=argparse.SUPPRESS)
    serve.add_argument("--chat", required=True)
    serve.add_argument("--sut-user-id", required=True, type=int)
    serve.add_argument("--sut-username", required=True)
    serve.add_argument("--socket", required=True)
    serve.add_argument("--pid-file", required=True)
    serve.add_argument("--journal", required=True)
    serve.add_argument("--media-root", required=True)
    serve.set_defaults(func=command_serve)

    terminate_observer = sub.add_parser("terminate-observer", help=argparse.SUPPRESS)
    terminate_observer.add_argument("--pid-file", required=True)
    terminate_observer.add_argument("--socket", required=True)
    terminate_observer.set_defaults(func=command_terminate_observer)

    args = parser.parse_args()
    try:
        args.func(args)
    except DriverError as error:
        print(str(error), file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()