shujakuin/codex-retry-gateway-tui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
codex-retry-gateway-tui/codex_retry_gateway_tui.py
T

2930 lines
120 KiB
Python
Raw Permalink Blame History

#!/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 hashlib
import importlib.metadata
import json
import os
import re
import shlex
import shutil
import subprocess
import sys
import tempfile
import tomllib
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any
APP_NAME = "codex-retry-gateway-tui"
FALLBACK_VERSION = "0.1.10"
DEFAULT_GATEWAY_ADMIN_PATH = "/__codex_retry_gateway"
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_ACCESS_KEY_FILE = "~/.config/codex-retry-gateway-tui/access-key"
DEFAULT_REQUEST_TABLE_PREFERENCES_FILE = "~/.config/codex-retry-gateway-tui/request-table-columns.txt"
LEGACY_REQUEST_TABLE_PREFERENCES_FILE = "~/.config/codex-retry-gateway-tui/request-table.json"
DEFAULT_GATEWAY_STATE_FILE = "~/.codex-retry-gateway/state.json"
DEFAULT_GATEWAY_JSON_CONFIG_FILE = "~/.codex-retry-gateway/config/config.json"
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
DEFAULT_PROFILE_REASONING_EQUALS = [516, 1034, 1552]
DEFAULT_PROFILE_RETRYABLE_STATUS_CODES = [429, 503]
DEFAULT_PROFILE_RETRYABLE_ERROR_MESSAGES = [
"Selected model is at capacity. Please try a different model.",
"stream disconnected before completion: Concurrency limit exceeded for account, please retry later",
]
DEFAULT_PROFILE_UPSTREAM_FETCH_RETRY_ATTEMPTS = 5
DEFAULT_PROFILE_UPSTREAM_FETCH_RETRY_BACKOFF_MS = 350
DEFAULT_PROFILE_REQUEST_HISTORY_LIMIT = 0
REQUEST_STATUS_SLOW_AFTER_SECONDS = 20
REQUEST_STATUS_STALLED_AFTER_SECONDS = 120
DEFAULT_PROFILE_ENDPOINTS = [
"/responses",
"/chat/completions",
"/v1/responses",
"/v1/chat/completions",
]
REQUEST_TABLE_COLUMNS = (
{"key": "seq", "label": "Seq", "width_key": "seq"},
{"key": "req_id", "label": "Req ID", "width_key": "req_id"},
{"key": "resp_id", "label": "Resp ID", "width_key": "resp_id"},
{"key": "thread", "label": "Thread", "width_key": "thread"},
{"key": "started", "label": "Started", "width_key": "started"},
{"key": "status", "label": "Status", "width_key": "status"},
{"key": "path", "label": "Path", "width_key": "path"},
{"key": "model", "label": "Model", "width_key": "model"},
{"key": "effort", "label": "Effort", "width_key": "effort"},
{"key": "reasoning", "label": "R.Tok", "width_key": "reasoning"},
{"key": "in", "label": "In", "width_key": "token_col"},
{"key": "out", "label": "Out", "width_key": "token_col"},
{"key": "cache", "label": "Cache", "width_key": "token_col"},
{"key": "req_bytes", "label": "Req Size", "width_key": "req"},
{"key": "resp_bytes", "label": "Resp Size", "width_key": "resp"},
{"key": "chunks", "label": "Chunks", "width_key": "chunks"},
{"key": "first", "label": "First", "width_key": "first"},
{"key": "duration", "label": "Duration", "width_key": "duration"},
{"key": "updated", "label": "Updated", "width_key": "updated"},
{"key": "note", "label": "Note", "width_key": "note"},
{"key": "round", "label": "Round", "width_key": "round"},
)
REQUEST_TABLE_COLUMN_BY_KEY = {column["key"]: column for column in REQUEST_TABLE_COLUMNS}
REQUEST_TABLE_COLUMN_KEYS = tuple(column["key"] for column in REQUEST_TABLE_COLUMNS)
REQUEST_TABLE_COLUMN_KEY_SET = set(REQUEST_TABLE_COLUMN_KEYS)
REQUEST_TABLE_CELL_INDEX = {key: index for index, key in enumerate(REQUEST_TABLE_COLUMN_KEYS)}
DEFAULT_REQUEST_TABLE_VISIBLE_COLUMNS = REQUEST_TABLE_COLUMN_KEYS
DEFAULT_REQUEST_TABLE_SORT_COLUMN = "seq"
THREAD_COLOR_PALETTE = (
"#5fd7ff",
"#87d75f",
"#ffaf5f",
"#d787ff",
"#5fafff",
"#ff87d7",
"#d7af5f",
"#5fd7af",
"#af87ff",
"#87afd7",
)
THREAD_COLOR_COLUMN_KEYS = frozenset({"req_id", "resp_id", "thread"})
REQUEST_TABLE_WIDTH_PROFILES = {
"compact": {
"seq": 7,
"req_id": 18,
"resp_id": 18,
"thread": 18,
"started": 19,
"status": 12,
"path": 22,
"model": 16,
"effort": 8,
"reasoning": 8,
"token_col": 8,
"req": 10,
"resp": 10,
"chunks": 14,
"first": 12,
"duration": 8,
"updated": 8,
"note": 22,
"round": 10,
},
"wide": {
"seq": 7,
"req_id": 24,
"resp_id": 24,
"thread": 36,
"started": 19,
"status": 14,
"path": 40,
"model": 24,
"effort": 8,
"reasoning": 8,
"token_col": 8,
"req": 12,
"resp": 12,
"chunks": 16,
"first": 14,
"duration": 8,
"updated": 8,
"note": 48,
"round": 10,
},
}
REQUEST_TABLE_MIN_COLUMN_WIDTH = 4
REQUEST_TABLE_COLUMN_ALIASES = {
"seq": ("seq",),
"req_id": ("req id", "request id", "request_id"),
"resp_id": ("resp id", "response id", "response_id"),
"thread": ("thread", "thread id", "thread_id"),
"started": ("started", "start", "started at"),
"status": ("status", "code", "status code"),
"path": ("path", "endpoint"),
"model": ("model",),
"effort": ("effort", "reasoning effort"),
"reasoning": ("reasoning", "r tok", "rtok", "reasoning tokens", "reasoning_tokens"),
"in": ("in", "input", "input tokens", "input_tokens"),
"out": ("out", "output", "output tokens", "output_tokens"),
"cache": ("cache", "cached", "cached tokens", "cached_tokens"),
"req_bytes": ("req", "req size", "request size", "request bytes", "request_bytes"),
"resp_bytes": ("resp", "resp size", "response size", "response bytes", "response_bytes"),
"chunks": ("chunks", "stream chunks", "chunk count"),
"first": ("first", "first response", "first delay"),
"duration": ("duration", "elapsed"),
"updated": ("updated", "last update"),
"note": ("note", "retry note"),
"round": ("round", "retry round"),
}
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 read_json_object(path: str) -> dict[str, Any]:
try:
data = json.loads(Path(path).expanduser().read_text(encoding="utf-8"))
except Exception:
return {}
return data if isinstance(data, dict) else {}
def gateway_admin_url(base_url: str) -> str:
normalized = normalize_gateway_url(base_url)
if not normalized:
return ""
parsed = urllib.parse.urlparse(normalized)
path = parsed.path.rstrip("/")
if not path:
path = DEFAULT_GATEWAY_ADMIN_PATH
elif not path.endswith(DEFAULT_GATEWAY_ADMIN_PATH):
path = f"{path}{DEFAULT_GATEWAY_ADMIN_PATH}"
parsed = parsed._replace(path=path, params="", query="", fragment="")
return urllib.parse.urlunparse(parsed).rstrip("/")
def normalize_listen_host(value: Any) -> str:
host = str(value or "").strip()
if host in {"", "0.0.0.0", "::", "[::]", "*"}:
return "127.0.0.1"
return host
def discover_gateway_url() -> str:
state = read_json_object(DEFAULT_GATEWAY_STATE_FILE)
gateway_base_url = str(state.get("gateway_base_url") or "").strip()
if gateway_base_url:
return gateway_admin_url(gateway_base_url)
config = read_json_object(DEFAULT_GATEWAY_JSON_CONFIG_FILE)
listen_host = normalize_listen_host(config.get("listen_host"))
listen_port = as_int(config.get("listen_port"))
if listen_host and listen_port > 0:
return gateway_admin_url(f"http://{listen_host}:{listen_port}")
return ""
def default_api_url() -> str:
configured = configured_url(
("CODEX_RETRY_GATEWAY_TUI_API_URL",),
os.environ.get("CODEX_RETRY_GATEWAY_TUI_API_URL_FILE", DEFAULT_CONFIG_FILE),
"",
)
if configured:
return gateway_admin_url(configured)
discovered = discover_gateway_url()
if discovered:
return discovered
return gateway_admin_url(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 access_key_file_path() -> Path:
return Path(os.environ.get("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY_FILE", DEFAULT_ACCESS_KEY_FILE)).expanduser()
def request_table_preferences_file_path() -> Path:
return Path(
os.environ.get(
"CODEX_RETRY_GATEWAY_TUI_REQUEST_TABLE_PREFERENCES_FILE",
DEFAULT_REQUEST_TABLE_PREFERENCES_FILE,
)
).expanduser()
def legacy_request_table_preferences_file_path() -> Path:
return Path(LEGACY_REQUEST_TABLE_PREFERENCES_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 default_access_key() -> str:
return configured_url(
("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY",),
os.environ.get("CODEX_RETRY_GATEWAY_TUI_ACCESS_KEY_FILE", DEFAULT_ACCESS_KEY_FILE),
"",
)
def request_table_column_token(value: Any) -> str:
return re.sub(r"[^a-z0-9]+", "", str(value or "").strip().lower())
def build_request_table_column_token_map() -> dict[str, str]:
token_map: dict[str, str] = {}
for column in REQUEST_TABLE_COLUMNS:
key = column["key"]
aliases = set(REQUEST_TABLE_COLUMN_ALIASES.get(key, ()))
aliases.add(key)
aliases.add(column["label"])
for alias in aliases:
token = request_table_column_token(alias)
if token:
token_map[token] = key
return token_map
REQUEST_TABLE_COLUMN_TOKEN_MAP = build_request_table_column_token_map()
def request_table_column_key(value: Any) -> str:
token = request_table_column_token(value)
return REQUEST_TABLE_COLUMN_TOKEN_MAP.get(token, "")
def request_table_column_keys(value: Any) -> list[str]:
token = request_table_column_token(value)
if token == "usage":
return ["in", "out", "cache"]
column_key = REQUEST_TABLE_COLUMN_TOKEN_MAP.get(token, "")
return [column_key] if column_key else []
def request_table_column_order(value: Any) -> list[str]:
requested = value if isinstance(value, (list, tuple, set)) else []
columns = []
seen: set[str] = set()
for item in requested:
for column_key in request_table_column_keys(item):
if not column_key or column_key in seen:
continue
columns.append(column_key)
seen.add(column_key)
for column_key in REQUEST_TABLE_COLUMN_KEYS:
if column_key not in seen:
columns.append(column_key)
return columns or list(DEFAULT_REQUEST_TABLE_VISIBLE_COLUMNS)
def request_table_hidden_columns(value: Any, columns: list[str]) -> list[str]:
requested = value if isinstance(value, (list, tuple, set)) else []
allowed = set(columns)
hidden = []
seen: set[str] = set()
for item in requested:
for column_key in request_table_column_keys(item):
if not column_key or column_key not in allowed or column_key in seen:
continue
hidden.append(column_key)
seen.add(column_key)
return hidden
def request_table_column_width_value(value: Any) -> int | None:
try:
width = int(str(value or "").strip())
except Exception:
return None
if width <= 0:
return None
return max(REQUEST_TABLE_MIN_COLUMN_WIDTH, width)
def request_table_width_overrides(value: Any) -> dict[str, int]:
requested = value if isinstance(value, dict) else {}
overrides: dict[str, int] = {}
for raw_key, raw_width in requested.items():
width = request_table_column_width_value(raw_width)
if width is None:
continue
for column_key in request_table_column_keys(raw_key):
if column_key:
overrides[column_key] = width
return overrides
def normalize_request_table_sort_column(value: Any) -> str:
column_key = str(value or "").strip()
if column_key in REQUEST_TABLE_COLUMN_KEY_SET:
return column_key
return DEFAULT_REQUEST_TABLE_SORT_COLUMN
def normalize_request_table_preferences(data: Any) -> dict[str, Any]:
source = data if isinstance(data, dict) else {}
if isinstance(source.get("visible_columns"), (list, tuple, set)):
legacy_visible = []
seen: set[str] = set()
for item in source.get("visible_columns") or []:
for column_key in request_table_column_keys(item):
if not column_key or column_key in seen:
continue
legacy_visible.append(column_key)
seen.add(column_key)
columns = request_table_column_order(legacy_visible)
hidden_columns = [column_key for column_key in columns if column_key not in seen]
else:
columns = request_table_column_order(source.get("columns"))
hidden_columns = request_table_hidden_columns(source.get("hidden_columns"), columns)
if len(hidden_columns) >= len(columns) and columns:
hidden_columns = [column_key for column_key in hidden_columns if column_key != columns[0]]
width_overrides = request_table_width_overrides(source.get("widths"))
return {
"columns": columns,
"hidden_columns": hidden_columns,
"widths": {column_key: width for column_key, width in width_overrides.items() if column_key in columns},
}
def request_table_visible_columns(preferences: dict[str, Any]) -> list[str]:
normalized = normalize_request_table_preferences(preferences)
hidden = set(normalized["hidden_columns"])
visible = [column_key for column_key in normalized["columns"] if column_key not in hidden]
if visible:
return visible
return [normalized["columns"][0]]
def request_table_preferences_from_text(text: str) -> dict[str, Any]:
columns = []
hidden_columns = []
width_overrides: dict[str, int] = {}
seen: set[str] = set()
for line_number, raw_line in enumerate(text.splitlines(), start=1):
stripped = raw_line.strip()
if not stripped:
continue
hidden = False
token = stripped
if token.startswith("#"):
hidden = True
token = token[1:].strip()
if not token:
continue
width = None
width_match = re.match(r"^(.*?)(?:\s+(\d+))?$", token)
if width_match:
token = (width_match.group(1) or "").strip()
width = request_table_column_width_value(width_match.group(2))
column_keys = request_table_column_keys(token)
if not column_keys:
if hidden:
continue
raise ValueError(f"line {line_number}: unknown request column {token!r}")
for column_key in column_keys:
if column_key in seen:
continue
seen.add(column_key)
columns.append(column_key)
if hidden:
hidden_columns.append(column_key)
if width is not None:
width_overrides[column_key] = width
if not columns:
raise ValueError("no request columns found in editor document")
missing = [column_key for column_key in REQUEST_TABLE_COLUMN_KEYS if column_key not in seen]
return normalize_request_table_preferences(
{
"columns": columns + missing,
"hidden_columns": hidden_columns + missing,
"widths": width_overrides,
}
)
def request_table_preferences_document(preferences: dict[str, Any]) -> str:
normalized = normalize_request_table_preferences(preferences)
hidden = set(normalized["hidden_columns"])
widths = normalized["widths"]
lines = [
"# Request table columns.",
"# One column key per line.",
"# Prefix with # to hide a column.",
"# Optional: append an integer width, e.g. `first 14` or `in 10`.",
"# Move lines up or down to change display order.",
"# Common keys: in, out, cache, req_id, resp_id, req_bytes, resp_bytes, round.",
"",
]
for column_key in normalized["columns"]:
prefix = "# " if column_key in hidden else ""
width_suffix = f" {widths[column_key]}" if column_key in widths else ""
lines.append(f"{prefix}{column_key}{width_suffix}")
return "\n".join(lines).rstrip() + "\n"
def load_request_table_preferences() -> dict[str, Any]:
configured_path = request_table_preferences_file_path()
configured_env = os.environ.get("CODEX_RETRY_GATEWAY_TUI_REQUEST_TABLE_PREFERENCES_FILE", "").strip()
candidate_paths = [configured_path]
if not configured_env:
candidate_paths.append(legacy_request_table_preferences_file_path())
for path in candidate_paths:
try:
text = path.read_text(encoding="utf-8")
except OSError:
continue
stripped = text.strip()
if not stripped:
continue
try:
if stripped.startswith("{"):
return normalize_request_table_preferences(json.loads(text))
return request_table_preferences_from_text(text)
except Exception:
continue
return normalize_request_table_preferences({})
def write_request_table_preferences(preferences: dict[str, Any]) -> Path:
path = request_table_preferences_file_path()
path.parent.mkdir(parents=True, exist_ok=True)
try:
path.parent.chmod(0o700)
except OSError:
pass
normalized = normalize_request_table_preferences(preferences)
path.write_text(request_table_preferences_document(normalized), encoding="utf-8")
try:
path.chmod(0o600)
except OSError:
pass
return path
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/image-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 = gateway_admin_url(api_url)
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 write_access_key_config(access_key: str) -> Path:
path = access_key_file_path()
path.parent.mkdir(parents=True, exist_ok=True)
try:
path.parent.chmod(0o700)
except OSError:
pass
path.write_text(str(access_key or "").strip() + "\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_duration_ms_as_seconds(value: Any) -> str:
number = as_float(value)
if number <= 0:
return "-"
return format_elapsed_seconds(number / 1000.0)
def format_bytes(value: Any) -> str:
if value is None or str(value).strip() == "":
return "-"
number = as_int(value)
if number < 0:
return "-"
if number == 0:
return "0B"
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 cached_ratio(input_tokens: Any, cached_tokens: Any) -> float | None:
total = as_float(input_tokens)
if total <= 0:
return None
cached = max(0.0, as_float(cached_tokens))
return max(0.0, min(1.0, cached / total))
def effective_input_tokens(input_tokens: Any, cached_tokens: Any) -> int | None:
total = parse_int_value(input_tokens)
if total is None:
return None
cached = max(0, as_int(cached_tokens))
return max(0, total - cached)
def request_id_text(row: dict[str, Any]) -> str:
return str(row.get("request_id") or "").strip() or "-"
def response_id_text(row: dict[str, Any]) -> str:
return str(row.get("response_id") or "").strip() or "-"
def request_usage_summary(row: dict[str, Any]) -> str:
effective_in = effective_input_tokens(row.get("input_tokens"), row.get("cached_tokens"))
output_tokens = row.get("output_tokens")
cached_tokens = row.get("cached_tokens")
ratio = cached_ratio(row.get("input_tokens"), cached_tokens)
in_text = format_count(effective_in) if effective_in is not None else "-"
out_text = format_count(output_tokens) if output_tokens is not None else "-"
cached_text = format_count(cached_tokens) if cached_tokens is not None else "-"
if ratio is not None:
cached_text = f"{cached_text} ({format_percent(ratio * 100)})"
return f"in {in_text} | out {out_text} | cached {cached_text}"
def request_input_tokens_text(row: dict[str, Any]) -> str:
effective_in = effective_input_tokens(row.get("input_tokens"), row.get("cached_tokens"))
return format_count(effective_in) if effective_in is not None else "-"
def request_output_tokens_text(row: dict[str, Any]) -> str:
output_tokens = row.get("output_tokens")
return format_count(output_tokens) if output_tokens is not None else "-"
def request_cached_tokens_text(row: dict[str, Any]) -> str:
cached_tokens = row.get("cached_tokens")
return format_count(cached_tokens) if cached_tokens is not None else "-"
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 request_updated_at(row: dict[str, Any]) -> Any:
for key in ("last_activity_at", "usage_last_updated_at", "finished_at"):
value = row.get(key)
if parse_datetime(value):
return value
return None
def request_updated_elapsed(row: dict[str, Any]) -> str:
return elapsed_between(row.get("started_at"), request_updated_at(row))
def request_chunk_progress(row: dict[str, Any]) -> str:
if not row.get("response_stream"):
return "-"
return f"{row.get('stream_chunk_count') or 0} / {format_bytes(row.get('response_bytes_received'))}"
def request_effort_text(row: dict[str, Any]) -> str:
return str(row.get("reasoning_effort") or "").strip() or "-"
def request_model_text(row: dict[str, Any]) -> str:
return str(row.get("model") or row.get("requested_model") or row.get("forwarded_model") or "").strip()
def normalize_request_retry_firsts(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, (list, tuple)):
return []
firsts = []
for item in value:
if not isinstance(item, dict):
continue
firsts.append(
{
"round": parse_int_value(item.get("round")),
"slot": parse_int_value(item.get("slot")),
"first_response_at": item.get("first_response_at"),
"first_delay_ms": item.get("first_response_delay_ms", item.get("first_delay_ms", item.get("first_ms"))),
"outcome": str(item.get("outcome") or "").strip(),
"status_code": item.get("status_code", item.get("upstream_status_code", item.get("status"))),
"reasoning_effort": str(item.get("reasoning_effort") or "").strip(),
"reasoning_tokens": item.get("reasoning_tokens"),
"reasoning": str(item.get("reasoning") or item.get("reason") or item.get("error") or "").strip(),
}
)
return firsts
def request_retry_first_label(first: dict[str, Any], fallback_round: Any = None) -> str:
round_number = parse_int_value(first.get("round"))
if round_number is None:
round_number = parse_int_value(fallback_round)
slot = parse_int_value(first.get("slot"))
if round_number is not None:
if slot is not None and slot > 1:
return f"{round_number}-{slot}"
return str(round_number)
if slot is not None:
return str(slot)
return "?"
def request_retry_round_text(row: dict[str, Any]) -> str:
round_number = parse_int_value(row.get("reasoning_retry_current_round"))
width = parse_int_value(row.get("reasoning_retry_current_width"))
if round_number is None and width is None:
return "-"
if round_number is None:
return f"-({max(0, width or 0)})"
if width is None:
return str(round_number)
return f"{round_number}({max(0, width)})"
def request_retry_first_summary(first: dict[str, Any], fallback_round: Any = None) -> str:
bits = []
round_label = request_retry_first_label(first, fallback_round)
if round_label != "?":
bits.append(f"round {round_label}")
slot = parse_int_value(first.get("slot"))
if slot is not None:
bits.append(f"slot {slot}")
first_delay = format_duration_ms_as_seconds(first.get("first_delay_ms"))
if first_delay != "-":
bits.append(f"first {first_delay}")
outcome = str(first.get("outcome") or "").strip()
status_code = first.get("status_code")
status_bits = [bit for bit in (outcome, str(status_code).strip() if status_code not in (None, "") else "") if bit]
if status_bits:
bits.append(f"status {'/'.join(status_bits)}")
reasoning_bits = []
if first.get("reasoning_effort"):
reasoning_bits.append(str(first["reasoning_effort"]))
reasoning_tokens = parse_int_value(first.get("reasoning_tokens"))
if reasoning_tokens is not None:
reasoning_bits.append(format_count(reasoning_tokens))
reasoning = str(first.get("reasoning") or "").strip()
if reasoning_bits or reasoning:
detail = "/".join(reasoning_bits)
if reasoning:
detail = f"{detail} {short_text(reasoning, 40)}".strip()
bits.append(f"reasoning {detail}".strip())
return " ".join(bits)
def request_retry_wave_summary(row: dict[str, Any]) -> str:
summaries = []
fallback_round = row.get("reasoning_retry_current_round")
for first in row.get("reasoning_retry_current_firsts") or []:
if not isinstance(first, dict):
continue
summary = request_retry_first_summary(first, fallback_round)
if summary:
summaries.append(summary)
return "; ".join(summaries)
def request_retry_firsts_compact_text(row: dict[str, Any]) -> str:
parts = []
for first in row.get("reasoning_retry_current_firsts") or []:
if not isinstance(first, dict):
continue
delay = format_duration_ms_as_seconds(first.get("first_delay_ms"))
parts.append(delay if delay != "-" else "?")
return " ".join(parts)
def request_first_text(row: dict[str, Any]) -> str:
retry_firsts = request_retry_firsts_compact_text(row)
if retry_firsts:
return retry_firsts
return format_duration_ms_as_seconds(row["first_response_delay_ms"])
def request_retry_note(row: dict[str, Any]) -> str:
attempts = as_int(row.get("upstream_attempt_count"))
if attempts <= 1:
return ""
return str(row.get("error") or "").strip()
def request_reasoning_tokens_text(row: dict[str, Any]) -> str:
return format_count(row["reasoning_tokens"]) if row["reasoning_tokens"] is not None else "-"
def request_row_cells(row: dict[str, Any], text_limits: dict[str, int] | None = None) -> tuple[Any, ...]:
limits = text_limits or {}
note_width = max(4, limits.get("note", 22))
retry_note = request_retry_note(row)
return (
str(row["seq"]),
short_text(request_id_text(row), limits.get("req_id", 18)) or "-",
short_text(response_id_text(row), limits.get("resp_id", 18)) or "-",
short_text(row["thread_id"], limits.get("thread", 18)) or "-",
short_time(row["started_at"]),
request_status_label(row),
short_text(row["path"], limits.get("path", 22)),
short_text(request_model_text(row), limits.get("model", 16)),
short_text(request_effort_text(row), limits.get("effort", 10)),
request_reasoning_tokens_text(row),
request_input_tokens_text(row),
request_output_tokens_text(row),
request_cached_tokens_text(row),
format_bytes(row.get("request_body_bytes")),
format_bytes(row.get("response_bytes_received")),
request_chunk_progress(row),
short_text(request_first_text(row), limits.get("first", 10)),
format_duration_ms_as_seconds(row["duration_ms"]),
request_updated_elapsed(row),
short_text(retry_note, note_width),
short_text(request_retry_round_text(row), limits.get("round", 10)),
)
def request_thread_color(thread_id: Any) -> str | None:
text = str(thread_id or "").strip()
if not text:
return None
digest = hashlib.blake2b(text.encode("utf-8"), digest_size=4).digest()
color_index = int.from_bytes(digest[:2], "big") % len(THREAD_COLOR_PALETTE)
return THREAD_COLOR_PALETTE[color_index]
def request_visible_cells(
row: dict[str, Any],
cells: tuple[Any, ...],
visible_column_keys: list[str],
) -> list[Any]:
visible_cells = [cells[REQUEST_TABLE_CELL_INDEX[column_key]] for column_key in visible_column_keys]
try:
from rich.text import Text
except ImportError:
return visible_cells
thread_color = request_thread_color(row.get("thread_id"))
styled_cells: list[Any] = []
for column_key, cell in zip(visible_column_keys, visible_cells):
if column_key == "status":
styled_cells.append(Text(str(cell), style=request_status_style(row)))
elif thread_color and column_key in THREAD_COLOR_COLUMN_KEYS:
styled_cells.append(Text(str(cell), style=f"bold {thread_color}"))
else:
styled_cells.append(cell)
return styled_cells
def request_table_column_label(column_key: str) -> str:
column = REQUEST_TABLE_COLUMN_BY_KEY.get(column_key)
return str(column["label"]) if column else column_key
def request_sort_value(row: dict[str, Any], column_key: str) -> Any:
if column_key == "seq":
return parse_int_value(row.get("seq"))
if column_key == "req_id":
text = request_id_text(row)
return None if text == "-" else text.lower()
if column_key == "resp_id":
text = response_id_text(row)
return None if text == "-" else text.lower()
if column_key == "thread":
text = str(row.get("thread_id") or "").strip()
return text.lower() if text else None
if column_key == "started":
started = parse_datetime(row.get("started_at"))
return started.timestamp() if started else None
if column_key == "status":
return parse_int_value(row.get("status_code"))
if column_key == "path":
text = str(row.get("path") or "").strip()
return text.lower() if text else None
if column_key == "model":
text = request_model_text(row)
return text.lower() if text else None
if column_key == "effort":
text = str(row.get("reasoning_effort") or "").strip()
return text.lower() if text else None
if column_key == "reasoning":
return parse_int_value(row.get("reasoning_tokens"))
if column_key == "usage":
total_tokens = parse_int_value(row.get("total_tokens"))
if total_tokens is not None:
return total_tokens
input_tokens = parse_int_value(row.get("input_tokens")) or 0
output_tokens = parse_int_value(row.get("output_tokens")) or 0
cached_tokens = parse_int_value(row.get("cached_tokens")) or 0
combined = input_tokens + output_tokens + cached_tokens
return combined if combined else None
if column_key == "req_bytes":
return parse_int_value(row.get("request_body_bytes"))
if column_key == "resp_bytes":
return parse_int_value(row.get("response_bytes_received"))
if column_key == "chunks":
return parse_int_value(row.get("stream_chunk_count"))
if column_key == "first":
return parse_int_value(row.get("first_response_delay_ms"))
if column_key == "duration":
return parse_int_value(row.get("duration_ms"))
if column_key == "updated":
updated = parse_datetime(request_updated_at(row))
return updated.timestamp() if updated else None
if column_key == "note":
text = request_retry_note(row)
return text.lower() if text else None
if column_key == "round":
round_number = parse_int_value(row.get("reasoning_retry_current_round"))
width = parse_int_value(row.get("reasoning_retry_current_width"))
if round_number is None and width is None:
return None
return (round_number or 0, width or 0)
return None
def sort_request_rows(
rows: list[dict[str, Any]],
sort_column: str = DEFAULT_REQUEST_TABLE_SORT_COLUMN,
*,
reverse: bool = True,
) -> list[dict[str, Any]]:
column_key = normalize_request_table_sort_column(sort_column)
known: list[tuple[Any, dict[str, Any]]] = []
missing: list[dict[str, Any]] = []
for row in rows:
value = request_sort_value(row, column_key)
if value is None or value == "":
missing.append(row)
continue
known.append((value, row))
known.sort(key=lambda item: item[0], reverse=reverse)
return [row for _, row in known] + missing
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 request_headers(access_key: str = "", *, accept: str = "application/json") -> dict[str, str]:
headers = {"Accept": accept}
if str(access_key or "").strip():
headers["x-codex-retry-gateway-key"] = str(access_key).strip()
return headers
def fetch_payload(api_url: str, timeout: int, access_key: str = "") -> dict[str, Any]:
req = urllib.request.Request(api_url, headers=request_headers(access_key))
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,
access_key: str = "",
*,
allow_not_found: bool = False,
) -> tuple[dict[str, Any], str]:
if not str(url or "").strip():
return {}, ""
try:
return fetch_payload(url, timeout, access_key), ""
except urllib.error.HTTPError as exc:
if allow_not_found and exc.code == 404:
return {}, ""
return {}, str(exc)
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 parse_bool_value(value: Any, default: bool = False) -> bool:
if isinstance(value, bool):
return value
if value is None:
return default
text = str(value).strip().lower()
if text in {"1", "true", "yes", "on"}:
return True
if text in {"0", "false", "no", "off", ""}:
return False
return bool(value)
def parse_int_value(value: Any) -> int | None:
if value is None or str(value).strip() == "":
return None
try:
return int(str(value).strip())
except Exception:
try:
return int(float(str(value).strip()))
except Exception:
return None
def normalize_editor_integer_list(value: Any) -> list[int]:
if isinstance(value, (list, tuple, set)):
raw_values = list(value)
else:
raw_values = re.split(r"[\s,]+", str(value or ""))
numbers: list[int] = []
for item in raw_values:
parsed = parse_int_value(item)
if parsed is not None:
numbers.append(parsed)
return numbers
def normalize_editor_string_list(value: Any) -> list[str]:
if isinstance(value, (list, tuple, set)):
raw_values = list(value)
else:
raw_values = re.split(r"[\r\n,]+", str(value or ""))
values = []
for item in raw_values:
text = str(item or "").strip()
if text:
values.append(text)
return values
def normalize_editor_phrase_list(value: Any) -> list[str]:
if isinstance(value, (list, tuple, set)):
raw_values = list(value)
else:
raw_values = str(value or "").splitlines()
values = []
for item in raw_values:
text = str(item or "").strip()
if text:
values.append(text)
return values
def toml_string(value: Any) -> str:
return json.dumps(str(value or ""), ensure_ascii=False)
def toml_bool(value: Any) -> str:
return "true" if parse_bool_value(value) else "false"
def toml_int_list(values: Any) -> str:
numbers = normalize_editor_integer_list(values)
return "[" + ", ".join(str(value) for value in numbers) + "]"
def toml_string_list(values: Any) -> str:
items = [toml_string(value) for value in normalize_editor_string_list(values)]
return "[" + ", ".join(items) + "]"
def toml_phrase_list(values: Any) -> str:
items = [toml_string(value) for value in normalize_editor_phrase_list(values)]
return "[" + ", ".join(items) + "]"
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 "-"
image_active = config.get("image_profile_name") or state.get("image_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} | text {active} | image {image_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 '-'}",
f"image profiles {paths.get('image_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} | text {config.get('profile_name') or state.get('profile_name') or '-'} | image {config.get('image_profile_name') or state.get('image_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",
"response_id",
"thread_id",
"profile_name",
"method",
"path",
"model",
"requested_model",
"forwarded_model",
"reasoning_effort",
"lifecycle_state",
"discard_reason",
"error",
"upstream_origin",
"upstream_path",
"status_code",
"upstream_status_code",
"reasoning_tokens",
"input_tokens",
"output_tokens",
"total_tokens",
"cached_tokens",
"reasoning_retry_current_round",
"reasoning_retry_current_width",
)
).lower()
current_firsts = json.dumps(row.get("reasoning_retry_current_firsts") or [], ensure_ascii=False).lower()
haystack = f"{haystack} {current_firsts}"
return needle in haystack
def request_retry_first_received_at(row: dict[str, Any]) -> Any:
latest_value = None
latest_at = None
for first in row.get("reasoning_retry_current_firsts") or []:
if not isinstance(first, dict):
continue
value = first.get("first_response_at")
parsed = parse_datetime(value)
if parsed and (latest_at is None or parsed > latest_at):
latest_value = value
latest_at = parsed
return latest_value
def request_has_received_first(row: dict[str, Any]) -> bool:
if parse_datetime(row.get("first_response_at")):
return True
for first in row.get("reasoning_retry_current_firsts") or []:
if not isinstance(first, dict):
continue
if parse_datetime(first.get("first_response_at")) or parse_int_value(first.get("first_delay_ms")) is not None:
return True
return False
def request_lifecycle_label(row: dict[str, Any]) -> str:
state = str(row.get("lifecycle_state") or "").strip().lower()
if request_is_discarded(row):
return "discarded"
if state in {"finish", "finished", "complete", "completed"} or parse_datetime(row.get("finished_at")):
return "finished"
if state in {"receive_first", "streaming"} or request_has_received_first(row):
return "streaming"
return "waiting"
def request_is_discarded(row: dict[str, Any]) -> bool:
state = str(row.get("lifecycle_state") or "").strip().lower()
if bool(row.get("discarded")) or state in {"discarded", "abandoned", "cancelled", "canceled"}:
return True
error = str(row.get("error") or "").strip().lower()
return any(
marker in error
for marker in (
"client disconnected",
"client closed",
"cancelled by client",
"canceled by client",
"request aborted",
)
)
def request_activity_at(row: dict[str, Any]) -> Any:
candidates = [row.get("last_activity_at"), row.get("first_response_at"), request_retry_first_received_at(row), row.get("started_at")]
parsed_candidates = [(parse_datetime(value), value) for value in candidates]
valid_candidates = [(parsed, value) for parsed, value in parsed_candidates if parsed]
if not valid_candidates:
return None
return max(valid_candidates, key=lambda item: item[0])[1]
def request_activity_age_seconds(row: dict[str, Any], now: dt.datetime | None = None) -> float | None:
activity_at = parse_datetime(request_activity_at(row))
if not activity_at:
return None
current = now or dt.datetime.now(activity_at.tzinfo)
if current.tzinfo is None and activity_at.tzinfo is not None:
current = current.replace(tzinfo=activity_at.tzinfo)
return max(0.0, (current - activity_at).total_seconds())
def request_status_label(row: dict[str, Any], now: dt.datetime | None = None) -> str:
if request_is_discarded(row):
return "discarded"
status_code = parse_int_value(row.get("status_code"))
if status_code is not None:
return f"HTTP {status_code}"
lifecycle = request_lifecycle_label(row)
if lifecycle == "finished":
return "finished"
age_seconds = request_activity_age_seconds(row, now)
if age_seconds is not None and age_seconds >= REQUEST_STATUS_STALLED_AFTER_SECONDS:
return f"stalled {format_elapsed_seconds(age_seconds)}"
if age_seconds is not None and age_seconds >= REQUEST_STATUS_SLOW_AFTER_SECONDS:
return f"{lifecycle} {format_elapsed_seconds(age_seconds)}"
return lifecycle
def request_status_style(row: dict[str, Any], now: dt.datetime | None = None) -> str:
label = request_status_label(row, now).lower()
if label.startswith(("discarded", "stalled")):
return "bold #ff5f5f"
if label.startswith("waiting"):
return "bold #ffaf5f"
if label.startswith("streaming"):
return "bold #5fd7ff"
if label.startswith("http "):
status_code = parse_int_value(row.get("status_code")) or 0
if 200 <= status_code < 400:
return "bold #87d75f"
if 400 <= status_code < 500:
return "bold #ffaf5f"
return "bold #ff5f5f"
if row.get("error"):
return "bold #ff5f5f"
return "#b0b0b0"
def status_symbol(entry: dict[str, Any]) -> str:
"""Compatibility helper for callers that previously consumed a status marker."""
return request_status_label(entry)
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
usage = entry.get("usage") if isinstance(entry.get("usage"), dict) else {}
rows.append(
{
"seq": as_int(entry.get("seq")),
"request_id": str(entry.get("request_id") or ""),
"response_id": str(entry.get("response_id") or ""),
"thread_id": str(entry.get("thread_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 ""),
"reasoning_effort": str(entry.get("reasoning_effort") or ""),
"reasoning_summary": str(entry.get("reasoning_summary") or ""),
"request_stream": bool(entry.get("request_stream")),
"response_stream": bool(entry.get("response_stream")),
"lifecycle_state": str(entry.get("lifecycle_state") or ""),
"discarded": bool(entry.get("discarded")),
"discard_reason": str(entry.get("discard_reason") or ""),
"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", usage.get("reasoning_tokens")),
"input_tokens": entry.get("input_tokens", usage.get("input_tokens")),
"output_tokens": entry.get("output_tokens", usage.get("output_tokens")),
"total_tokens": entry.get("total_tokens", usage.get("total_tokens")),
"cached_tokens": entry.get("cached_tokens", usage.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_at": entry.get("first_response_at"),
"first_response_delay_ms": entry.get("first_response_delay_ms"),
"reasoning_retry_current_round": parse_int_value(entry.get("reasoning_retry_current_round")),
"reasoning_retry_current_width": parse_int_value(entry.get("reasoning_retry_current_width")),
"reasoning_retry_current_firsts": normalize_request_retry_firsts(entry.get("reasoning_retry_current_firsts")),
"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_route": str(entry.get("upstream")["route"] if isinstance(entry.get("upstream"), dict) else entry.get("upstream_route") 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"req {request_id_text(row)}",
f"resp {response_id_text(row)}",
f"thread {row.get('thread_id') or '-'}",
f"started {short_time(row['started_at'])}",
f"{row['method']} {row['path']}",
f"profile {row.get('profile_name') or '-'}",
f"status {request_status_label(row)}",
f"upstream {row['upstream_status_code'] or '-'}",
f"attempts {row['upstream_attempt_count'] or 0}",
f"first {request_first_text(row)}",
f"duration {format_duration_ms_as_seconds(row['duration_ms'])}",
f"request {format_bytes(row.get('request_body_bytes'))}",
f"response {format_bytes(row.get('response_bytes_received'))}",
f"chunks {row['stream_chunk_count'] or 0}",
f"life {request_lifecycle_label(row)}",
f"finished {short_time(row['finished_at'])}",
f"updated {request_updated_elapsed(row)}",
]
retry_round = request_retry_round_text(row)
if retry_round != "-":
bits.append(f"retry round {retry_round}")
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_effort"):
bits.append(f"effort {row['reasoning_effort']}")
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 or row.get("cached_tokens") is not None:
bits.append(f"usage {request_usage_summary(row)} | total {format_count(row.get('total_tokens'))}")
if row.get("cached_tokens") is not None:
bits.append(f"cached raw {format_count(row['cached_tokens'])}")
if row.get("error"):
bits.append(f"error {row['error']}")
if row.get("discard_reason"):
bits.append(f"discard {row['discard_reason']}")
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_route"):
bits.append(f"route {row['upstream_route']}")
if row.get("upstream_auth_mode"):
bits.append(f"auth {row['upstream_auth_mode']}/{row.get('upstream_auth_source') or '-'}")
retry_wave = request_retry_wave_summary(row)
if retry_wave:
bits.append(f"current wave {retry_wave}")
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 normalize_image_profile_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]:
needle = filter_text.strip().lower()
rows = []
for profile in payload.get("image_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 ""),
"base_url": str(summary.get("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 ""),
"manual_secret_file": str(summary.get("manual_secret_file") or ""),
"auth_json_path": str(summary.get("auth_json_path") or ""),
"auth_json_key": str(summary.get("auth_json_key") or "-"),
"auth_source": str(summary.get("auth_source") or "disabled"),
"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"text upstream {short_text(row['upstream_base_url'], 64)}",
f"text 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 render_image_profile_detail(row: dict[str, Any]) -> str:
bits = [
f"image {row['name']}",
"active" if row["active"] else "inactive",
f"upstream {short_text(row['base_url'], 64)}",
f"auth {row['auth_mode']}/{row['auth_source']}",
]
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 profile_form_state(row: dict[str, Any]) -> dict[str, Any]:
raw = row.get("raw") if isinstance(row.get("raw"), dict) else {}
form = raw.get("form") if isinstance(raw.get("form"), dict) else {}
return {
"name": row.get("name") or "",
"listen_host": form.get("listen_host") or row.get("listen_host") or "",
"listen_port": parse_int_value(form.get("listen_port") or row.get("listen_port")),
"upstream_base_url": form.get("upstream_base_url") or row.get("upstream_base_url") or "",
"auth_mode": form.get("auth_mode") or row.get("auth_mode") or "passthrough",
"auth_env": form.get("auth_env") or "",
"auth_file": form.get("auth_file") or "",
"manual_secret": "",
"manual_secret_file": form.get("manual_secret_file") or "",
"manual_secret_configured": parse_bool_value(form.get("manual_secret_configured"), False),
"auth_json_path": form.get("auth_json_path") or "",
"auth_json_key": form.get("auth_json_key") or row.get("auth_json_key") or "",
"request_history_limit": parse_int_value(form.get("request_history_limit") or row.get("request_history_limit"))
if parse_int_value(form.get("request_history_limit") or row.get("request_history_limit")) is not None
else DEFAULT_PROFILE_REQUEST_HISTORY_LIMIT,
"model_remap": form.get("model_remap") or row.get("model_remap") or "",
"reasoning_equals": normalize_editor_integer_list(form.get("reasoning_equals") or row.get("reasoning_equals"))
or list(DEFAULT_PROFILE_REASONING_EQUALS),
"retryable_status_codes": normalize_editor_integer_list(form.get("retryable_status_codes"))
or list(DEFAULT_PROFILE_RETRYABLE_STATUS_CODES),
"retryable_error_messages": normalize_editor_phrase_list(form.get("retryable_error_messages"))
or list(DEFAULT_PROFILE_RETRYABLE_ERROR_MESSAGES),
"upstream_fetch_retry_attempts": parse_int_value(form.get("upstream_fetch_retry_attempts"))
if parse_int_value(form.get("upstream_fetch_retry_attempts")) is not None
else DEFAULT_PROFILE_UPSTREAM_FETCH_RETRY_ATTEMPTS,
"upstream_fetch_retry_backoff_ms": parse_int_value(form.get("upstream_fetch_retry_backoff_ms"))
if parse_int_value(form.get("upstream_fetch_retry_backoff_ms")) is not None
else DEFAULT_PROFILE_UPSTREAM_FETCH_RETRY_BACKOFF_MS,
"endpoints": normalize_editor_string_list(form.get("endpoints")) or list(DEFAULT_PROFILE_ENDPOINTS),
}
def profile_payload_from_row(row: dict[str, Any]) -> dict[str, Any]:
state = profile_form_state(row)
return {
"name": str(state.get("name") or "").strip(),
"listen_host": str(state.get("listen_host") or "").strip(),
"listen_port": state.get("listen_port"),
"upstream_base_url": str(state.get("upstream_base_url") or "").strip(),
"auth_mode": str(state.get("auth_mode") or "passthrough").strip(),
"auth_env": str(state.get("auth_env") or "").strip(),
"auth_file": str(state.get("auth_file") or "").strip(),
"manual_secret": "",
"manual_secret_file": str(state.get("manual_secret_file") or "").strip(),
"manual_secret_configured": bool(state.get("manual_secret_configured")),
"auth_json_path": str(state.get("auth_json_path") or "").strip(),
"auth_json_key": str(state.get("auth_json_key") or "").strip(),
"request_history_limit": state.get("request_history_limit"),
"model_remap": str(state.get("model_remap") or "").strip(),
"reasoning_equals": state.get("reasoning_equals") or [],
"retryable_status_codes": state.get("retryable_status_codes") or [],
"retryable_error_messages": state.get("retryable_error_messages") or [],
"upstream_fetch_retry_attempts": state.get("upstream_fetch_retry_attempts"),
"upstream_fetch_retry_backoff_ms": state.get("upstream_fetch_retry_backoff_ms"),
"endpoints": state.get("endpoints") or [],
}
def profile_editor_document(row: dict[str, Any]) -> str:
state = profile_form_state(row)
model_remap = str(state.get("model_remap") or "")
if "\n" in model_remap:
model_remap_value = '"""\n' + model_remap.rstrip("\n").replace('"""', '\\"""') + '\n"""'
else:
model_remap_value = toml_string(model_remap)
return "\n".join(
[
"# Edit the selected codex-retry-gateway profile and save.",
"# Leave manual_secret empty to keep the current secret file.",
"# Changing name creates a new profile file; it does not delete the old one.",
"",
f"name = {toml_string(state.get('name'))}",
f"listen_host = {toml_string(state.get('listen_host'))}",
f"listen_port = {state.get('listen_port') if state.get('listen_port') is not None else 4610}",
f"upstream_base_url = {toml_string(state.get('upstream_base_url'))}",
f"auth_mode = {toml_string(state.get('auth_mode'))}",
f"auth_env = {toml_string(state.get('auth_env'))}",
f"auth_file = {toml_string(state.get('auth_file'))}",
'manual_secret = ""',
f"manual_secret_file = {toml_string(state.get('manual_secret_file'))}",
f"manual_secret_configured = {toml_bool(state.get('manual_secret_configured'))}",
f"auth_json_path = {toml_string(state.get('auth_json_path'))}",
f"auth_json_key = {toml_string(state.get('auth_json_key'))}",
"",
f"request_history_limit = {state.get('request_history_limit') if state.get('request_history_limit') is not None else 0}",
f"model_remap = {model_remap_value}",
f"reasoning_equals = {toml_int_list(state.get('reasoning_equals'))}",
f"retryable_status_codes = {toml_int_list(state.get('retryable_status_codes'))}",
f"retryable_error_messages = {toml_phrase_list(state.get('retryable_error_messages'))}",
f"upstream_fetch_retry_attempts = {state.get('upstream_fetch_retry_attempts') if state.get('upstream_fetch_retry_attempts') is not None else 5}",
f"upstream_fetch_retry_backoff_ms = {state.get('upstream_fetch_retry_backoff_ms') if state.get('upstream_fetch_retry_backoff_ms') is not None else 350}",
f"endpoints = {toml_string_list(state.get('endpoints'))}",
"",
]
)
def profile_payload_from_editor_text(text: str) -> dict[str, Any]:
data = tomllib.loads(text)
if not isinstance(data, dict):
raise ValueError("editor payload must be a TOML object")
return {
"name": str(data.get("name") or "").strip(),
"listen_host": str(data.get("listen_host") or "").strip(),
"listen_port": parse_int_value(data.get("listen_port")),
"upstream_base_url": str(data.get("upstream_base_url") or "").strip(),
"auth_mode": str(data.get("auth_mode") or "passthrough").strip(),
"auth_env": str(data.get("auth_env") or "").strip(),
"auth_file": str(data.get("auth_file") or "").strip(),
"manual_secret": str(data.get("manual_secret") or "").strip(),
"manual_secret_file": str(data.get("manual_secret_file") or "").strip(),
"manual_secret_configured": parse_bool_value(data.get("manual_secret_configured"), False),
"auth_json_path": str(data.get("auth_json_path") or "").strip(),
"auth_json_key": str(data.get("auth_json_key") or "").strip(),
"request_history_limit": parse_int_value(data.get("request_history_limit")),
"model_remap": str(data.get("model_remap") or "").strip(),
"reasoning_equals": normalize_editor_integer_list(data.get("reasoning_equals")),
"retryable_status_codes": normalize_editor_integer_list(data.get("retryable_status_codes")),
"retryable_error_messages": normalize_editor_phrase_list(data.get("retryable_error_messages")),
"upstream_fetch_retry_attempts": parse_int_value(data.get("upstream_fetch_retry_attempts")),
"upstream_fetch_retry_backoff_ms": parse_int_value(data.get("upstream_fetch_retry_backoff_ms")),
"endpoints": normalize_editor_string_list(data.get("endpoints")),
}
def resolve_editor_command() -> list[str]:
configured = (os.environ.get("VISUAL") or os.environ.get("EDITOR") or "").strip()
if configured:
return shlex.split(configured)
for candidate in ("nano", "vim", "vi"):
if shutil.which(candidate):
return [candidate]
raise RuntimeError("no editor found; set $VISUAL or $EDITOR")
def edit_profile_payload_with_editor(row: dict[str, Any]) -> dict[str, Any] | None:
original_text = profile_editor_document(row)
editor_command = resolve_editor_command()
fd, raw_path = tempfile.mkstemp(prefix=f"codex-retry-profile-{row.get('name') or 'profile'}-", suffix=".toml")
os.close(fd)
temp_path = Path(raw_path)
temp_path.write_text(original_text, encoding="utf-8")
try:
result = subprocess.run(editor_command + [str(temp_path)], check=False)
edited_text = temp_path.read_text(encoding="utf-8")
if edited_text == original_text:
return None
payload = profile_payload_from_editor_text(edited_text)
if result.returncode != 0:
raise RuntimeError(f"editor exited with status {result.returncode}")
return payload
except Exception as exc:
raise RuntimeError(f"{exc}; kept draft at {temp_path}") from exc
finally:
if temp_path.exists():
try:
if temp_path.read_text(encoding="utf-8") == original_text:
temp_path.unlink()
except Exception:
pass
def image_profile_form_state(row: dict[str, Any]) -> dict[str, Any]:
raw = row.get("raw") if isinstance(row.get("raw"), dict) else {}
form = raw.get("form") if isinstance(raw.get("form"), dict) else {}
return {
"name": row.get("name") or "",
"base_url": form.get("base_url") or row.get("base_url") or "",
"auth_mode": form.get("auth_mode") or row.get("auth_mode") or "fixed_bearer",
"auth_env": form.get("auth_env") or "CODEX_RETRY_GATEWAY_IMAGE_API_KEY",
"auth_file": form.get("auth_file") or "",
"manual_secret": "",
"manual_secret_file": form.get("manual_secret_file") or "",
"manual_secret_configured": parse_bool_value(form.get("manual_secret_configured"), False),
"auth_json_path": form.get("auth_json_path") or "",
"auth_json_key": form.get("auth_json_key") or "OPENAI_API_KEY",
}
def image_profile_payload_from_row(row: dict[str, Any]) -> dict[str, Any]:
state = image_profile_form_state(row)
return {
"name": str(state.get("name") or "").strip(),
"base_url": str(state.get("base_url") or "").strip(),
"auth_mode": str(state.get("auth_mode") or "fixed_bearer").strip(),
"auth_env": str(state.get("auth_env") or "").strip(),
"auth_file": str(state.get("auth_file") or "").strip(),
"manual_secret": "",
"manual_secret_file": str(state.get("manual_secret_file") or "").strip(),
"manual_secret_configured": bool(state.get("manual_secret_configured")),
"auth_json_path": str(state.get("auth_json_path") or "").strip(),
"auth_json_key": str(state.get("auth_json_key") or "").strip(),
}
def image_profile_editor_document(row: dict[str, Any]) -> str:
state = image_profile_form_state(row)
return "\n".join(
[
"# Edit the selected codex-retry-gateway image profile and save.",
"# base_url applies to both /images/* and /v1/images/*.",
"# Leave manual_secret empty to keep the current image secret file.",
"# Changing name creates a new image profile file; it does not delete the old one.",
"",
f"name = {toml_string(state.get('name'))}",
f"base_url = {toml_string(state.get('base_url'))}",
f"auth_mode = {toml_string(state.get('auth_mode'))}",
f"auth_env = {toml_string(state.get('auth_env'))}",
f"auth_file = {toml_string(state.get('auth_file'))}",
'manual_secret = ""',
f"manual_secret_file = {toml_string(state.get('manual_secret_file'))}",
f"manual_secret_configured = {toml_bool(state.get('manual_secret_configured'))}",
f"auth_json_path = {toml_string(state.get('auth_json_path'))}",
f"auth_json_key = {toml_string(state.get('auth_json_key'))}",
"",
]
)
def image_profile_payload_from_editor_text(text: str) -> dict[str, Any]:
data = tomllib.loads(text)
if not isinstance(data, dict):
raise ValueError("editor payload must be a TOML object")
return {
"name": str(data.get("name") or "").strip(),
"base_url": str(data.get("base_url") or "").strip(),
"auth_mode": str(data.get("auth_mode") or "fixed_bearer").strip(),
"auth_env": str(data.get("auth_env") or "").strip(),
"auth_file": str(data.get("auth_file") or "").strip(),
"manual_secret": str(data.get("manual_secret") or "").strip(),
"manual_secret_file": str(data.get("manual_secret_file") or "").strip(),
"manual_secret_configured": parse_bool_value(data.get("manual_secret_configured"), False),
"auth_json_path": str(data.get("auth_json_path") or "").strip(),
"auth_json_key": str(data.get("auth_json_key") or "").strip(),
}
def edit_image_profile_payload_with_editor(row: dict[str, Any]) -> dict[str, Any] | None:
original_text = image_profile_editor_document(row)
editor_command = resolve_editor_command()
fd, raw_path = tempfile.mkstemp(prefix=f"codex-retry-image-profile-{row.get('name') or 'profile'}-", suffix=".toml")
os.close(fd)
temp_path = Path(raw_path)
temp_path.write_text(original_text, encoding="utf-8")
try:
result = subprocess.run(editor_command + [str(temp_path)], check=False)
edited_text = temp_path.read_text(encoding="utf-8")
if edited_text == original_text:
return None
payload = image_profile_payload_from_editor_text(edited_text)
if result.returncode != 0:
raise RuntimeError(f"editor exited with status {result.returncode}")
return payload
except Exception as exc:
raise RuntimeError(f"{exc}; kept draft at {temp_path}") from exc
finally:
if temp_path.exists():
try:
if temp_path.read_text(encoding="utf-8") == original_text:
temp_path.unlink()
except Exception:
pass
def edit_request_table_preferences_with_editor(preferences: dict[str, Any]) -> dict[str, Any] | None:
original_text = request_table_preferences_document(preferences)
editor_command = resolve_editor_command()
fd, raw_path = tempfile.mkstemp(prefix="codex-retry-columns-", suffix=".txt")
os.close(fd)
temp_path = Path(raw_path)
temp_path.write_text(original_text, encoding="utf-8")
try:
result = subprocess.run(editor_command + [str(temp_path)], check=False)
edited_text = temp_path.read_text(encoding="utf-8")
if edited_text == original_text:
return None
preferences = request_table_preferences_from_text(edited_text)
if result.returncode != 0:
raise RuntimeError(f"editor exited with status {result.returncode}")
return preferences
except Exception as exc:
raise RuntimeError(f"{exc}; kept draft at {temp_path}") from exc
finally:
if temp_path.exists():
try:
if temp_path.read_text(encoding="utf-8") == original_text:
temp_path.unlink()
except Exception:
pass
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], access_key: str = "") -> tuple[int, dict[str, Any]]:
req = urllib.request.Request(
url,
data=json.dumps(payload).encode("utf-8"),
headers={**request_headers(access_key), "Content-Type": "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, access_key: str = "") -> tuple[int, dict[str, Any]]:
req = urllib.request.Request(url, headers=request_headers(access_key), 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 gateway_supports_image_profiles(status_payload: dict[str, Any]) -> bool:
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 {}
paths = status_payload.get("paths") if isinstance(status_payload.get("paths"), dict) else {}
return (
"image_profile_name" in config
or "image_profile_name" in state
or "image_profiles_dir" in paths
)
def fetch_dashboard_snapshot(
gateway_url: str,
status_url: str,
timeout: int,
*,
access_key: str = "",
filter_text: str = "",
current_view: str = "overview",
) -> dict[str, Any]:
gateway_root = gateway_admin_url(gateway_url)
status_endpoint = status_url or gateway_status_url(gateway_root)
status_payload, status_error = fetch_optional_payload(status_endpoint, timeout, access_key)
payload = status_payload if isinstance(status_payload, dict) else {}
endpoints = {
"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"),
}
image_profiles_supported = gateway_supports_image_profiles(payload)
if image_profiles_supported:
endpoints["image_profiles"] = build_api_url(gateway_root, "/api/image-profiles")
def fetch_optional(url: str, *, allow_not_found: bool = False) -> tuple[dict[str, Any], str]:
return fetch_optional_payload(url, timeout, access_key, allow_not_found=allow_not_found)
with ThreadPoolExecutor(max_workers=len(endpoints)) as pool:
futures = {
name: pool.submit(fetch_optional, url, allow_not_found=name == "image_profiles")
for name, url in endpoints.items()
}
logs_payload, logs_error = futures["logs"].result()
requests_payload, requests_error = futures["requests"].result()
profiles_payload, profiles_error = futures["profiles"].result()
if image_profiles_supported:
image_profiles_payload, image_profiles_error = futures["image_profiles"].result()
else:
image_profiles_payload, image_profiles_error = {}, ""
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 {}
image_profiles_source = image_profiles_payload if isinstance(image_profiles_payload, dict) else {}
errors = [message for message in (status_error, logs_error, requests_error, profiles_error, image_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 "",
),
"image_profiles": normalize_image_profile_rows(
image_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()
),
"active_image_profile_name": (
str(
payload.get("config", {}).get("image_profile_name")
or payload.get("state", {}).get("image_profile_name")
or image_profiles_source.get("active_image_profile")
or ""
).strip()
),
}
def run_textual(
api_url: str,
status_url: str,
refresh_seconds: int,
timeout: int,
version_message: str = "",
access_key: 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 = gateway_admin_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: 1; padding: 0 1; color: $text-muted; }
#tables { height: 1; padding: 0 1; color: $text-muted; }
#requests_table, #logs_table, #profiles_table { height: 1fr; }
#detail { height: 3; padding: 0 1; border-top: solid $panel; }
#status { height: 1; 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"),
("i", "toggle_profile_kind", "Text/Image"),
("m", "toggle_request_table_density", "Wide/Compact"),
("c", "edit_request_columns", "Columns"),
("shift+left", "scroll_table_left", "Scroll Left"),
("shift+right", "scroll_table_right", "Scroll Right"),
("e", "edit_profile", "Edit Profile"),
("p", "probe_profile", "Probe Profile"),
("s", "switch_profile", "Switch Profile"),
("w", "save_active_profile", "Save Profile"),
("d", "delete_profile", "Delete Profile"),
("u", "open_upstream", "Open Upstream"),
("/", "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.image_profiles: list[dict[str, Any]] = []
self.request_by_key: dict[str, dict[str, Any]] = {}
self.profile_by_key: dict[str, dict[str, Any]] = {}
self.image_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_key = ""
self.loading = False
self.executor = ThreadPoolExecutor(max_workers=1)
self.refresh_generation = 0
self.active_profile_name = ""
self.active_image_profile_name = ""
self.profile_kind = "text"
self.visible_tables = {
"overview": "requests",
"requests": "requests",
"logs": "logs",
"profiles": "profiles",
}
self.request_table_column_keys: dict[str, Any] = {}
self.request_table_preferences = load_request_table_preferences()
self.request_table_density = "compact"
self.request_table_width_profiles = {
density: dict(widths) for density, widths in REQUEST_TABLE_WIDTH_PROFILES.items()
}
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
self._rebuild_request_table_columns()
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
self._rebuild_profile_table_columns()
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 | i text/image | / filter | r refresh | m compact/wide | c columns | Shift-Left/Right scroll | e edit | 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:
visible_columns = request_table_visible_columns(self.request_table_preferences)
counts = (
f"requests {len(self.requests)} | logs {len(self.logs)} | text profiles {len(self.profiles)} | image profiles {len(self.image_profiles)} | active {self.profile_kind} | "
f"req cols {len(visible_columns)}/{len(REQUEST_TABLE_COLUMNS)} | c edit columns"
)
self.query_one("#tables", Static).update(counts)
def _current_filter(self) -> str:
return self.query_one("#filter", Input).value.strip()
def _visible_request_column_keys(self) -> list[str]:
return request_table_visible_columns(self.request_table_preferences)
def _request_table_actions_enabled(self) -> bool:
if self.visible_tables.get(self.current_view, "requests") == "requests":
return True
self._set_status("switch to requests or overview first")
return False
def _persist_request_table_preferences(self) -> bool:
self.request_table_preferences = normalize_request_table_preferences(self.request_table_preferences)
try:
write_request_table_preferences(self.request_table_preferences)
except Exception as exc:
self._set_status(f"save request table prefs error: {exc}")
return False
return True
def _request_table_column_width(self, column_key: str, density: str | None = None) -> int:
normalized = normalize_request_table_preferences(self.request_table_preferences)
override = request_table_column_width_value(normalized.get("widths", {}).get(column_key))
if override is not None:
return override
active_density = density or self.request_table_density
width_profile = self.request_table_width_profiles[active_density]
column = REQUEST_TABLE_COLUMN_BY_KEY[column_key]
return width_profile[column["width_key"]]
def _request_row_key(self, row: dict[str, Any]) -> str:
return f"{row['seq']}:{row['request_id']}"
def _selected_request_key(self) -> str:
row = self._selected_request_row()
if row:
return self._request_row_key(row)
return self.last_request_key
def _select_request_row(self, row_key: str) -> bool:
if not row_key:
return False
table = self.query_one("#requests_table", DataTable)
for index, row in enumerate(self.requests):
if self._request_row_key(row) == row_key:
return self._select_table_row(table, index)
return False
def _rebuild_request_table_columns(self) -> None:
table = self.query_one("#requests_table", DataTable)
selected_row_index = table.cursor_row if table.cursor_row is not None and table.cursor_row >= 0 else 0
selected_row_key = self._selected_request_key()
table.clear(columns=True)
self.request_table_column_keys = {}
for column_key in self._visible_request_column_keys():
column = REQUEST_TABLE_COLUMN_BY_KEY[column_key]
width = self._request_table_column_width(column_key)
self.request_table_column_keys[column_key] = table.add_column(
column["label"],
width=width,
key=column_key,
)
if self.requests:
self.render_requests()
if not self._select_request_row(selected_row_key):
self._select_table_row(table, min(selected_row_index, len(self.requests) - 1))
else:
self.request_by_key = {}
def _profile_rows(self) -> list[dict[str, Any]]:
return self.image_profiles if self.profile_kind == "image" else self.profiles
def _active_profile_name(self) -> str:
return self.active_image_profile_name if self.profile_kind == "image" else self.active_profile_name
def _profile_api_base(self) -> str:
return "/api/image-profiles" if self.profile_kind == "image" else "/api/profiles"
def _profile_kind_label(self) -> str:
return "image profile" if self.profile_kind == "image" else "text profile"
def _rebuild_profile_table_columns(self) -> None:
table = self.query_one("#profiles_table", DataTable)
table.clear(columns=True)
if self.profile_kind == "image":
table.add_columns("Name", "Active", "Image Upstream", "Image Auth", "Source")
else:
table.add_columns(
"Name",
"Active",
"Listen",
"Text Upstream",
"Text Auth",
"History",
"Reasoning",
"Source",
)
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)
rows = self._profile_rows()
if not rows:
return False
if profile_name:
for index, row in enumerate(rows):
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
if table.cursor_row >= len(self.requests):
return None
return self.requests[table.cursor_row]
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
rows = self._profile_rows()
if table.cursor_row >= len(rows):
return None
return rows[table.cursor_row]
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
if table.cursor_row >= len(self.logs):
return None
return self.logs[table.cursor_row]
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,
access_key=access_key,
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
selected_request_key = self._selected_request_key()
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.image_profiles = result["image_profiles"]
self.active_profile_name = result.get("active_profile_name") or ""
self.active_image_profile_name = result.get("active_image_profile_name") or ""
self.last_request_key = selected_request_key
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":
row = self._selected_request_row()
if row:
self.render_detail(row, kind="request")
else:
self.query_one("#detail", Static).update("no requests")
elif self.current_view == "logs":
row = self._selected_log_row()
if row:
self.render_detail(row, kind="log")
else:
self.query_one("#detail", Static).update("no logs")
elif self.current_view == "profiles":
row = self._selected_profile_row()
if row:
self.render_detail(row, kind="image_profile" if self.profile_kind == "image" else "profile")
else:
self.query_one("#detail", Static).update("no profiles")
def render_requests(self) -> None:
table = self.query_one("#requests_table", DataTable)
selected_row_index = table.cursor_row if table.cursor_row is not None and table.cursor_row >= 0 else 0
selected_row_key = self._selected_request_key()
text_limits = self._request_text_limits(table)
visible_column_keys = self._visible_request_column_keys()
table.clear()
self.request_by_key = {}
for row in self.requests:
key = self._request_row_key(row)
self.request_by_key[key] = row
cells = request_row_cells(row, text_limits)
visible_cells = request_visible_cells(row, cells, visible_column_keys)
table.add_row(*visible_cells, key=key)
if self.requests:
if not self._select_request_row(selected_row_key):
self._select_table_row(table, min(selected_row_index, len(self.requests) - 1))
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()
rows = self._profile_rows()
if self.profile_kind == "image":
self.image_profile_by_key = {}
else:
self.profile_by_key = {}
for row in rows:
key = row["name"]
if self.profile_kind == "image":
self.image_profile_by_key[key] = row
table.add_row(
row["name"],
"yes" if row["active"] else "",
short_text(row["base_url"], 36),
f"{row['auth_mode']}/{row['auth_source']}",
short_text(row["file_path"], 36),
key=key,
)
else:
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"], 28),
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"], 32),
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))
elif kind == "image_profile":
self.query_one("#detail", Static).update(render_image_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_toggle_profile_kind(self) -> None:
self.profile_kind = "image" if self.profile_kind == "text" else "text"
self._rebuild_profile_table_columns()
self.render_profiles()
if self.current_view == "profiles":
if not self._select_profile_row(self._active_profile_name()):
self._select_profile_row("")
row = self._selected_profile_row()
if row:
self.render_detail(row, kind="image_profile" if self.profile_kind == "image" else "profile")
self._update_table_headers()
self._set_status(f"profile view {self.profile_kind}")
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 _request_text_limits(self, table: DataTable) -> dict[str, int]:
width_profile = self.request_table_width_profiles[self.request_table_density]
limits = {
"req_id": width_profile["req_id"],
"resp_id": width_profile["resp_id"],
"thread": width_profile["thread"],
"path": width_profile["path"],
"model": width_profile["model"],
"effort": width_profile["effort"],
"first": width_profile["first"],
"note": width_profile["note"],
"round": width_profile["round"],
}
for name in ("req_id", "resp_id", "thread", "path", "model", "effort", "first", "note", "round"):
key = self.request_table_column_keys.get(name)
column = table.columns.get(key) if key is not None else None
if column is None or not getattr(column, "width", 0):
continue
limits[name] = int(column.width)
return limits
def _rerender_requests_after_layout_change(self) -> None:
self.render_requests()
self._update_table_headers()
if self.current_view not in {"overview", "requests"}:
return
row = self._selected_request_row()
if row:
self.render_detail(row, kind="request")
else:
self.query_one("#detail", Static).update("no requests")
def _apply_request_table_width_profile(self, density: str) -> None:
table = self.query_one("#requests_table", DataTable)
for name, key in self.request_table_column_keys.items():
column = table.columns.get(key)
if column is None:
continue
column.width = self._request_table_column_width(name, density)
self.request_table_density = density
self._rerender_requests_after_layout_change()
self._set_status(f"request table mode {density}")
def action_toggle_request_table_density(self) -> None:
next_density = "wide" if self.request_table_density == "compact" else "compact"
self._apply_request_table_width_profile(next_density)
def action_edit_request_columns(self) -> None:
try:
with self.suspend():
updated = edit_request_table_preferences_with_editor(self.request_table_preferences)
if updated is None:
self._set_status("request columns unchanged")
return
self.request_table_preferences = updated
if not self._persist_request_table_preferences():
return
self._rebuild_request_table_columns()
self._rerender_requests_after_layout_change()
self._set_status(
f"request columns saved -> {request_table_preferences_file_path()} | "
f"{len(self._visible_request_column_keys())}/{len(REQUEST_TABLE_COLUMNS)} shown"
)
except Exception as exc:
self._set_status(f"request columns error: {exc}")
def action_scroll_table_left(self) -> None:
table = self._current_table_widget()
if table:
table.action_scroll_left()
def action_scroll_table_right(self) -> None:
table = self._current_table_widget()
if table:
table.action_scroll_right()
def action_open_upstream(self) -> None:
row = self._selected_profile_row()
if not row:
return
upstream_value = row.get("base_url") if self.profile_kind == "image" else row.get("upstream_base_url")
upstream = str(upstream_value or "").strip()
if not upstream or upstream == "-":
return
open_url(upstream)
def action_edit_profile(self) -> None:
if self.current_view != "profiles":
self._set_status("switch to profiles view first")
return
row = self._selected_profile_row()
if not row:
self._set_status("no profile selected")
return
try:
with self.suspend():
payload = (
edit_image_profile_payload_with_editor(row)
if self.profile_kind == "image"
else edit_profile_payload_with_editor(row)
)
if payload is None:
self._set_status(f"edit {row['name']} cancelled (no changes)")
return
status, result = post_json(build_api_url(gateway_base_url, self._profile_api_base()), timeout, payload, access_key)
self._set_status(f"edit {payload.get('name') or row['name']} -> {status} | {short_text(result, 96)}")
self.refresh_data(refresh=True)
except Exception as exc:
self._set_status(f"edit error: {exc}")
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, f"{self._profile_api_base()}/probe"),
timeout,
{"profile": row["name"]},
access_key,
)
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, f"{self._profile_api_base()}/switch"),
timeout,
{"profile": row["name"]},
access_key,
)
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
payload = image_profile_payload_from_row(row) if self.profile_kind == "image" else profile_payload_from_row(row)
try:
status, result = post_json(build_api_url(gateway_base_url, self._profile_api_base()), timeout, payload, access_key)
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"{self._profile_api_base()}/{urllib.parse.quote(row['name'])}"),
timeout,
access_key,
)
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)
self.last_request_key = key
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.image_profile_by_key.get(key) if self.profile_kind == "image" else self.profile_by_key.get(key)
if row:
self.render_detail(row, kind="image_profile" if self.profile_kind == "image" else "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(
"--access-key",
default=default_access_key(),
help="optional management access key for protected gateway admin pages and APIs",
)
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(snapshot: dict[str, Any], filter_text: str = "") -> None:
payload = snapshot.get("payload") if isinstance(snapshot, dict) and isinstance(snapshot.get("payload"), dict) else {}
status_payload = snapshot.get("status_payload") if isinstance(snapshot, dict) and isinstance(snapshot.get("status_payload"), dict) else {}
status_error = str(snapshot.get("status_error") or "") if isinstance(snapshot, dict) else ""
print(summary_line(payload))
print(state_summary(payload))
if status_payload or status_error:
print(status_summary(status_payload, status_error))
print("requests")
for row in snapshot.get("requests") or []:
try:
print(
f"{row['seq']:>6} {short_text(request_id_text(row), 16):<16} {short_text(response_id_text(row), 16):<16} {short_text(row.get('thread_id'), 14):<14} {short_time(row.get('started_at')):<19} {short_text(request_status_label(row), 14):<14} "
f"{short_text(row['path'], 20):<20} {short_text(row['model'] or row['requested_model'] or row['forwarded_model'], 16):<16} {short_text(request_effort_text(row), 6):<6} {request_reasoning_tokens_text(row):<6} "
f"{short_text(request_usage_summary(row), 26):<26} {format_bytes(row.get('request_body_bytes')):<8} {request_chunk_progress(row):<14} {format_duration_ms_as_seconds(row['duration_ms']):<8} {request_updated_elapsed(row):<8} "
f"{short_text(request_retry_note(row), 30)}"
)
except BrokenPipeError:
return
def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
args.api_url = gateway_admin_url(args.api_url)
access_key = str(args.access_key or "").strip()
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 access_key:
key_path = write_access_key_config(access_key)
print(f"saved access key to {key_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,
access_key=access_key,
filter_text=args.filter,
current_view="overview",
)
print_once(snapshot, args.filter)
return 0
return run_textual(
args.api_url,
status_url,
max(1, args.refresh_seconds),
max(1, args.timeout),
version_message,
access_key,
)
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink