commit 88361206eeba21a25855572bad8391ed68e13ddb Author: yunyaozhou Date: Tue Jun 30 08:11:31 2026 +0800 feat: add codex retry gateway tui client diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7fe6add --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.venv/ +__pycache__/ +.pytest_cache/ +dist/ +build/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..729c91b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Shujakuin + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..c988b4b --- /dev/null +++ b/README.md @@ -0,0 +1,73 @@ +# codex-retry-gateway-tui + +Terminal UI for `codex-retry-gateway`. + +## Quick Start + +Run once from a public git repo with `uvx`: + +```bash +uvx --from git+https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui.git codex-retry-gateway-tui +``` + +Install as a user command: + +```bash +uv tool install git+https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui.git +codex-retry-gateway-tui --api-url http://127.0.0.1:4610/__codex_retry_gateway +``` + +Default gateway URL: + +```text +http://127.0.0.1:4610/__codex_retry_gateway +``` + +Configuration: + +- `--api-url` / `CODEX_RETRY_GATEWAY_TUI_API_URL` +- `--status-url` / `CODEX_RETRY_GATEWAY_TUI_STATUS_URL` +- `--save-config` +- `--install` +- `--refresh-seconds` / `CODEX_RETRY_GATEWAY_TUI_REFRESH_SECONDS` +- `--timeout` / `CODEX_RETRY_GATEWAY_TUI_TIMEOUT` + +Environment variables: + +- `CODEX_RETRY_GATEWAY_TUI_API_URL` +- `CODEX_RETRY_GATEWAY_TUI_API_URL_FILE` +- `CODEX_RETRY_GATEWAY_TUI_STATUS_URL` +- `CODEX_RETRY_GATEWAY_TUI_STATUS_URL_FILE` +- `CODEX_RETRY_GATEWAY_TUI_VERSION_CHECK_URL` +- `CODEX_RETRY_GATEWAY_TUI_VERSION_CHECK_TIMEOUT` +- `CODEX_RETRY_GATEWAY_TUI_NO_VERSION_CHECK` + +## Local Development + +```bash +cd apps/codex-retry-gateway-tui +uv run codex-retry-gateway-tui +``` + +Views: + +- overview: current gateway status and summary counts +- requests: recent requests, request ID, timing, usage, attempts, and `usage_last_updated_at` +- logs: recent gateway logs +- profiles: saved profiles, active profile default selection, profile actions + +Controls: + +- `1` overview +- `2` requests +- `3` logs +- `4` profiles +- `/` filter +- `r` refresh +- `p` probe selected profile +- `s` switch to selected profile +- `w` save selected profile snapshot +- `d` delete selected inactive profile +- `o` open selected profile upstream URL + +This client uses only the public gateway admin API and does not need SSH or secrets. diff --git a/codex_retry_gateway_tui.py b/codex_retry_gateway_tui.py new file mode 100644 index 0000000..2846f62 --- /dev/null +++ b/codex_retry_gateway_tui.py @@ -0,0 +1,1221 @@ +#!/usr/bin/env python3 +"""Textual TUI for codex-retry-gateway monitoring and control.""" +from __future__ import annotations + +import argparse +import datetime as dt +from concurrent.futures import ThreadPoolExecutor +import importlib.metadata +import json +import os +import re +import subprocess +import sys +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Any + + +APP_NAME = "codex-retry-gateway-tui" +FALLBACK_VERSION = "0.1.0" +DEFAULT_GATEWAY_URL = "http://127.0.0.1:4610/__codex_retry_gateway" +DEFAULT_API_URL = DEFAULT_GATEWAY_URL +DEFAULT_CONFIG_FILE = "~/.config/codex-retry-gateway-tui/api-url" +DEFAULT_STATUS_CONFIG_FILE = "~/.config/codex-retry-gateway-tui/status-url" +DEFAULT_VERSION_CHECK_URL = "https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui/raw/branch/main/pyproject.toml" +DEFAULT_REFRESH_SECONDS = 10 +DEFAULT_TIMEOUT_SECONDS = 5 +DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS = 2 +INSTALL_COMMAND = "uv tool install --force git+https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui.git" +INSTALL_COMMAND_ARGS = [ + "uv", + "tool", + "install", + "--force", + "git+https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui.git", +] + + +def env_int(name: str, default: int, *, minimum: int = 1) -> int: + try: + return max(minimum, int(os.environ.get(name, default))) + except Exception: + return default + + +def configured_url(env_names: tuple[str, ...], config_path: str, default: str = "") -> str: + for name in env_names: + value = os.environ.get(name, "").strip() + if value: + return value + config_file = Path(config_path).expanduser() + try: + for line in config_file.read_text(encoding="utf-8").splitlines(): + value = line.strip() + if value and not value.startswith("#"): + return value + except OSError: + pass + return default + + +def default_api_url() -> str: + return normalize_gateway_url( + configured_url( + ("CODEX_RETRY_GATEWAY_TUI_API_URL",), + os.environ.get("CODEX_RETRY_GATEWAY_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE), + DEFAULT_API_URL, + ) + ) + + +def config_file_path() -> Path: + return Path(os.environ.get("CODEX_RETRY_GATEWAY_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser() + + +def default_status_url() -> str: + return configured_url( + ("CODEX_RETRY_GATEWAY_TUI_STATUS_URL",), + os.environ.get("CODEX_RETRY_GATEWAY_TUI_STATUS_URL_FILE", DEFAULT_STATUS_CONFIG_FILE), + "", + ) + + +def normalize_gateway_url(api_url: str) -> str: + value = str(api_url or "").strip() + if not value: + return "" + + parsed = urllib.parse.urlparse(value) + path = parsed.path.rstrip("/") + for suffix in ( + "/api/status", + "/api/logs", + "/api/requests", + "/api/profiles", + "/api/config", + "/api/restore", + ): + if path.endswith(suffix): + path = path[: -len(suffix)] + break + if path.endswith("/api"): + path = path[:-4] + + normalized = parsed._replace(path=path, params="", query="", fragment="") + return urllib.parse.urlunparse(normalized).rstrip("/") + + +def gateway_status_url(gateway_url: str) -> str: + return f"{normalize_gateway_url(gateway_url).rstrip('/')}/api/status" + + +def build_api_url(gateway_url: str, suffix: str, params: dict[str, Any] | None = None) -> str: + parsed = urllib.parse.urlparse(action_url(gateway_url, suffix)) + query_items: list[tuple[str, str]] = [] + if parsed.query: + query_items.extend(urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)) + for key, value in (params or {}).items(): + if value is None: + continue + if isinstance(value, str) and not value.strip(): + continue + query_items.append((key, str(value))) + if query_items: + parsed = parsed._replace(query=urllib.parse.urlencode(query_items, doseq=True)) + return urllib.parse.urlunparse(parsed) + + +def write_api_url_config(api_url: str) -> Path: + value = str(api_url or "").strip() + if not value: + raise ValueError("api url is empty") + path = config_file_path() + path.parent.mkdir(parents=True, exist_ok=True) + try: + path.parent.chmod(0o700) + except OSError: + pass + path.write_text(value + "\n", encoding="utf-8") + try: + path.chmod(0o600) + except OSError: + pass + return path + + +def run_install_command() -> int: + try: + return subprocess.run(INSTALL_COMMAND_ARGS, check=False).returncode + except FileNotFoundError: + print("uv command not found; install uv first, then run:", INSTALL_COMMAND, file=sys.stderr) + return 127 + + +def current_version() -> str: + try: + return importlib.metadata.version(APP_NAME) + except importlib.metadata.PackageNotFoundError: + return FALLBACK_VERSION + except Exception: + return FALLBACK_VERSION + + +def parse_version(value: Any) -> tuple[int, ...]: + parts = [] + for part in re.split(r"[^0-9]+", str(value or "")): + if part: + parts.append(int(part)) + return tuple(parts) + + +def version_is_newer(latest: str, current: str) -> bool: + latest_parts = parse_version(latest) + current_parts = parse_version(current) + width = max(len(latest_parts), len(current_parts), 1) + return latest_parts + (0,) * (width - len(latest_parts)) > current_parts + (0,) * (width - len(current_parts)) + + +def latest_version_from_text(text: str) -> str: + match = re.search(r'(?m)^version\s*=\s*"([^"]+)"', text) + return "" if not match else match.group(1).strip() + + +def fetch_latest_version(url: str, timeout: int) -> str: + req = urllib.request.Request(url, headers={"Accept": "text/plain"}) + with urllib.request.urlopen(req, timeout=timeout) as response: + return latest_version_from_text(response.read(65536).decode("utf-8", errors="replace")) + + +def version_update_message(latest: str, current: str | None = None) -> str: + current = current or current_version() + if not latest or not version_is_newer(latest, current): + return "" + return f"update available: {APP_NAME} {current} -> {latest}; run `{INSTALL_COMMAND}`" + + +def check_version_update(url: str, timeout: int, *, disabled: bool = False) -> str: + if disabled or os.environ.get("CODEX_RETRY_GATEWAY_TUI_NO_VERSION_CHECK", "").strip().lower() in {"1", "true", "yes", "on"}: + return "" + try: + return version_update_message(fetch_latest_version(url, timeout)) + except Exception: + return "" + + +def as_float(value: Any) -> float: + try: + if value is None or str(value).strip() == "": + return 0.0 + return float(value) + except Exception: + return 0.0 + + +def as_int(value: Any) -> int: + try: + if value is None or str(value).strip() == "": + return 0 + return int(float(value)) + except Exception: + return 0 + + +def short_text(value: Any, length: int = 48) -> str: + text = str(value or "").strip() + if not text: + return "-" + return text if len(text) <= length else f"{text[: length - 3]}..." + + +def short_time(value: Any) -> str: + text = str(value or "").strip() + if not text: + return "-" + return text.replace("T", " ")[:19] + + +def format_ms(value: Any) -> str: + number = as_int(value) + return "-" if number <= 0 else f"{number}ms" + + +def format_bytes(value: Any) -> str: + number = as_int(value) + if number <= 0: + return "-" + if number >= 1_000_000: + return f"{number / 1_000_000:.1f}MB".rstrip("0").rstrip(".") + if number >= 1_000: + return f"{number / 1_000:.1f}KB".rstrip("0").rstrip(".") + return f"{number}B" + + +def format_percent(value: Any) -> str: + if value is None or str(value).strip() == "": + return "-" + percent = as_float(value) + rounded = round(percent, 1) + if rounded.is_integer(): + return f"{int(rounded)}%" + return f"{rounded}%" + + +def format_count(value: Any) -> str: + number = as_int(value) + if abs(number) >= 1_000_000: + return f"{number / 1_000_000:.1f}M".rstrip("0").rstrip(".") + if abs(number) >= 1_000: + return f"{number / 1_000:.1f}K".rstrip("0").rstrip(".") + return str(number) + + +def parse_datetime(value: Any) -> dt.datetime | None: + text = str(value or "").strip() + if not text: + return None + try: + normalized = text[:-1] + "+00:00" if text.endswith("Z") else text + return dt.datetime.fromisoformat(normalized) + except Exception: + return None + + +def format_elapsed_seconds(value: Any) -> str: + number = as_float(value) + if number <= 0: + return "-" + if number >= 100: + return f"{number:.0f}s" + return f"{number:.1f}s" + + +def elapsed_between(started_at: Any, finished_at: Any) -> str: + start = parse_datetime(started_at) + finish = parse_datetime(finished_at) + if not start or not finish: + return "-" + seconds = max(0.0, (finish - start).total_seconds()) + return format_elapsed_seconds(seconds) + + +def age_since(value: Any) -> str: + parsed = parse_datetime(value) + if not parsed: + return "-" + now = dt.datetime.now(parsed.tzinfo) if parsed.tzinfo else dt.datetime.now() + seconds = max(0.0, (now - parsed).total_seconds()) + return format_elapsed_seconds(seconds) + + +def status_kind(value: Any) -> str: + text = str(value or "").strip().lower() + if text in {"ok", "operational", "success"}: + return "ok" + if text in {"failed", "error", "failure"}: + return "failed" + return "unknown" + + +def fetch_payload(api_url: str, timeout: int) -> dict[str, Any]: + req = urllib.request.Request(api_url, headers={"Accept": "application/json"}) + with urllib.request.urlopen(req, timeout=timeout) as response: + data = json.loads(response.read().decode("utf-8")) + if not isinstance(data, dict): + raise RuntimeError("API did not return a JSON object") + return data + + +def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]: + if not str(url or "").strip(): + return {}, "" + try: + return fetch_payload(url, timeout), "" + except Exception as exc: + return {}, str(exc) + + +def text_tokens(value: Any) -> set[str]: + tokens = set() + for token in "".join(ch.lower() if ch.isalnum() else " " for ch in str(value or "")).split(): + if token and not token.isdigit(): + tokens.add(token) + return tokens + + +def format_bool(value: Any) -> str: + return "yes" if bool(value) else "no" + + +def summary_line(payload: dict[str, Any]) -> str: + config = payload.get("config") if isinstance(payload.get("config"), dict) else {} + state = payload.get("state") if isinstance(payload.get("state"), dict) else {} + metrics = payload.get("metrics") if isinstance(payload.get("metrics"), dict) else {} + listen = payload.get("listen") or "-" + upstream = config.get("upstream_base_url") or "-" + active = config.get("profile_name") or state.get("profile_name") or "-" + request_total = format_count(metrics.get("total_proxy_request_count")) + inspected = format_count(metrics.get("inspected_response_count")) + matched = format_count(metrics.get("matched_response_count")) + reasoning_516 = format_count(metrics.get("reasoning_516_count")) + return f"{listen} | profile {active} | upstream {upstream} | req {request_total} | inspected {inspected} | matched {matched} | 516 {reasoning_516}" + + +def state_summary(payload: dict[str, Any]) -> str: + state = payload.get("state") if isinstance(payload.get("state"), dict) else {} + paths = payload.get("paths") if isinstance(payload.get("paths"), dict) else {} + bits = [ + f"state {state.get('state_root') or paths.get('state_root') or '-'}", + f"config {paths.get('config_path') or '-'}", + f"requests {paths.get('requests_path') or '-'}", + f"profiles {paths.get('profiles_dir') or '-'}", + ] + if state.get("gateway_base_url"): + bits.append(f"gateway {state.get('gateway_base_url')}") + if state.get("codex_current_base_url"): + bits.append(f"codex {state.get('codex_current_base_url')}") + return " | ".join(bits) + + +def status_summary(status_payload: dict[str, Any], status_error: str = "") -> str: + if status_error: + return f"status error: {status_error}" + config = status_payload.get("config") if isinstance(status_payload.get("config"), dict) else {} + state = status_payload.get("state") if isinstance(status_payload.get("state"), dict) else {} + metrics = status_payload.get("metrics") if isinstance(status_payload.get("metrics"), dict) else {} + listen = status_payload.get("listen") or "-" + return ( + f"live {listen} | active {config.get('profile_name') or state.get('profile_name') or '-'} | " + f"516 {metrics.get('reasoning_516_count') or 0} | latest seq {metrics.get('total_proxy_request_count') or 0}" + ) + + +def request_match_score(row: dict[str, Any], needle: str) -> bool: + if not needle: + return True + haystack = " ".join( + str(row.get(key) or "") + for key in ( + "request_id", + "profile_name", + "method", + "path", + "model", + "requested_model", + "forwarded_model", + "error", + "upstream_origin", + "upstream_path", + "status_code", + "upstream_status_code", + "reasoning_tokens", + "input_tokens", + "output_tokens", + "total_tokens", + "cached_tokens", + ) + ).lower() + return needle in haystack + + +def status_symbol(entry: dict[str, Any]) -> str: + if entry.get("error"): + return "!" + if entry.get("matched"): + return "*" + if entry.get("response_stream"): + return "~" + return " " + + +def request_lifecycle_label(row: dict[str, Any]) -> str: + state = str(row.get("lifecycle_state") or "").strip().lower() + if state == "finish": + return "done" + if state == "receive_first": + return "first" + if state == "streaming": + return "live" + return "sent" + + +def normalize_request_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: + needle = filter_text.strip().lower() + rows = [] + for entry in payload.get("entries") or []: + if not isinstance(entry, dict): + continue + if needle and not request_match_score(entry, needle): + continue + rows.append( + { + "seq": as_int(entry.get("seq")), + "request_id": str(entry.get("request_id") or ""), + "profile_name": str(entry.get("profile_name") or ""), + "method": str(entry.get("method") or ""), + "path": str(entry.get("path") or ""), + "model": str(entry.get("model") or ""), + "requested_model": str(entry.get("requested_model") or ""), + "forwarded_model": str(entry.get("forwarded_model") or ""), + "request_stream": bool(entry.get("request_stream")), + "response_stream": bool(entry.get("response_stream")), + "matched": bool(entry.get("matched")), + "inspected": bool(entry.get("inspected")), + "status_code": entry.get("status_code"), + "upstream_status_code": entry.get("upstream_status_code"), + "reasoning_tokens": entry.get("reasoning_tokens"), + "input_tokens": entry.get("input_tokens"), + "output_tokens": entry.get("output_tokens"), + "total_tokens": entry.get("total_tokens"), + "cached_tokens": entry.get("cached_tokens"), + "upstream_attempt_count": entry.get("upstream_attempt_count"), + "request_body_bytes": entry.get("request_body_bytes"), + "response_bytes_received": entry.get("response_bytes_received"), + "stream_chunk_count": entry.get("stream_chunk_count"), + "first_response_delay_ms": entry.get("first_response_delay_ms"), + "duration_ms": entry.get("duration_ms"), + "started_at": entry.get("started_at"), + "finished_at": entry.get("finished_at"), + "last_activity_at": entry.get("last_activity_at"), + "usage_last_updated_at": entry.get("usage_last_updated_at"), + "error": str(entry.get("error") or ""), + "upstream_origin": str(entry.get("upstream")["origin"] if isinstance(entry.get("upstream"), dict) else entry.get("upstream_origin") or ""), + "upstream_path": str(entry.get("upstream")["path"] if isinstance(entry.get("upstream"), dict) else entry.get("upstream_path") or ""), + "upstream_auth_mode": str(entry.get("upstream")["auth_mode"] if isinstance(entry.get("upstream"), dict) else entry.get("upstream_auth_mode") or ""), + "upstream_auth_source": str(entry.get("upstream")["auth_source"] if isinstance(entry.get("upstream"), dict) else entry.get("upstream_auth_source") or ""), + "raw": entry, + } + ) + rows.sort(key=lambda item: (item["seq"], item["request_id"]), reverse=True) + return rows + + +def render_request_detail(row: dict[str, Any]) -> str: + bits = [ + f"seq {row['seq']}", + f"id {row['request_id'] or '-'}", + f"{row['method']} {row['path']}", + f"profile {row.get('profile_name') or '-'}", + f"status {row['status_code'] or '-'}", + f"upstream {row['upstream_status_code'] or '-'}", + f"attempts {row['upstream_attempt_count'] or 0}", + f"first {format_ms(row['first_response_delay_ms'])}", + f"duration {format_ms(row['duration_ms'])}", + f"request {format_bytes(row['request_body_bytes'])}", + f"response {format_bytes(row['response_bytes_received'])}", + f"chunks {row['stream_chunk_count'] or 0}", + f"life {request_lifecycle_label(row)}", + f"started {short_time(row['started_at'])}", + f"finished {short_time(row['finished_at'])}", + f"updated {age_since(row['usage_last_updated_at'])}", + ] + if row.get("model"): + bits.append(f"model {row['model']}") + if row.get("requested_model") and row.get("requested_model") != row.get("model"): + bits.append(f"requested {row['requested_model']}") + if row.get("forwarded_model") and row.get("forwarded_model") != row.get("model"): + bits.append(f"forwarded {row['forwarded_model']}") + if row.get("reasoning_tokens") is not None: + bits.append(f"reasoning {row['reasoning_tokens']}") + if row.get("input_tokens") is not None or row.get("output_tokens") is not None or row.get("total_tokens") is not None: + bits.append( + f"usage in/out/total {row.get('input_tokens') or 0}/{row.get('output_tokens') or 0}/{row.get('total_tokens') or 0}" + ) + if row.get("cached_tokens") is not None: + bits.append(f"cached {row['cached_tokens']}") + if row.get("error"): + bits.append(f"error {row['error']}") + if row.get("upstream_origin"): + bits.append(f"origin {row['upstream_origin']}") + if row.get("upstream_path"): + bits.append(f"upstream path {row['upstream_path']}") + if row.get("upstream_auth_mode"): + bits.append(f"auth {row['upstream_auth_mode']}/{row.get('upstream_auth_source') or '-'}") + return " | ".join(bits) + + +def normalize_profile_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]: + needle = filter_text.strip().lower() + rows = [] + for profile in payload.get("profiles") or []: + if not isinstance(profile, dict): + continue + summary = profile.get("summary") if isinstance(profile.get("summary"), dict) else {} + haystack = " ".join( + str(profile.get(key) or "") + for key in ("name", "file_path", "active") + ).lower() + " " + " ".join(str(summary.get(key) or "") for key in summary.keys()).lower() + if needle and needle not in haystack: + continue + rows.append( + { + "name": str(profile.get("name") or ""), + "active": bool(profile.get("active")), + "file_path": str(profile.get("file_path") or ""), + "listen_host": str(summary.get("listen_host") or "-"), + "listen_port": str(summary.get("listen_port") or "-"), + "upstream_base_url": str(summary.get("upstream_base_url") or "-"), + "auth_mode": str(summary.get("auth_mode") or "-"), + "auth_env": str(summary.get("auth_env") or "-"), + "auth_file": str(summary.get("auth_file") or ""), + "auth_json_path": str(summary.get("auth_json_path") or ""), + "auth_json_key": str(summary.get("auth_json_key") or "-"), + "request_history_limit": summary.get("request_history_limit"), + "model_remap": str(summary.get("model_remap") or ""), + "auth_source": str(summary.get("auth_source") or "-"), + "reasoning_equals": summary.get("reasoning_equals"), + "raw": profile, + } + ) + rows.sort(key=lambda item: (not item["active"], item["name"].lower())) + return rows + + +def render_profile_detail(row: dict[str, Any]) -> str: + bits = [ + f"{row['name']}", + "active" if row["active"] else "inactive", + f"listen {row['listen_host']}:{row['listen_port']}", + f"upstream {short_text(row['upstream_base_url'], 64)}", + f"auth {row['auth_mode']}/{row['auth_source']}", + ] + if row.get("request_history_limit") is not None: + bits.append(f"history {row['request_history_limit']}") + if row.get("reasoning_equals"): + bits.append(f"reasoning {row['reasoning_equals']}") + if row.get("model_remap"): + bits.append(f"remap {short_text(row['model_remap'], 96)}") + if row.get("auth_file"): + bits.append("auth file configured") + if row.get("auth_json_path"): + bits.append("auth json configured") + return " | ".join(bits) + + +def action_url(api_url: str, suffix: str) -> str: + base = api_url.rstrip("/") + return f"{base}{suffix}" + + +def post_json(url: str, timeout: int, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json", "Accept": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as response: + data = json.loads(response.read().decode("utf-8")) + if not isinstance(data, dict): + raise RuntimeError("API did not return a JSON object") + return response.status, data + + +def delete_json(url: str, timeout: int) -> tuple[int, dict[str, Any]]: + req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="DELETE") + with urllib.request.urlopen(req, timeout=timeout) as response: + data = json.loads(response.read().decode("utf-8")) + if not isinstance(data, dict): + raise RuntimeError("API did not return a JSON object") + return response.status, data + + +def open_url(url: str) -> None: + if sys.platform == "darwin": + command = ["open", url] + elif sys.platform == "win32": + command = ["cmd", "/c", "start", "", url] + else: + command = ["xdg-open", url] + subprocess.Popen(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + +def fetch_dashboard_snapshot( + gateway_url: str, + status_url: str, + timeout: int, + *, + filter_text: str = "", + current_view: str = "overview", +) -> dict[str, Any]: + gateway_root = normalize_gateway_url(gateway_url) + endpoints = { + "status": status_url or gateway_status_url(gateway_root), + "logs": build_api_url(gateway_root, "/api/logs", {"limit": 200}), + "requests": build_api_url(gateway_root, "/api/requests", {"limit": 200}), + "profiles": build_api_url(gateway_root, "/api/profiles"), + } + + def fetch_optional(url: str) -> tuple[dict[str, Any], str]: + return fetch_optional_payload(url, timeout) + + with ThreadPoolExecutor(max_workers=4) as pool: + futures = { + name: pool.submit(fetch_optional, url) + for name, url in endpoints.items() + } + status_payload, status_error = futures["status"].result() + logs_payload, logs_error = futures["logs"].result() + requests_payload, requests_error = futures["requests"].result() + profiles_payload, profiles_error = futures["profiles"].result() + + payload = status_payload if isinstance(status_payload, dict) else {} + requests_source = requests_payload if isinstance(requests_payload, dict) else {} + logs_source = logs_payload if isinstance(logs_payload, dict) else {} + profiles_source = profiles_payload if isinstance(profiles_payload, dict) else {} + errors = [message for message in (status_error, logs_error, requests_error, profiles_error) if message] + return { + "payload": payload, + "status_payload": payload, + "status_error": " | ".join(errors), + "requests": normalize_request_rows(requests_source, filter_text), + "logs": normalize_log_rows(logs_source), + "profiles": normalize_profile_rows( + profiles_source, + filter_text if current_view == "profiles" else "", + ), + "active_profile_name": ( + str( + payload.get("config", {}).get("profile_name") + or payload.get("state", {}).get("profile_name") + or profiles_source.get("active_profile") + or "" + ).strip() + ), + } + + +def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: int, version_message: str = "") -> int: + try: + from textual.app import App, ComposeResult + from textual.coordinate import Coordinate + from textual.widgets import DataTable, Footer, Header, Input, Static + except ImportError: + print("Textual is required. Run with: uv run --with textual python codex_retry_gateway_tui.py", file=sys.stderr) + return 2 + + gateway_base_url = normalize_gateway_url(api_url) + status_api_url = status_url or gateway_status_url(gateway_base_url) + + class CodexRetryGatewayTui(App[None]): + CSS = """ + Screen { layout: vertical; } + #summary { height: 1; padding: 0 1; color: $accent; } + #paths { height: 1; padding: 0 1; color: $text-muted; } + #controls { height: 2; } + #tables { height: 1fr; } + #requests_table, #logs_table, #profiles_table { height: 1fr; } + #detail { height: 5; padding: 0 1; border-top: solid $panel; } + #status { height: 2; padding: 0 1; color: $text-muted; } + """ + + BINDINGS = [ + ("q", "quit", "Quit"), + ("r", "refresh", "Refresh"), + ("1", "show_overview", "Overview"), + ("2", "show_requests", "Requests"), + ("3", "show_logs", "Logs"), + ("4", "show_profiles", "Profiles"), + ("/", "focus_filter", "Filter"), + ] + + def __init__(self) -> None: + super().__init__() + self.payload: dict[str, Any] = {} + self.status_payload: dict[str, Any] = {} + self.status_error = "" + self.requests: list[dict[str, Any]] = [] + self.logs: list[dict[str, Any]] = [] + self.profiles: list[dict[str, Any]] = [] + self.request_by_key: dict[str, dict[str, Any]] = {} + self.profile_by_key: dict[str, dict[str, Any]] = {} + self.log_by_key: dict[str, dict[str, Any]] = {} + self.current_view = "overview" + self.filter_mode = "requests" + self.last_request_id = "" + self.loading = False + self.executor = ThreadPoolExecutor(max_workers=1) + self.refresh_generation = 0 + self.active_profile_name = "" + self.visible_tables = { + "overview": "requests", + "requests": "requests", + "logs": "logs", + "profiles": "profiles", + } + + def compose(self) -> ComposeResult: + yield Header(show_clock=True) + yield Static("", id="summary") + yield Static("", id="paths") + yield Input(placeholder="requests filter", id="filter") + yield Static("", id="controls") + yield Static("", id="tables") + yield DataTable(id="requests_table") + yield DataTable(id="logs_table") + yield DataTable(id="profiles_table") + yield Static("", id="detail") + yield Static("", id="status") + yield Footer() + + def on_mount(self) -> None: + requests = self.query_one("#requests_table", DataTable) + requests.cursor_type = "row" + requests.zebra_stripes = True + requests.add_columns("Seq", "ID", "Status", "Path", "Model", "Reasoning", "Req", "Resp", "First", "Duration", "Updated", "Note") + + logs = self.query_one("#logs_table", DataTable) + logs.cursor_type = "row" + logs.zebra_stripes = True + logs.add_columns("Seq", "At", "Message") + + profiles = self.query_one("#profiles_table", DataTable) + profiles.cursor_type = "row" + profiles.zebra_stripes = True + profiles.add_columns("Name", "Active", "Listen", "Upstream", "Auth", "History", "Reasoning", "Source") + + self._set_view("overview") + self.refresh_data(refresh=True) + self.set_interval(refresh_seconds, self.refresh_data) + + def on_unmount(self) -> None: + self.executor.shutdown(wait=False, cancel_futures=True) + + def _set_view(self, view: str) -> None: + self.current_view = view + active_table = self.visible_tables.get(view, "requests") + self.query_one("#requests_table").display = active_table == "requests" + self.query_one("#logs_table").display = active_table == "logs" + self.query_one("#profiles_table").display = active_table == "profiles" + self.query_one("#controls", Static).update( + "views: 1 overview | 2 requests | 3 logs | 4 profiles | / filter | r refresh | enter action | n newest request | p probe | s switch | w save | d delete | u open upstream" + ) + self._focus_table() + self._update_table_headers() + + def _focus_table(self) -> None: + table = self._current_table_widget() + if table: + table.focus() + + def _update_summary(self) -> None: + self.query_one("#summary", Static).update(summary_line(self.payload)) + self.query_one("#paths", Static).update(state_summary(self.payload)) + + def _update_table_headers(self) -> None: + counts = f"requests {len(self.requests)} | logs {len(self.logs)} | profiles {len(self.profiles)}" + self.query_one("#tables", Static).update(counts) + + def _current_filter(self) -> str: + return self.query_one("#filter", Input).value.strip() + + def _current_table_widget(self) -> DataTable | None: + table_id = self.visible_tables.get(self.current_view, "requests") + if table_id == "logs": + return self.query_one("#logs_table", DataTable) + if table_id == "profiles": + return self.query_one("#profiles_table", DataTable) + return self.query_one("#requests_table", DataTable) + + def _select_table_row(self, table: DataTable, row_index: int) -> bool: + if row_index < 0 or row_index >= table.row_count: + return False + table.cursor_coordinate = Coordinate(row_index, 0) + return True + + def _select_profile_row(self, profile_name: str) -> bool: + table = self.query_one("#profiles_table", DataTable) + if not self.profiles: + return False + if profile_name: + for index, row in enumerate(self.profiles): + if row["name"] == profile_name: + return self._select_table_row(table, index) + return self._select_table_row(table, 0) + + def _selected_request_row(self) -> dict[str, Any] | None: + table = self.query_one("#requests_table", DataTable) + if table.cursor_row is None or table.cursor_row < 0: + return None + key = str(table.get_row_at(table.cursor_row).key.value) + return self.request_by_key.get(key) + + def _selected_profile_row(self) -> dict[str, Any] | None: + table = self.query_one("#profiles_table", DataTable) + if table.cursor_row is None or table.cursor_row < 0: + return None + key = str(table.get_row_at(table.cursor_row).key.value) + return self.profile_by_key.get(key) + + def _selected_log_row(self) -> dict[str, Any] | None: + table = self.query_one("#logs_table", DataTable) + if table.cursor_row is None or table.cursor_row < 0: + return None + key = str(table.get_row_at(table.cursor_row).key.value) + return self.log_by_key.get(key) + + def _set_status(self, text: str) -> None: + self.query_one("#status", Static).update(text) + + def refresh_data(self, refresh: bool = False) -> None: + if self.loading: + return + self.loading = True + self.refresh_generation += 1 + generation = self.refresh_generation + self._set_status("refreshing...") + current_filter = self._current_filter() + self.executor.submit(self._refresh_in_worker, generation, current_filter) + + def _refresh_in_worker(self, generation: int, current_filter: str) -> None: + try: + result = fetch_dashboard_snapshot( + gateway_base_url, + status_api_url, + timeout, + filter_text=current_filter, + current_view=self.current_view, + ) + result["generation"] = generation + self.call_from_thread(self._apply_refresh_result, result) + except Exception as exc: + self.call_from_thread(self._apply_refresh_error, generation, exc) + + def _apply_refresh_result(self, result: dict[str, Any]) -> None: + if result.get("generation") != self.refresh_generation: + return + self.payload = result["payload"] + self.status_payload = result["status_payload"] + self.status_error = result["status_error"] + self.requests = result["requests"] + self.logs = result["logs"] + self.profiles = result["profiles"] + self.active_profile_name = result.get("active_profile_name") or "" + self.loading = False + self.render_all() + if self.current_view == "profiles": + if not self._select_profile_row(self.active_profile_name): + self._select_profile_row("") + version_bits = [version_message] if version_message else [] + version_bits.append(status_summary(self.payload, self.status_error)) + self._set_status(" | ".join(bit for bit in version_bits if bit)) + + def _apply_refresh_error(self, generation: int, exc: Exception) -> None: + if generation != self.refresh_generation: + return + self.loading = False + self._set_status(f"error: {exc}") + + def render_all(self) -> None: + self._update_summary() + self.render_requests() + self.render_logs() + self.render_profiles() + if self.current_view == "requests": + if self.requests: + self.render_detail(self.requests[0], kind="request") + else: + self.query_one("#detail", Static).update("no requests") + elif self.current_view == "logs": + if self.logs: + self.render_detail(self.logs[0], kind="log") + else: + self.query_one("#detail", Static).update("no logs") + elif self.current_view == "profiles": + if self.profiles: + self.render_detail(self.profiles[0], kind="profile") + else: + self.query_one("#detail", Static).update("no profiles") + + def render_requests(self) -> None: + table = self.query_one("#requests_table", DataTable) + table.clear() + self.request_by_key = {} + for row in self.requests: + key = f"{row['seq']}:{row['request_id']}" + self.request_by_key[key] = row + table.add_row( + str(row["seq"]), + short_text(row["request_id"], 18) or "-", + f"{status_symbol(row)} {row.get('status_code') or '-'}", + short_text(row["path"], 22), + short_text(row["model"] or row["requested_model"] or row["forwarded_model"], 16), + row["reasoning_tokens"] if row["reasoning_tokens"] is not None else "-", + format_bytes(row["request_body_bytes"]), + format_bytes(row["response_bytes_received"]), + format_ms(row["first_response_delay_ms"]), + format_ms(row["duration_ms"]), + age_since(row.get("usage_last_updated_at")), + f"{row['upstream_attempt_count'] or 0}x {short_text(row['error'], 18)}", + key=key, + ) + + def render_logs(self) -> None: + table = self.query_one("#logs_table", DataTable) + table.clear() + self.log_by_key = {} + for row in self.logs: + key = str(row["seq"]) + self.log_by_key[key] = row + table.add_row(str(row["seq"]), short_time(row["at"]), short_text(row["message"], 80), key=key) + + def render_profiles(self) -> None: + table = self.query_one("#profiles_table", DataTable) + table.clear() + self.profile_by_key = {} + for row in self.profiles: + key = row["name"] + self.profile_by_key[key] = row + table.add_row( + row["name"], + "yes" if row["active"] else "", + f"{row['listen_host']}:{row['listen_port']}", + short_text(row["upstream_base_url"], 24), + f"{row['auth_mode']}/{row['auth_source']}", + str(row["request_history_limit"] if row["request_history_limit"] is not None else "-"), + short_text(row["reasoning_equals"], 12), + short_text(row["file_path"], 28), + key=key, + ) + if self.current_view == "profiles" and self.active_profile_name: + self._select_profile_row(self.active_profile_name) + + def render_detail(self, row: dict[str, Any], *, kind: str) -> None: + if kind == "request": + self.query_one("#detail", Static).update(render_request_detail(row)) + elif kind == "profile": + self.query_one("#detail", Static).update(render_profile_detail(row)) + else: + message = str(row.get("message") or "").strip() + self.query_one("#detail", Static).update(f"seq {row['seq']} | {short_time(row['at'])} | {message}") + + def action_refresh(self) -> None: + self.refresh_data(refresh=True) + + def action_focus_filter(self) -> None: + self.query_one("#filter", Input).focus() + + def action_show_overview(self) -> None: + self._set_view("overview") + + def action_show_requests(self) -> None: + self._set_view("requests") + + def action_show_logs(self) -> None: + self._set_view("logs") + + def action_show_profiles(self) -> None: + self._set_view("profiles") + if not self._select_profile_row(self.active_profile_name): + self._select_profile_row("") + + def action_newest_request(self) -> None: + if self.requests: + self.render_detail(self.requests[0], kind="request") + + def action_focus_requests(self) -> None: + self._set_view("requests") + + def action_focus_logs(self) -> None: + self._set_view("logs") + + def action_focus_profiles(self) -> None: + self._set_view("profiles") + + def action_open_upstream(self) -> None: + row = self._selected_profile_row() + if not row: + return + upstream = str(row.get("upstream_base_url") or "").strip() + if not upstream or upstream == "-": + return + open_url(upstream) + + def action_probe_profile(self) -> None: + row = self._selected_profile_row() + if not row: + return + try: + status, payload = post_json( + build_api_url(gateway_base_url, "/api/profiles/probe"), + timeout, + {"profile": row["name"]}, + ) + self._set_status(f"probe {row['name']} -> {status} | {short_text(payload, 96)}") + except Exception as exc: + self._set_status(f"probe error: {exc}") + + def action_switch_profile(self) -> None: + row = self._selected_profile_row() + if not row: + return + try: + status, payload = post_json( + build_api_url(gateway_base_url, "/api/profiles/switch"), + timeout, + {"profile": row["name"]}, + ) + self._set_status(f"switch {row['name']} -> {status} | {short_text(payload, 96)}") + self.refresh_data(refresh=True) + except Exception as exc: + self._set_status(f"switch error: {exc}") + + def action_save_active_profile(self) -> None: + row = self._selected_profile_row() + if not row: + return + raw = row.get("raw") if isinstance(row.get("raw"), dict) else {} + form = raw.get("form") if isinstance(raw.get("form"), dict) else {} + payload = { + "name": row["name"], + "listen_host": form.get("listen_host") or row["listen_host"], + "listen_port": as_int(form.get("listen_port") or row["listen_port"]), + "upstream_base_url": form.get("upstream_base_url") or row["upstream_base_url"], + "upstream_auth_mode": form.get("auth_mode") or row["auth_mode"], + "upstream_auth_env": form.get("auth_env") or "", + "upstream_auth_file": form.get("auth_file") or "", + "upstream_auth_json_path": form.get("auth_json_path") or "", + "upstream_auth_json_key": form.get("auth_json_key") or "", + "request_history_limit": as_int(form.get("request_history_limit") or row["request_history_limit"]), + "model_remap": form.get("model_remap") or "", + "reasoning_equals": form.get("reasoning_equals") or "", + } + try: + status, result = post_json(build_api_url(gateway_base_url, "/api/profiles"), timeout, payload) + self._set_status(f"save profile -> {status} | {short_text(result, 96)}") + self.refresh_data(refresh=True) + except Exception as exc: + self._set_status(f"save error: {exc}") + + def action_delete_profile(self) -> None: + row = self._selected_profile_row() + if not row or row["active"]: + return + try: + status, payload = delete_json(build_api_url(gateway_base_url, f"/api/profiles/{urllib.parse.quote(row['name'])}"), timeout) + self._set_status(f"delete {row['name']} -> {status} | {short_text(payload, 96)}") + self.refresh_data(refresh=True) + except Exception as exc: + self._set_status(f"delete error: {exc}") + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id == "filter": + self.refresh_data(refresh=False) + + def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None: + if event.data_table.id == "requests_table": + key = str(event.row_key.value) + row = self.request_by_key.get(key) + if row: + self.render_detail(row, kind="request") + elif event.data_table.id == "logs_table": + key = str(event.row_key.value) + row = self.log_by_key.get(key) + if row: + self.render_detail(row, kind="log") + elif event.data_table.id == "profiles_table": + key = str(event.row_key.value) + row = self.profile_by_key.get(key) + if row: + self.render_detail(row, kind="profile") + + CodexRetryGatewayTui().run() + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Codex Retry Gateway TUI") + parser.add_argument("--api-url", default=default_api_url()) + parser.add_argument( + "--status-url", + default=default_status_url(), + help="optional status API URL; leave empty to infer from --api-url", + ) + parser.add_argument("--save-config", action="store_true", help="persist --api-url to ~/.config/codex-retry-gateway-tui/api-url before running") + parser.add_argument("--install", action="store_true", help="persist --api-url, install codex-retry-gateway-tui as a uv tool, then exit") + parser.add_argument("--version-check-url", default=os.environ.get("CODEX_RETRY_GATEWAY_TUI_VERSION_CHECK_URL", DEFAULT_VERSION_CHECK_URL)) + parser.add_argument( + "--version-check-timeout", + type=int, + default=env_int("CODEX_RETRY_GATEWAY_TUI_VERSION_CHECK_TIMEOUT", DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS), + ) + parser.add_argument( + "--no-version-check", + action="store_true", + default=os.environ.get("CODEX_RETRY_GATEWAY_TUI_NO_VERSION_CHECK", "").strip().lower() in {"1", "true", "yes", "on"}, + ) + parser.add_argument( + "--refresh-seconds", + type=int, + default=env_int("CODEX_RETRY_GATEWAY_TUI_REFRESH_SECONDS", DEFAULT_REFRESH_SECONDS), + ) + parser.add_argument("--timeout", type=int, default=env_int("CODEX_RETRY_GATEWAY_TUI_TIMEOUT", DEFAULT_TIMEOUT_SECONDS)) + parser.add_argument("--once", action="store_true", help="print one snapshot and exit") + parser.add_argument("--filter", default="", help="initial filter for one-shot text output") + return parser + + +def normalize_log_rows(payload: dict[str, Any]) -> list[dict[str, Any]]: + rows = [] + for entry in payload.get("entries") or []: + if not isinstance(entry, dict): + continue + rows.append( + { + "seq": as_int(entry.get("seq")), + "at": entry.get("at"), + "message": str(entry.get("message") or ""), + "raw": entry, + } + ) + rows.sort(key=lambda item: item["seq"], reverse=True) + return rows + + +def print_once(payload: dict[str, Any], filter_text: str = "", status_payload: dict[str, Any] | None = None, status_error: str = "") -> None: + print(summary_line(payload)) + print(state_summary(payload)) + if status_payload or status_error: + print(status_summary(status_payload or {}, status_error)) + print("requests") + for row in normalize_request_rows(payload, filter_text): + print( + f"{row['seq']:>6} {short_text(row['request_id'], 16):<16} {status_symbol(row)} {row.get('status_code') or '-':<4} " + f"{short_text(row['path'], 20):<20} {short_text(row['model'] or row['requested_model'] or row['forwarded_model'], 16):<16} " + f"{format_ms(row['duration_ms']):<8} {age_since(row.get('usage_last_updated_at')):<8} " + f"{format_count(row['upstream_attempt_count'] or 0):<4} {short_text(row['error'], 30)}" + ) + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + status_url = str(args.status_url or "").strip() or default_status_url() + if args.save_config or args.install: + config_path = write_api_url_config(args.api_url) + print(f"saved api url to {config_path}") + if args.install: + print("installing codex-retry-gateway-tui with uv tool...") + return run_install_command() + version_message = check_version_update( + args.version_check_url, + max(1, args.version_check_timeout), + disabled=bool(args.no_version_check), + ) + if args.once: + if version_message: + print(version_message) + snapshot = fetch_dashboard_snapshot( + args.api_url, + status_url, + args.timeout, + filter_text=args.filter, + current_view="overview", + ) + print_once(snapshot["payload"], args.filter, snapshot["status_payload"], snapshot["status_error"]) + return 0 + return run_textual(args.api_url, status_url, max(1, args.refresh_seconds), max(1, args.timeout), version_message) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6b1d674 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,31 @@ +[project] +name = "codex-retry-gateway-tui" +version = "0.1.0" +description = "Terminal UI for codex-retry-gateway monitoring and control" +readme = "README.md" +requires-python = ">=3.11" +license = "MIT" +license-files = ["LICENSE"] +authors = [ + { name = "Shujakuin" }, +] +keywords = ["codex", "gateway", "tui", "textual", "client"] +dependencies = [ + "textual>=0.89.1", +] + +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project.scripts] +codex-retry-gateway-tui = "codex_retry_gateway_tui:main" +codex-gateway-tui = "codex_retry_gateway_tui:main" + +[project.urls] +Homepage = "https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui" +Repository = "https://gitea.shujk.top/shujakuin/codex-retry-gateway-tui" + +[tool.setuptools] +py-modules = ["codex_retry_gateway_tui"] +include-package-data = true diff --git a/tests/test_payload.py b/tests/test_payload.py new file mode 100644 index 0000000..a71da6c --- /dev/null +++ b/tests/test_payload.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +import unittest + + +def load_module(): + module_path = Path(__file__).resolve().parents[1] / "codex_retry_gateway_tui.py" + spec = importlib.util.spec_from_file_location("codex_retry_gateway_tui", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +class CodexRetryGatewayTUITests(unittest.TestCase): + def test_summary_line_uses_api_snapshot(self) -> None: + mod = load_module() + payload = { + "listen": "127.0.0.1:4610", + "config": {"profile_name": "pc", "upstream_base_url": "https://example.com/v1"}, + "metrics": {"total_proxy_request_count": 11, "inspected_response_count": 7, "matched_response_count": 2, "reasoning_516_count": 1}, + } + self.assertIn("profile pc", mod.summary_line(payload)) + self.assertIn("req 11", mod.summary_line(payload)) + + def test_request_rows_sort_desc_and_keep_request_id(self) -> None: + mod = load_module() + payload = { + "entries": [ + {"seq": 1, "request_id": "req_a", "path": "/responses"}, + {"seq": 2, "request_id": "req_b", "path": "/v1/responses"}, + ] + } + rows = mod.normalize_request_rows(payload) + self.assertEqual([row["request_id"] for row in rows], ["req_b", "req_a"]) + + def test_profile_rows_show_active_first(self) -> None: + mod = load_module() + payload = { + "profiles": [ + {"name": "beta", "active": False, "summary": {"listen_host": "127.0.0.1", "listen_port": 4611, "upstream_base_url": "u1", "auth_mode": "passthrough", "auth_source": "passthrough", "request_history_limit": 10}}, + {"name": "alpha", "active": True, "summary": {"listen_host": "127.0.0.1", "listen_port": 4610, "upstream_base_url": "u0", "auth_mode": "manual_bearer", "auth_source": "manual_file", "request_history_limit": 20}}, + ] + } + rows = mod.normalize_profile_rows(payload) + self.assertEqual([row["name"] for row in rows], ["alpha", "beta"]) + + def test_version_update_message_only_for_newer_versions(self) -> None: + mod = load_module() + self.assertIn("0.1.0 -> 0.1.1", mod.version_update_message("0.1.1", "0.1.0")) + self.assertEqual(mod.version_update_message("0.1.0", "0.1.0"), "") + + def test_gateway_url_helpers_and_request_age(self) -> None: + mod = load_module() + self.assertEqual( + mod.normalize_gateway_url("http://127.0.0.1:4610/__codex_retry_gateway/api/status"), + "http://127.0.0.1:4610/__codex_retry_gateway", + ) + self.assertEqual( + mod.gateway_status_url("http://127.0.0.1:4610/__codex_retry_gateway"), + "http://127.0.0.1:4610/__codex_retry_gateway/api/status", + ) + self.assertIn("updated", mod.render_request_detail({"seq": 1, "request_id": "r", "method": "POST", "path": "/responses", "status_code": 200, "upstream_status_code": 200, "upstream_attempt_count": 1, "first_response_delay_ms": 10, "duration_ms": 20, "request_body_bytes": 3, "response_bytes_received": 4, "stream_chunk_count": 1, "started_at": "2026-06-30T00:00:00Z", "finished_at": "2026-06-30T00:00:01Z", "usage_last_updated_at": "2026-06-30T00:00:01Z"})) + + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..53c18a5 --- /dev/null +++ b/uv.lock @@ -0,0 +1,130 @@ +version = 1 +revision = 3 +requires-python = ">=3.11" + +[[package]] +name = "codex-retry-gateway-tui" +version = "0.1.0" +source = { editable = "." } +dependencies = [ + { name = "textual" }, +] + +[package.metadata] +requires-dist = [{ name = "textual", specifier = ">=0.89.1" }] + +[[package]] +name = "linkify-it-py" +version = "2.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "uc-micro-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/2e/c9/06ea13676ef354f0af6169587ae292d3e2406e212876a413bf9eece4eb23/linkify_it_py-2.1.0.tar.gz", hash = "sha256:43360231720999c10e9328dc3691160e27a718e280673d444c38d7d3aaa3b98b", size = 29158, upload-time = "2026-03-01T07:48:47.683Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b4/de/88b3be5c31b22333b3ca2f6ff1de4e863d8fe45aaea7485f591970ec1d3e/linkify_it_py-2.1.0-py3-none-any.whl", hash = "sha256:0d252c1594ecba2ecedc444053db5d3a9b7ec1b0dd929c8f1d74dce89f86c05e", size = 19878, upload-time = "2026-03-01T07:48:46.098Z" }, +] + +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[package.optional-dependencies] +linkify = [ + { name = "linkify-it-py" }, +] + +[[package]] +name = "mdit-py-plugins" +version = "0.6.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/59/fc/f8d0863f8862f25602c0404d75568e89fb6b4109804645e5cdfb1be5cf56/mdit_py_plugins-0.6.1.tar.gz", hash = "sha256:a2bca0f039f39dbd35fb74ae1b5f998608c437463371f0ff7f49a19a17a114d0", size = 56114, upload-time = "2026-05-13T09:03:38.91Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a5/69/6da5581c6a7fede7dc261bf4e67d6adca4196f176b43288b55b3db395b6e/mdit_py_plugins-0.6.1-py3-none-any.whl", hash = "sha256:214c82fb2ac524472ab6a5bcab1de80f73b50443e187f401bfd77efbc7c6481d", size = 66663, upload-time = "2026-05-13T09:03:37.76Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + +[[package]] +name = "platformdirs" +version = "4.10.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/47/e4501f49c178ae1d9f4a75073fda4204f52647993f075a9db4d14930e0c5/platformdirs-4.10.0.tar.gz", hash = "sha256:31e761a6a0ca04faf7353ea759bdba55652be214725111e5aac52dfa29d4bef7", size = 31224, upload-time = "2026-05-28T03:32:53.587Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/81/e6/cd9575ac904136b3cbf7aa7ee819ef86eedb7274e46f230e94ea4342e729/platformdirs-4.10.0-py3-none-any.whl", hash = "sha256:fb516cdb12eb0d857d0cd85a7c57cea4d060bee4578d6cf5a14dfdf8cbf8784a", size = 22743, upload-time = "2026-05-28T03:32:52.175Z" }, +] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + +[[package]] +name = "textual" +version = "8.2.7" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py", extra = ["linkify"] }, + { name = "mdit-py-plugins" }, + { name = "platformdirs" }, + { name = "pygments" }, + { name = "rich" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/9b/7a/c519db0aba5024f86e71e9631810bfdd6866ed2c8695bd7fa34b90e7ef59/textual-8.2.7.tar.gz", hash = "sha256:658f568ff81e30ed43890c3e07520390e5cf1b4763822006e060656b0a88f105", size = 1859249, upload-time = "2026-05-19T10:52:49.531Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a8/f5/c1e18bc0707300a0e90204343abbf7d7acd6fb7ebe03a6d4893b99a234b8/textual-8.2.7-py3-none-any.whl", hash = "sha256:4caaa13a90bc4cf9c6c862c067ccd34fe84e9c161710a2a907a8026313b6bd73", size = 731129, upload-time = "2026-05-19T10:52:51.773Z" }, +] + +[[package]] +name = "typing-extensions" +version = "4.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/94/1a15dd82efb362ac84269196e94cf00f187f7ed21c242792a923cdb1c61f/typing_extensions-4.15.0.tar.gz", hash = "sha256:0cea48d173cc12fa28ecabc3b837ea3cf6f38c6d1136f85cbaaf598984861466", size = 109391, upload-time = "2025-08-25T13:49:26.313Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/18/67/36e9267722cc04a6b9f15c7f3441c2363321a3ea07da7ae0c0707beb2a9c/typing_extensions-4.15.0-py3-none-any.whl", hash = "sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548", size = 44614, upload-time = "2025-08-25T13:49:24.86Z" }, +] + +[[package]] +name = "uc-micro-py" +version = "2.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/78/67/9a363818028526e2d4579334460df777115bdec1bb77c08f9db88f6389f2/uc_micro_py-2.0.0.tar.gz", hash = "sha256:c53691e495c8db60e16ffc4861a35469b0ba0821fe409a8a7a0a71864d33a811", size = 6611, upload-time = "2026-03-01T06:31:27.526Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/61/73/d21edf5b204d1467e06500080a50f79d49ef2b997c79123a536d4a17d97c/uc_micro_py-2.0.0-py3-none-any.whl", hash = "sha256:3603a3859af53e5a39bc7677713c78ea6589ff188d70f4fee165db88e22b242c", size = 6383, upload-time = "2026-03-01T06:31:26.257Z" }, +]