From 8e941a57e7eddeaa88dc71fccf8b56615e7ac3e5 Mon Sep 17 00:00:00 2001 From: Brice Vandeputte Date: Sun, 18 Oct 2026 20:37:16 +0200 Subject: [PATCH 1/2] fix: improve Telegram timeout reporting (#18) --- README.md | 10 +- bot/telegram/bridge.py | 526 +++++++++++++++++++++++++++++++++- bot/telegram/bridge_config.py | 6 + bot/telegram/http_api.py | 79 ++++- bot/telegram/opencode_api.py | 56 +++- tests/test_telegram_bridge.py | 306 ++++++++++++++++++++ 6 files changed, 959 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 33ab703..b8641fa 100644 --- a/README.md +++ b/README.md @@ -185,7 +185,15 @@ without reviewing it. ``` 6. Send a message from an authorized chat. If there is an error, - `make logs SERVICE=telegram` shows the HTTP details returned by OpenCode. + Telegram returns a short reference. Search for `request_failed ref=` in + `make logs SERVICE=telegram` to find the client, phase, elapsed time, timeout, and safe error + details. Prompts, credentials, session IDs, and raw OpenCode response bodies are not logged. + +The bridge acknowledges a request immediately, then edits that message with the result. It may +also show generic thinking or tool-progress updates when OpenCode emits matching events; raw +reasoning content is never forwarded. At startup, logs show the effective OpenCode and Telegram +timeouts. OpenCode requests have a 300-second deadline; Telegram HTTP requests use 70 seconds, +with long polling configured for 50 seconds. If Telegram replies `Invalid Opper API key` or the logs mention `401`, regenerate the key in your Opper AI project and replace only the diff --git a/bot/telegram/bridge.py b/bot/telegram/bridge.py index c9446db..c316d24 100644 --- a/bot/telegram/bridge.py +++ b/bot/telegram/bridge.py @@ -1,17 +1,49 @@ """Telegram polling loop and orchestration for the OpenCode bridge.""" +from concurrent.futures import ThreadPoolExecutor import json +import queue +import re import sys +import threading import time import urllib.error import urllib.parse +import uuid import cost_reporting import http_api import markdown_renderer import opencode_api import session_state -from bridge_config import ALLOWED, MODEL, OC, SYSTEM +from bridge_config import ( + ALLOWED, + MODEL, + OC, + OPENCODE_HTTP_TIMEOUT_SECONDS, + OPENCODE_STATUS_TIMEOUT_SECONDS, + SYSTEM, + TELEGRAM_HTTP_TIMEOUT_SECONDS, + TELEGRAM_LONG_POLL_TIMEOUT_SECONDS, + OPENCODE_EVENT_TIMEOUT_SECONDS, +) + + +_REQUEST_EXECUTOR = ThreadPoolExecutor(max_workers=4, thread_name_prefix="telegram-request") +_REQUEST_SLOTS = threading.BoundedSemaphore(32) +_SESSIONS_LOCK = threading.RLock() +_CHAT_LOCKS_LOCK = threading.Lock() +_CHAT_LOCKS = {} + + +class OpenCodeRequestTimeout(TimeoutError): + """The overall time allowed for one OpenCode exchange elapsed.""" + + +def _chat_lock(chat_id): + key = str(chat_id) + with _CHAT_LOCKS_LOCK: + return _CHAT_LOCKS.setdefault(key, threading.Lock()) def _assistant_message_key(entry): @@ -92,6 +124,203 @@ def answer(chat_id, prompt, sessions): return text + cost_reporting.format_exchange_footer([entry["message"] for entry in new_messages]) +def _assistant_response_complete(message): + info = message.get("info", {}) + if not isinstance(info, dict): + return False + if info.get("finish") == "tool-calls": + return False + message_time = info.get("time") + created = message_time.get("completed") if isinstance(message_time, dict) else None + if created: + return True + return info.get("finish") in ("stop", "content-filter") + + +def _session_is_busy(status): + if isinstance(status, str): + return status in ("busy", "retry") + return isinstance(status, dict) and status.get("type") in ("busy", "retry") + + +def _event_details(event): + if not isinstance(event, dict): + return None + properties = event.get("properties") + if not isinstance(properties, dict): + return None + part = properties.get("part") + if not isinstance(part, dict): + part = {} + session_id = properties.get("sessionID") or part.get("sessionID") + message_id = properties.get("messageID") or part.get("messageID") + part_type = part.get("type") + return session_id, message_id, part_type + + +def _watch_events(session_id, reference, events, stop): + error_reported = False + while not stop.is_set(): + try: + for event in opencode_api.iter_events(): + if stop.is_set(): + return + details = _event_details(event) + if details is None: + continue + event_session, message_id, part_type = details + if event_session == session_id and message_id: + try: + events.put_nowait((message_id, part_type)) + except queue.Full: + pass + if not stop.is_set(): + stop.wait(1) + except (TimeoutError, OSError, urllib.error.URLError, RuntimeError, json.JSONDecodeError) as error: + if stop.is_set(): + return + if not error_reported: + _log_event_stream_error(reference, error) + error_reported = True + stop.wait(5) + + +def _log_event_stream_error(reference, error): + detail = _safe_error_detail(error) + print( + f"opencode_event_stream_retry ref={reference} timeout_s={OPENCODE_EVENT_TIMEOUT_SECONDS} " + f"error_type={type(error).__name__} detail={detail}", + file=sys.stderr, + flush=True, + ) + + +def answer_async(chat_id, prompt, sessions, reference, set_phase, progress): + """Submit a prompt asynchronously and collect its correlated result.""" + chat_key = str(chat_id) + set_phase("session_lookup") + with _SESSIONS_LOCK: + session_id = sessions.get(chat_key) + if not session_id: + set_phase("session_create") + session = http_api.http(OC + "/session", {"title": f"Telegram {chat_id}"}) + session_id = session["id"] + with _SESSIONS_LOCK: + sessions[chat_key] = session_id + session_state.save_sessions(sessions) + + set_phase("history_before") + previous_ids = None + try: + previous_ids = _message_ids(opencode_api.session_tree_messages(session_id)) + except Exception as error: + print( + f"opencode_history_read_failed ref={reference} phase=before error_type={type(error).__name__}", + file=sys.stderr, + flush=True, + ) + + request_body = {"system": SYSTEM, "parts": [{"type": "text", "text": prompt}]} + _model_options(request_body) + prompt_message_id = uuid.uuid4().hex + stop_events = threading.Event() + event_queue = queue.Queue(maxsize=128) + event_thread = threading.Thread( + target=_watch_events, + args=(session_id, reference, event_queue, stop_events), + name=f"opencode-events-{reference}", + daemon=True, + ) + event_thread.start() + + deadline = time.monotonic() + OPENCODE_HTTP_TIMEOUT_SECONDS + try: + set_phase("prompt_submit") + opencode_api.submit_async( + session_id, + prompt_message_id, + request_body, + OPENCODE_HTTP_TIMEOUT_SECONDS, + ) + + set_phase("response_wait") + idle_samples = 0 + last_progress = None + response_messages = [] + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise OpenCodeRequestTimeout("OpenCode response deadline exceeded") + + timeout = min(OPENCODE_STATUS_TIMEOUT_SECONDS, remaining) + statuses = opencode_api.session_status(timeout) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise OpenCodeRequestTimeout("OpenCode response deadline exceeded") + + busy = _session_is_busy(statuses.get(session_id)) + idle_samples = 0 if busy else idle_samples + 1 + history_timeout = min(OPENCODE_STATUS_TIMEOUT_SECONDS, remaining) + response_messages = opencode_api.session_messages(session_id, timeout=history_timeout) + if time.monotonic() >= deadline: + raise OpenCodeRequestTimeout("OpenCode response deadline exceeded") + correlated = opencode_api.assistant_messages_for_prompt(response_messages, prompt_message_id) + assistant_ids = { + message.get("info", {}).get("id") + for message in correlated + if message.get("info", {}).get("id") + } + while True: + try: + event_message_id, part_type = event_queue.get_nowait() + except queue.Empty: + break + if event_message_id not in assistant_ids: + continue + stage = { + "reasoning": "OpenCode is thinking…", + "tool": "OpenCode is working with a tool…", + }.get(part_type) + if stage and stage != last_progress: + progress(stage) + last_progress = stage + + if ( + correlated + and _assistant_response_complete(correlated[-1]) + and not busy + and idle_samples >= 2 + ): + break + time.sleep(min(1, max(0, remaining))) + + set_phase("history_after") + new_messages = [] + if previous_ids is not None: + try: + new_messages = _new_assistant_messages( + opencode_api.session_tree_messages(session_id), + previous_ids, + ) + except Exception as error: + print( + f"opencode_history_read_failed ref={reference} phase=after error_type={type(error).__name__}", + file=sys.stderr, + flush=True, + ) + + response = correlated[-1] if correlated else None + if response is None: + raise RuntimeError("OpenCode completed without a correlated assistant response") + _record_exchange_diagnostics(new_messages, response) + text = opencode_api.text_from_response(response) + if previous_ids is None: + return text + "\n\nCost: unavailable" + return text + cost_reporting.format_exchange_footer([entry["message"] for entry in new_messages]) + finally: + stop_events.set() + + def authorized_response(chat_id, prompt, sessions): command = prompt.split(maxsplit=1)[0].split("@", 1)[0].lower() if command == "/start": @@ -100,21 +329,209 @@ def authorized_response(chat_id, prompt, sessions): session_id = sessions.get(str(chat_id)) if not session_id: return "No OpenCode session or cost data yet." - try: - return cost_reporting.cost_report(session_id) - except Exception as error: - print(f"Cost report unavailable ({type(error).__name__}).", file=sys.stderr, flush=True) - return "Cost data unavailable. Please try again later." + return cost_reporting.cost_report(session_id) return answer(chat_id, prompt, sessions) def send(chat_id, text): + result = None for message in markdown_renderer.render_markdown(text): payload = {"chat_id": chat_id, "text": message.html, "parse_mode": "HTML"} try: - http_api.telegram("sendMessage", payload) + result = http_api.telegram("sendMessage", payload) + except http_api.TelegramFormattingError: + result = http_api.telegram("sendMessage", {"chat_id": chat_id, "text": message.plain}) + return result + + +def edit_reply(chat_id, message_id, text): + chunks = markdown_renderer.render_markdown(text) + if not chunks: + chunks = markdown_renderer.render_markdown(" ") + first = chunks[0] + payload = { + "chat_id": chat_id, + "message_id": message_id, + "text": first.html, + "parse_mode": "HTML", + } + try: + http_api.telegram("editMessageText", payload) + except http_api.TelegramFormattingError: + http_api.telegram( + "editMessageText", + {"chat_id": chat_id, "message_id": message_id, "text": first.plain}, + ) + for chunk in chunks[1:]: + try: + http_api.telegram( + "sendMessage", + {"chat_id": chat_id, "text": chunk.html, "parse_mode": "HTML"}, + ) except http_api.TelegramFormattingError: - http_api.telegram("sendMessage", {"chat_id": chat_id, "text": message.plain}) + http_api.telegram("sendMessage", {"chat_id": chat_id, "text": chunk.plain}) + + +def _safe_error_detail(error): + if isinstance(error, OpenCodeRequestTimeout): + return f"deadline_exceeded_s={OPENCODE_HTTP_TIMEOUT_SECONDS}" + if isinstance(error, http_api.OpenCodeHTTPError): + details = [f"http_status={error.status}"] + for key, value in (("error_name", error.error_name), ("error_code", error.error_code)): + if value is not None: + safe_value = re.sub(r"[^A-Za-z0-9_.-]", "", str(value))[:80] + if safe_value: + details.append(f"{key}={safe_value}") + return " ".join(details) + if isinstance(error, opencode_api.OpenCodeResponseError): + error_name = re.sub(r"[^A-Za-z0-9_.-]", "", error.error_name)[:80] or "unknown" + error_code = ( + re.sub(r"[^A-Za-z0-9_.-]", "", str(error.error_code))[:80] + if error.error_code is not None + else None + ) + suffix = f" error_code={error_code}" if error_code else "" + return f"response_error={error_name}{suffix}" + if isinstance(error, urllib.error.URLError): + reason = error.reason + if isinstance(reason, TimeoutError): + return "url_error=TimeoutError socket_timeout=true" + if isinstance(reason, OSError): + return f"url_error={type(reason).__name__} errno={reason.errno}" + return f"url_error={type(reason).__name__}" + if isinstance(error, OSError): + return f"os_error={type(error).__name__} errno={error.errno}" + if isinstance(error, TimeoutError): + return "socket_timeout=true" + if isinstance(error, http_api.TelegramRateLimit): + return f"retry_after={error.retry_after}" + if isinstance(error, RuntimeError): + detail = str(error) + safe_reasons = ( + r"OpenCode returned an invalid [a-z ]+", + r"OpenCode returned no text", + r"OpenCode completed without a correlated assistant response", + r"Telegram API HTTP [0-9]{3} \([A-Za-z]+\)", + ) + if any(re.fullmatch(reason, detail) for reason in safe_reasons): + return f"reason={detail}" + return "detail=redacted" + + +def _is_timeout_error(error): + if isinstance(error, TimeoutError): + return True + if isinstance(error, urllib.error.URLError): + return isinstance(error.reason, TimeoutError) + return isinstance(error, http_api.OpenCodeHTTPError) and error.status in (408, 504) + + +def _log_request_error(reference, update_id, phase, started, error): + if phase.startswith("telegram_"): + client = "telegram" + elif phase in ("request_enqueue", "command"): + client = "bridge" + else: + client = "opencode" + elapsed_ms = int((time.monotonic() - started) * 1000) + timeout = ( + TELEGRAM_HTTP_TIMEOUT_SECONDS + if client == "telegram" + else OPENCODE_HTTP_TIMEOUT_SECONDS + if client == "opencode" + else None + ) + timeout_field = f" timeout_s={timeout}" if timeout is not None else "" + if ( + client == "opencode" + and phase == "response_wait" + and _is_timeout_error(error) + and not isinstance(error, OpenCodeRequestTimeout) + ): + timeout_field += f" socket_timeout_s={OPENCODE_STATUS_TIMEOUT_SECONDS}" + print( + f"request_failed ref={reference} update_id={update_id} client={client} phase={phase} " + f"elapsed_ms={elapsed_ms}{timeout_field} error_type={type(error).__name__} " + f"detail={_safe_error_detail(error)}", + file=sys.stderr, + flush=True, + ) + + +def _request_worker(chat_id, prompt, sessions, update_id, reference, acknowledgement_id): + started = time.monotonic() + state = {"phase": "response"} + lock = _chat_lock(chat_id) + + def set_phase(phase): + state["phase"] = phase + + def progress(text): + set_phase("telegram_progress_edit") + try: + edit_reply(chat_id, acknowledgement_id, f"{text}\n\nRef: {reference}") + except Exception as error: + _log_request_error(reference, update_id, state["phase"], started, error) + set_phase("response_wait") + + try: + with lock: + command = prompt.split(maxsplit=1)[0].split("@", 1)[0].lower() + if command == "/start": + set_phase("command") + response = authorized_response(chat_id, prompt, sessions) + elif command == "/cost": + set_phase("opencode_cost") + response = authorized_response(chat_id, prompt, sessions) + else: + set_phase("opencode_request") + response = answer_async(chat_id, prompt, sessions, reference, set_phase, progress) + set_phase("telegram_final_edit") + edit_reply(chat_id, acknowledgement_id, response) + elapsed_ms = int((time.monotonic() - started) * 1000) + print( + f"request_completed ref={reference} update_id={update_id} elapsed_ms={elapsed_ms}", + file=sys.stderr, + flush=True, + ) + except Exception as error: + failed_phase = state["phase"] + _log_request_error(reference, update_id, state["phase"], started, error) + if failed_phase.startswith("telegram_"): + return + try: + set_phase("telegram_error_edit") + if failed_phase.startswith("opencode") or failed_phase in ( + "session_lookup", + "session_create", + "history_before", + "prompt_submit", + "response_wait", + "history_after", + ): + if _is_timeout_error(error): + error_text = ( + "OpenCode timed out. The request may still be running; " + f"please check before retrying. Ref: {reference}" + ) + else: + error_text = f"OpenCode request failed. Please share this reference with the operator: {reference}" + else: + error_text = f"Request failed. Please share this reference with the operator: {reference}" + edit_reply( + chat_id, + acknowledgement_id, + error_text, + ) + except Exception as delivery_error: + _log_request_error(reference, update_id, state["phase"], started, delivery_error) + + +def _request_worker_with_slot(*args): + try: + _request_worker(*args) + finally: + _REQUEST_SLOTS.release() def handle_update(update, sessions): @@ -122,29 +539,96 @@ def handle_update(update, sessions): chat_id = message.get("chat", {}).get("id") prompt = message.get("text", "").strip() if chat_id is None or not prompt: - return + return True if str(chat_id) not in ALLOWED: print(f"Rejected message for chat_id={chat_id}", file=sys.stderr, flush=True) - return + return True + update_id = update.get("update_id", "unknown") + reference = uuid.uuid4().hex[:8].upper() + started = time.monotonic() try: - send(chat_id, authorized_response(chat_id, prompt, sessions)) + acknowledgement = send(chat_id, f"Request received. Ref: {reference}") except Exception as error: - print(f"OpenCode error: {error}", file=sys.stderr, flush=True) - send(chat_id, "Temporary OpenCode error. Check the service logs.") + _log_request_error(reference, update_id, "telegram_acknowledgement", started, error) + # Telegram may have accepted the send despite a transport timeout. Consume the update + # rather than retrying it and potentially creating a duplicate acknowledgement. + return True + + acknowledgement_id = acknowledgement.get("message_id") if isinstance(acknowledgement, dict) else None + if acknowledgement_id is None: + print( + f"request_failed ref={reference} update_id={update_id} client=telegram " + "phase=telegram_acknowledgement error_type=MissingMessageId detail=redacted", + file=sys.stderr, + flush=True, + ) + return True + print( + f"request_accepted ref={reference} update_id={update_id}", + file=sys.stderr, + flush=True, + ) + if not _REQUEST_SLOTS.acquire(blocking=False): + print( + f"request_rejected ref={reference} update_id={update_id} reason=bridge_queue_full", + file=sys.stderr, + flush=True, + ) + try: + edit_reply( + chat_id, + acknowledgement_id, + f"The bridge is busy. Please retry later. Ref: {reference}", + ) + except Exception as error: + _log_request_error(reference, update_id, "telegram_error_edit", started, error) + return True + try: + _REQUEST_EXECUTOR.submit( + _request_worker_with_slot, + chat_id, + prompt, + sessions, + update_id, + reference, + acknowledgement_id, + ) + except Exception as error: + _REQUEST_SLOTS.release() + _log_request_error(reference, update_id, "request_enqueue", started, error) + try: + edit_reply( + chat_id, + acknowledgement_id, + f"Request failed. Please share this reference with the operator: {reference}", + ) + except Exception as delivery_error: + _log_request_error(reference, update_id, "telegram_error_edit", started, delivery_error) + return True def main(): polling_lock = http_api.acquire_polling_lock() sessions = session_state.load_sessions() + _log_timeout_configuration() + offset = 0 conflict_backoff = 5 while True: try: - updates = http_api.telegram("getUpdates", {"offset": offset, "timeout": 50, "allowed_updates": ["message"]}) or [] + updates = http_api.telegram( + "getUpdates", + { + "offset": offset, + "timeout": TELEGRAM_LONG_POLL_TIMEOUT_SECONDS, + "allowed_updates": ["message"], + }, + ) or [] conflict_backoff = 5 for update in updates: + if not handle_update(update, sessions): + break offset = update["update_id"] + 1 - handle_update(update, sessions) except http_api.TelegramConflict as error: print(f"Telegram polling conflict: {error}; retrying in {conflict_backoff}s.", file=sys.stderr, flush=True) time.sleep(conflict_backoff) @@ -154,5 +638,17 @@ def main(): time.sleep(5) +def _log_timeout_configuration(): + print( + "HTTP timeouts: " + f"OpenCode={OPENCODE_HTTP_TIMEOUT_SECONDS}s " + f"(status={OPENCODE_STATUS_TIMEOUT_SECONDS}s, events={OPENCODE_EVENT_TIMEOUT_SECONDS}s); " + f"Telegram={TELEGRAM_HTTP_TIMEOUT_SECONDS}s " + f"(getUpdates long poll={TELEGRAM_LONG_POLL_TIMEOUT_SECONDS}s).", + file=sys.stderr, + flush=True, + ) + + if __name__ == "__main__": main() diff --git a/bot/telegram/bridge_config.py b/bot/telegram/bridge_config.py index 5575670..0ac09fa 100644 --- a/bot/telegram/bridge_config.py +++ b/bot/telegram/bridge_config.py @@ -9,6 +9,12 @@ OC = os.getenv("OPENCODE_URL", "http://opencode:4096").rstrip("/") MODEL = os.getenv("OPENCODE_MODEL", "") +OPENCODE_HTTP_TIMEOUT_SECONDS = 300 +OPENCODE_STATUS_TIMEOUT_SECONDS = 10 +OPENCODE_EVENT_TIMEOUT_SECONDS = 15 +TELEGRAM_HTTP_TIMEOUT_SECONDS = 70 +TELEGRAM_LONG_POLL_TIMEOUT_SECONDS = 50 + SESSIONS_PATH = "/data/sessions.json" POLLING_LOCK_PATH = "/data/telegram-polling.lock" COST_DIAGNOSTICS_ENABLED = os.getenv("TELEGRAM_COST_DIAGNOSTICS", "false").strip().lower() == "true" diff --git a/bot/telegram/http_api.py b/bot/telegram/http_api.py index 18343d3..d5199f2 100644 --- a/bot/telegram/http_api.py +++ b/bot/telegram/http_api.py @@ -8,7 +8,14 @@ import urllib.request from typing import NoReturn -from bridge_config import OC, POLLING_LOCK_PATH, TG +from bridge_config import ( + OC, + OPENCODE_EVENT_TIMEOUT_SECONDS, + OPENCODE_HTTP_TIMEOUT_SECONDS, + POLLING_LOCK_PATH, + TG, + TELEGRAM_HTTP_TIMEOUT_SECONDS, +) class TelegramConflict(RuntimeError): @@ -27,6 +34,24 @@ class TelegramFormattingError(RuntimeError): """Raised when Telegram rejects message entity or HTML formatting.""" +class OpenCodeHTTPError(RuntimeError): + """An OpenCode HTTP error with a safe status code for diagnostics.""" + + def __init__(self, status, error_name=None, error_code=None): + self.status = status + self.error_name = error_name + self.error_code = error_code + super().__init__(f"OpenCode API HTTP {status}") + + +def http_timeout(url): + if url.startswith(TG): + return TELEGRAM_HTTP_TIMEOUT_SECONDS + if url.startswith(OC): + return OPENCODE_HTTP_TIMEOUT_SECONDS + return TELEGRAM_HTTP_TIMEOUT_SECONDS + + def acquire_polling_lock(): lock = open(POLLING_LOCK_PATH, "w", encoding="utf-8") try: @@ -37,7 +62,9 @@ def acquire_polling_lock(): return lock -def http(url, payload=None, timeout=70): +def http(url, payload=None, timeout=None): + if timeout is None: + timeout = http_timeout(url) data = None if payload is None else json.dumps(payload).encode() request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}) @@ -51,7 +78,22 @@ def http(url, payload=None, timeout=70): _handle_telegram_error(url, error.code, details) if url.startswith(OC): # Session URLs and response bodies can contain private identifiers or content. - raise RuntimeError(f"OpenCode API HTTP {error.code}") from None + error_name = None + error_code = None + try: + response_error = json.loads(details) + if isinstance(response_error, dict): + error_value = response_error.get("error") + if isinstance(error_value, dict): + error_name = error_value.get("name") or error_value.get("code") + elif isinstance(error_value, str): + error_name = error_value + candidate_code = response_error.get("code") + if isinstance(candidate_code, (str, int)): + error_code = candidate_code + except json.JSONDecodeError: + pass + raise OpenCodeHTTPError(error.code, error_name, error_code) from None raise RuntimeError(f"HTTP {error.code} {url}: {details[:1000]}") from error @@ -92,7 +134,36 @@ def _handle_telegram_error(url, status, details) -> NoReturn: def telegram(method, payload): while True: try: - return http(TG + method, payload, 70).get("result") + return http(TG + method, payload, TELEGRAM_HTTP_TIMEOUT_SECONDS).get("result") except TelegramRateLimit as error: print(f"Telegram rate limited ({method}); retrying in {error.retry_after}s.", file=sys.stderr, flush=True) time.sleep(error.retry_after) + + +def opencode_events(): + """Yield decoded OpenCode SSE events without logging response content.""" + request = urllib.request.Request( + OC + "/event", + headers={"Accept": "text/event-stream"}, + ) + try: + with urllib.request.urlopen(request, timeout=OPENCODE_EVENT_TIMEOUT_SECONDS) as response: + for line in response: + if isinstance(line, str): + line = line.encode() + if not line.startswith(b"data:"): + if line.startswith(b":"): + yield None + continue + data = line[5:].strip() + if not data: + yield None + continue + event = json.loads(data) + if isinstance(event, dict) and isinstance(event.get("payload"), dict): + event = event["payload"] + if isinstance(event, dict): + yield event + except urllib.error.HTTPError as error: + error.close() + raise OpenCodeHTTPError(error.code) from None diff --git a/bot/telegram/opencode_api.py b/bot/telegram/opencode_api.py index d7467ed..c08345d 100644 --- a/bot/telegram/opencode_api.py +++ b/bot/telegram/opencode_api.py @@ -3,9 +3,19 @@ import urllib.parse from bridge_config import OC +import http_api from http_api import http +class OpenCodeResponseError(RuntimeError): + """OpenCode returned an assistant error; keep user/provider text out of logs.""" + + def __init__(self, error_name, error_code=None): + self.error_name = error_name if isinstance(error_name, str) else "unknown" + self.error_code = error_code if isinstance(error_code, (str, int)) else None + super().__init__("OpenCode returned an error response") + + def text_from_response(response): text = "\n".join( part.get("text", "") @@ -17,15 +27,17 @@ def text_from_response(response): error = response.get("info", {}).get("error", {}) details = error.get("data", {}) if isinstance(error, dict) else {} - message = details.get("message") or error.get("name") if isinstance(error, dict) else None + detail_message = details.get("message") if isinstance(details, dict) else None + message = detail_message or error.get("name") if isinstance(error, dict) else None if message: - raise RuntimeError(f"OpenCode/Opper: {message}") + error_code = details.get("code") if isinstance(details, dict) else None + raise OpenCodeResponseError(error.get("name"), error_code) raise RuntimeError("OpenCode returned no text") -def session_messages(session_id): +def session_messages(session_id, timeout=None): encoded_id = urllib.parse.quote(session_id) - messages = http(f"{OC}/session/{encoded_id}/message") + messages = http(f"{OC}/session/{encoded_id}/message", timeout=timeout) if not isinstance(messages, list): raise RuntimeError("OpenCode returned an invalid message history") return messages @@ -56,3 +68,39 @@ def visit(current_session_id): visit(session_id) return messages + + +def submit_async(session_id, message_id, body, timeout): + """Submit a prompt without waiting for the assistant response.""" + encoded_id = urllib.parse.quote(session_id) + request_body = dict(body) + request_body["messageID"] = message_id + return http(f"{OC}/session/{encoded_id}/prompt_async", request_body, timeout) + + +def session_status(timeout): + """Return the server's active session status map.""" + statuses = http(f"{OC}/session/status", timeout=timeout) + if not isinstance(statuses, dict): + raise RuntimeError("OpenCode returned an invalid session status map") + return statuses + + +def assistant_messages_for_prompt(messages, prompt_message_id): + """Select the assistant chain rooted at this prompt, preserving message order.""" + parent_ids = {prompt_message_id} + selected = [] + for message in messages: + if not isinstance(message, dict) or not isinstance(message.get("info"), dict): + continue + info = message["info"] + if info.get("role") != "assistant" or info.get("parentID") not in parent_ids: + continue + selected.append(message) + if info.get("id"): + parent_ids.add(info["id"]) + return selected + + +def iter_events(): + return http_api.opencode_events() diff --git a/tests/test_telegram_bridge.py b/tests/test_telegram_bridge.py index 0bf5abc..961deef 100644 --- a/tests/test_telegram_bridge.py +++ b/tests/test_telegram_bridge.py @@ -3,9 +3,11 @@ import os import sys import tempfile +import time import urllib.error import unittest from datetime import datetime, timezone +from types import SimpleNamespace from unittest.mock import call, patch @@ -34,6 +36,9 @@ def __exit__(self, *_args): def read(self): return self.body + def __iter__(self): + return iter(self.body.splitlines(keepends=True)) + def telegram_http_error(code, body, method="getUpdates"): return urllib.error.HTTPError( @@ -46,6 +51,26 @@ def telegram_http_error(code, body, method="getUpdates"): class TelegramBridgeTests(unittest.TestCase): + def test_http_timeout_defaults_are_separate_for_each_client(self): + for url, expected in ( + (bridge_config.OC + "/global/health", 300), + (bridge_config.TG + "getMe", 70), + ): + with self.subTest(url=url): + with patch("urllib.request.urlopen", return_value=FakeResponse(b"{}")) as urlopen: + http_api.http(url) + self.assertEqual(urlopen.call_args.kwargs["timeout"], expected) + + def test_opencode_sse_events_are_decoded_without_losing_payload_fields(self): + event = {"type": "message.part.updated", "properties": {"sessionID": "session-1"}} + body = f"data: {json.dumps({'payload': event})}\n\n".encode() + with patch("urllib.request.urlopen", return_value=FakeResponse(body)) as urlopen: + result = list(http_api.opencode_events()) + + self.assertEqual(result, [event]) + self.assertEqual(urlopen.call_args.kwargs["timeout"], 15) + self.assertEqual(urlopen.call_args.args[0].full_url, bridge_config.OC + "/event") + def test_http_requests_do_not_add_basic_authentication(self): for url in (bridge_config.OC + "/global/health", bridge_config.TG + "getMe"): with self.subTest(url=url): @@ -106,6 +131,262 @@ def test_send_splits_long_replies(self): self.assertEqual(send_message.call_count, 3) self.assertEqual([len(item.args[1]["text"]) for item in send_message.call_args_list], [4000, 4000, 500]) + def test_edit_reply_replaces_ack_and_sends_remaining_chunks(self): + with patch.object(http_api, "telegram") as telegram: + bridge.edit_reply(123, 456, "x" * 8500) + + self.assertEqual( + [item.args[0] for item in telegram.call_args_list], + ["editMessageText", "sendMessage", "sendMessage"], + ) + self.assertEqual(telegram.call_args_list[0].args[1]["message_id"], 456) + self.assertEqual( + [len(item.args[1]["text"]) for item in telegram.call_args_list], + [4000, 4000, 500], + ) + + def test_handle_update_sends_ack_before_enqueuing_worker(self): + update = { + "update_id": 987, + "message": {"chat": {"id": 123}, "text": "Explain this"}, + } + with patch.object(bridge, "send", return_value={"message_id": 654}) as send: + with patch.object(bridge._REQUEST_EXECUTOR, "submit") as submit: + with patch.object(bridge, "_REQUEST_SLOTS") as slots: + slots.acquire.return_value = True + bridge.handle_update(update, {}) + + send.assert_called_once() + self.assertIn("Ref:", send.call_args.args[1]) + submit.assert_called_once() + self.assertEqual( + submit.call_args.args[:3], + (bridge._request_worker_with_slot, 123, "Explain this"), + ) + self.assertEqual( + submit.call_args.args[3:], + ({}, 987, send.call_args.args[1].rsplit(" ", 1)[-1], 654), + ) + + def test_async_answer_uses_correlated_completion_and_reports_cost(self): + prompt_id = "prompt-id" + assistant = { + "info": { + "id": "assistant-id", + "parentID": prompt_id, + "role": "assistant", + "time": {"completed": 123}, + "cost": 0.25, + }, + "parts": [{"type": "text", "text": "Done"}], + } + sessions = {"123": "session-1"} + phases = [] + with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id)): + with patch.object( + opencode_api, + "session_tree_messages", + side_effect=[[], [{"session_id": "session-1", "message": assistant}]], + ): + with patch.object(opencode_api, "submit_async") as submit: + with patch.object(opencode_api, "session_status", return_value={}): + with patch.object(opencode_api, "session_messages", return_value=[assistant]): + with patch.object(opencode_api, "iter_events", return_value=iter(())): + result = bridge.answer_async( + 123, + "Hi", + sessions, + "REF12345", + phases.append, + lambda _text: None, + ) + + self.assertEqual(result, "Done\n\nCost: $0.250000") + submit.assert_called_once() + self.assertEqual( + submit.call_args.args[:3], + ( + "session-1", + prompt_id, + { + "system": bridge_config.SYSTEM, + "parts": [{"type": "text", "text": "Hi"}], + }, + ), + ) + self.assertEqual(submit.call_args.args[3], 300) + self.assertIn("response_wait", phases) + + def test_async_answer_forwards_only_safe_progress_for_its_assistant_message(self): + prompt_id = "prompt-id" + incomplete = { + "info": {"id": "assistant-id", "parentID": prompt_id, "role": "assistant"}, + "parts": [], + } + assistant = { + "info": { + "id": "assistant-id", + "parentID": prompt_id, + "role": "assistant", + "time": {"completed": 123}, + "cost": 0, + }, + "parts": [{"type": "text", "text": "Done"}], + } + + def event_stream(): + yield { + "type": "message.part.updated", + "properties": { + "part": { + "sessionID": "session-1", + "messageID": "assistant-id", + "type": "reasoning", + "text": "private reasoning content", + } + }, + } + time.sleep(0.02) + + progress = [] + with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id)): + with patch.object( + opencode_api, + "session_tree_messages", + side_effect=[[], [{"session_id": "session-1", "message": assistant}]], + ): + with patch.object(opencode_api, "submit_async"): + with patch.object( + opencode_api, + "session_status", + side_effect=[{"session-1": {"type": "busy"}}, {}, {}], + ): + with patch.object( + opencode_api, + "session_messages", + side_effect=[[incomplete], [assistant], [assistant]], + ): + with patch.object(opencode_api, "iter_events", side_effect=event_stream): + result = bridge.answer_async( + 123, + "Hi", + {"123": "session-1"}, + "REF12345", + lambda _phase: None, + progress.append, + ) + + self.assertEqual(result, "Done\n\nCost: $0.000000") + self.assertEqual(progress, ["OpenCode is thinking…"]) + self.assertNotIn("private reasoning content", " ".join(progress)) + + def test_async_answer_raises_explicit_timeout_at_deadline(self): + with patch.object(bridge, "OPENCODE_HTTP_TIMEOUT_SECONDS", 0): + with patch.object(opencode_api, "session_tree_messages", return_value=[]): + with patch.object(opencode_api, "submit_async"): + with patch.object(opencode_api, "iter_events", return_value=iter(())): + with self.assertRaises(bridge.OpenCodeRequestTimeout): + bridge.answer_async( + 123, + "Hi", + {"123": "session-1"}, + "REF12345", + lambda _phase: None, + lambda _text: None, + ) + + def test_assistant_message_chain_is_correlated_to_its_user_prompt(self): + messages = [ + {"info": {"id": "other", "parentID": "other-prompt", "role": "assistant"}}, + {"info": {"id": "step-1", "parentID": "prompt", "role": "assistant"}}, + {"info": {"id": "step-2", "parentID": "step-1", "role": "assistant"}}, + ] + + result = opencode_api.assistant_messages_for_prompt(messages, "prompt") + + self.assertEqual([message["info"]["id"] for message in result], ["step-1", "step-2"]) + + def test_safe_error_details_do_not_include_untrusted_exception_text(self): + private_detail = "private prompt and token-like-value" + self.assertEqual(bridge._safe_error_detail(RuntimeError(private_detail)), "detail=redacted") + wrapped_timeout = urllib.error.URLError(TimeoutError(private_detail)) + self.assertTrue(bridge._is_timeout_error(wrapped_timeout)) + self.assertEqual( + bridge._safe_error_detail(wrapped_timeout), + "url_error=TimeoutError socket_timeout=true", + ) + with self.assertRaises(opencode_api.OpenCodeResponseError) as caught: + opencode_api.text_from_response( + {"info": {"error": {"name": "ProviderError", "data": {"message": private_detail}}}} + ) + self.assertNotIn(private_detail, str(caught.exception)) + self.assertEqual(bridge._safe_error_detail(caught.exception), "response_error=ProviderError") + + def test_open_code_retry_status_is_still_considered_busy(self): + self.assertTrue(bridge._session_is_busy({"type": "retry"})) + + def test_worker_reports_timeout_without_claiming_request_was_cancelled(self): + output = io.StringIO() + with patch.object(bridge, "answer_async", side_effect=bridge.OpenCodeRequestTimeout()): + with patch.object(bridge, "edit_reply") as edit: + with patch("sys.stderr", output): + bridge._request_worker(123, "private prompt", {}, 987, "REF12345", 654) + + self.assertIn("may still be running", edit.call_args.args[2]) + self.assertIn("REF12345", edit.call_args.args[2]) + self.assertIn("phase=opencode_request", output.getvalue()) + self.assertIn("timeout_s=300", output.getvalue()) + self.assertNotIn("private prompt", output.getvalue()) + + def test_worker_reports_generic_opencode_error_with_reference_only(self): + output = io.StringIO() + with patch.object(bridge, "answer_async", side_effect=RuntimeError("private prompt text")): + with patch.object(bridge, "edit_reply") as edit: + with patch("sys.stderr", output): + bridge._request_worker(123, "Hi", {}, 990, "REF98765", 656) + + self.assertIn("OpenCode request failed", edit.call_args.args[2]) + self.assertIn("REF98765", edit.call_args.args[2]) + self.assertNotIn("private prompt text", output.getvalue()) + self.assertIn("detail=redacted", output.getvalue()) + + def test_worker_does_not_send_a_second_message_when_final_edit_fails(self): + output = io.StringIO() + with patch.object(bridge, "answer_async", return_value="Done"): + with patch.object(bridge, "edit_reply", side_effect=TimeoutError()) as edit: + with patch("sys.stderr", output): + bridge._request_worker(123, "Hi", {}, 989, "REF54321", 655) + + edit.assert_called_once() + self.assertIn("phase=telegram_final_edit", output.getvalue()) + self.assertNotIn("phase=telegram_error_edit", output.getvalue()) + + def test_telegram_ack_timeout_is_not_misreported_as_opencode_failure(self): + update = { + "update_id": 988, + "message": {"chat": {"id": 123}, "text": "Hi"}, + } + output = io.StringIO() + with patch.object(bridge, "send", side_effect=TimeoutError()): + with patch("sys.stderr", output): + handled = bridge.handle_update(update, {}) + + self.assertTrue(handled) + self.assertIn("client=telegram", output.getvalue()) + self.assertIn("phase=telegram_acknowledgement", output.getvalue()) + self.assertIn("timeout_s=70", output.getvalue()) + + def test_timeout_configuration_is_logged_without_secrets(self): + output = io.StringIO() + with patch("sys.stderr", output): + bridge._log_timeout_configuration() + + self.assertIn("OpenCode=300s", output.getvalue()) + self.assertIn("events=15s", output.getvalue()) + self.assertIn("Telegram=70s", output.getvalue()) + self.assertIn("long poll=50s", output.getvalue()) + self.assertNotIn("unit-test-secret", output.getvalue()) + def test_409_conflict_is_recognized_without_exposing_token(self): conflict = telegram_http_error(409, {"description": "another getUpdates request is active"}) with patch("urllib.request.urlopen", side_effect=conflict): @@ -130,6 +411,31 @@ def test_opencode_http_errors_do_not_log_session_ids_or_response_bodies(self): self.assertEqual(str(caught.exception), "OpenCode API HTTP 500") self.assertTrue(caught.exception.__suppress_context__) + def test_opencode_http_error_retains_only_safe_classification_metadata(self): + error = urllib.error.HTTPError( + bridge_config.OC + "/session/private-session-id/message", + 503, + "Server error", + {}, + io.BytesIO( + json.dumps( + { + "error": {"name": "ProviderUnavailable", "message": "private response text"}, + "code": "UPSTREAM_DOWN", + } + ).encode() + ), + ) + with patch("urllib.request.urlopen", side_effect=error): + with self.assertRaises(http_api.OpenCodeHTTPError) as caught: + http_api.http(bridge_config.OC + "/session/private-session-id/message") + + self.assertEqual(caught.exception.status, 503) + self.assertEqual(caught.exception.error_name, "ProviderUnavailable") + self.assertEqual(caught.exception.error_code, "UPSTREAM_DOWN") + self.assertNotIn("private response text", str(caught.exception)) + self.assertNotIn("private-session-id", str(caught.exception)) + def test_429_waits_for_retry_after_then_retries(self): rate_limit = telegram_http_error( 429, From fa037f4ef7ee1d3f8060afc562b5f2606341781e Mon Sep 17 00:00:00 2001 From: Brice Vandeputte Date: Sun, 18 Oct 2026 20:46:30 +0200 Subject: [PATCH 2/2] fix: use valid OpenCode message IDs (#18) --- bot/telegram/bridge.py | 6 ++++- bot/telegram/http_api.py | 50 ++++++++++++++++++++++++++++++----- tests/test_telegram_bridge.py | 46 +++++++++++++++++++++++++++----- 3 files changed, 89 insertions(+), 13 deletions(-) diff --git a/bot/telegram/bridge.py b/bot/telegram/bridge.py index c316d24..a90d656 100644 --- a/bot/telegram/bridge.py +++ b/bot/telegram/bridge.py @@ -222,7 +222,7 @@ def answer_async(chat_id, prompt, sessions, reference, set_phase, progress): request_body = {"system": SYSTEM, "parts": [{"type": "text", "text": prompt}]} _model_options(request_body) - prompt_message_id = uuid.uuid4().hex + prompt_message_id = f"msg_{uuid.uuid4().hex}" stop_events = threading.Event() event_queue = queue.Queue(maxsize=128) event_thread = threading.Thread( @@ -382,6 +382,10 @@ def _safe_error_detail(error): safe_value = re.sub(r"[^A-Za-z0-9_.-]", "", str(value))[:80] if safe_value: details.append(f"{key}={safe_value}") + if error.validation_fields: + details.append(f"validation_fields={','.join(error.validation_fields)}") + if error.validation_codes: + details.append(f"validation_codes={','.join(error.validation_codes)}") return " ".join(details) if isinstance(error, opencode_api.OpenCodeResponseError): error_name = re.sub(r"[^A-Za-z0-9_.-]", "", error.error_name)[:80] or "unknown" diff --git a/bot/telegram/http_api.py b/bot/telegram/http_api.py index d5199f2..e5f6785 100644 --- a/bot/telegram/http_api.py +++ b/bot/telegram/http_api.py @@ -2,6 +2,7 @@ import fcntl import json +import re import sys import time import urllib.error @@ -37,10 +38,12 @@ class TelegramFormattingError(RuntimeError): class OpenCodeHTTPError(RuntimeError): """An OpenCode HTTP error with a safe status code for diagnostics.""" - def __init__(self, status, error_name=None, error_code=None): + def __init__(self, status, error_name=None, error_code=None, validation_fields=(), validation_codes=()): self.status = status self.error_name = error_name self.error_code = error_code + self.validation_fields = validation_fields + self.validation_codes = validation_codes super().__init__(f"OpenCode API HTTP {status}") @@ -80,20 +83,55 @@ def http(url, payload=None, timeout=None): # Session URLs and response bodies can contain private identifiers or content. error_name = None error_code = None + validation_fields = set() + validation_codes = set() try: response_error = json.loads(details) if isinstance(response_error, dict): + error_name = response_error.get("name") error_value = response_error.get("error") - if isinstance(error_value, dict): - error_name = error_value.get("name") or error_value.get("code") - elif isinstance(error_value, str): - error_name = error_value + if error_name is None: + if isinstance(error_value, dict): + error_name = error_value.get("name") or error_value.get("code") + elif isinstance(error_value, str): + error_name = error_value + error_data = response_error.get("data") candidate_code = response_error.get("code") + if candidate_code is None and isinstance(error_data, dict): + candidate_code = error_data.get("code") if isinstance(candidate_code, (str, int)): error_code = candidate_code + issue_list = response_error.get("issues") or response_error.get("errors") + if isinstance(error_data, dict): + issue_list = issue_list or error_data.get("issues") or error_data.get("errors") + if isinstance(issue_list, list): + for issue in issue_list: + if not isinstance(issue, dict): + continue + path = issue.get("path") + if isinstance(path, list): + safe_parts = [ + re.sub(r"[^A-Za-z0-9_.-]", "", str(part))[:40] + for part in path + if isinstance(part, (str, int)) + ] + safe_path = ".".join(part for part in safe_parts if part)[:100] + if safe_path: + validation_fields.add(safe_path) + issue_code = issue.get("code") + if isinstance(issue_code, str): + safe_code = re.sub(r"[^A-Za-z0-9_.-]", "", issue_code)[:80] + if safe_code: + validation_codes.add(safe_code) except json.JSONDecodeError: pass - raise OpenCodeHTTPError(error.code, error_name, error_code) from None + raise OpenCodeHTTPError( + error.code, + error_name, + error_code, + tuple(sorted(validation_fields)), + tuple(sorted(validation_codes)), + ) from None raise RuntimeError(f"HTTP {error.code} {url}: {details[:1000]}") from error diff --git a/tests/test_telegram_bridge.py b/tests/test_telegram_bridge.py index 961deef..eeedc75 100644 --- a/tests/test_telegram_bridge.py +++ b/tests/test_telegram_bridge.py @@ -169,7 +169,7 @@ def test_handle_update_sends_ack_before_enqueuing_worker(self): ) def test_async_answer_uses_correlated_completion_and_reports_cost(self): - prompt_id = "prompt-id" + prompt_id = "msg_prompt-id" assistant = { "info": { "id": "assistant-id", @@ -182,7 +182,7 @@ def test_async_answer_uses_correlated_completion_and_reports_cost(self): } sessions = {"123": "session-1"} phases = [] - with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id)): + with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id.removeprefix("msg_"))): with patch.object( opencode_api, "session_tree_messages", @@ -218,7 +218,7 @@ def test_async_answer_uses_correlated_completion_and_reports_cost(self): self.assertIn("response_wait", phases) def test_async_answer_forwards_only_safe_progress_for_its_assistant_message(self): - prompt_id = "prompt-id" + prompt_id = "msg_prompt-id" incomplete = { "info": {"id": "assistant-id", "parentID": prompt_id, "role": "assistant"}, "parts": [], @@ -249,7 +249,7 @@ def event_stream(): time.sleep(0.02) progress = [] - with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id)): + with patch("bridge.uuid.uuid4", return_value=SimpleNamespace(hex=prompt_id.removeprefix("msg_"))): with patch.object( opencode_api, "session_tree_messages", @@ -298,11 +298,11 @@ def test_async_answer_raises_explicit_timeout_at_deadline(self): def test_assistant_message_chain_is_correlated_to_its_user_prompt(self): messages = [ {"info": {"id": "other", "parentID": "other-prompt", "role": "assistant"}}, - {"info": {"id": "step-1", "parentID": "prompt", "role": "assistant"}}, + {"info": {"id": "step-1", "parentID": "msg_prompt", "role": "assistant"}}, {"info": {"id": "step-2", "parentID": "step-1", "role": "assistant"}}, ] - result = opencode_api.assistant_messages_for_prompt(messages, "prompt") + result = opencode_api.assistant_messages_for_prompt(messages, "msg_prompt") self.assertEqual([message["info"]["id"] for message in result], ["step-1", "step-2"]) @@ -436,6 +436,40 @@ def test_opencode_http_error_retains_only_safe_classification_metadata(self): self.assertNotIn("private response text", str(caught.exception)) self.assertNotIn("private-session-id", str(caught.exception)) + def test_opencode_validation_error_logs_field_names_but_not_values(self): + error = urllib.error.HTTPError( + bridge_config.OC + "/session/session-1/prompt_async", + 400, + "Bad request", + {}, + io.BytesIO( + json.dumps( + { + "name": "InvalidRequestError", + "data": { + "issues": [ + { + "path": ["body", "messageID"], + "code": "invalid_value", + "message": "private validation detail", + } + ] + }, + } + ).encode() + ), + ) + with patch("urllib.request.urlopen", side_effect=error): + with self.assertRaises(http_api.OpenCodeHTTPError) as caught: + http_api.http(bridge_config.OC + "/session/session-1/prompt_async") + + detail = bridge._safe_error_detail(caught.exception) + self.assertIn("http_status=400", detail) + self.assertIn("error_name=InvalidRequestError", detail) + self.assertIn("validation_fields=body.messageID", detail) + self.assertIn("validation_codes=invalid_value", detail) + self.assertNotIn("private validation detail", detail) + def test_429_waits_for_retry_after_then_retries(self): rate_limit = telegram_http_error( 429,