feat: merge cn+us ops errors page into shusub2

Add an errors screen that pulls admin ops error logs from sub2apicn and
the server4 fixed origin in parallel, refreshes every 60s while active,
and keeps the 0.2.3 key-usage panel and logs enhancements.

README.md

@@ -68,6 +68,13 @@ Environment variables override the config file:
 - `SHUSUB2_LOGS_TOKEN_FILE`
 - `SHUSUB2_LOGS_REFRESH_SECONDS`
 - `SHUSUB2_LOGS_LIMIT`
+- `SHUSUB2_ERRORS_CN_URL`
+- `SHUSUB2_ERRORS_CN_URL_FILE`
+- `SHUSUB2_ERRORS_US_URL`
+- `SHUSUB2_ERRORS_US_URL_FILE`
+- `SHUSUB2_ERRORS_REFRESH_SECONDS`
+- `SHUSUB2_ERRORS_LIMIT`
+- `SHUSUB2_ERRORS_TIME_RANGE`
 - `SHUSUB2_VERSION_CHECK_URL`
 - `SHUSUB2_VERSION_CHECK_TIMEOUT`
 - `SHUSUB2_NO_VERSION_CHECK`
@@ -117,6 +124,37 @@ never written anywhere else by the client. Without a token the logs page stays
 empty and shows a configuration hint; the accounts page keeps working without
 any secret.
 
+
+## Errors Page
+
+Press `e` from the accounts or logs page to open the merged errors page; press
+`a` / `l` to jump back. `shusub2 --errors` starts directly on the errors page,
+and `shusub2 --once --errors` prints one merged snapshot to stdout.
+
+The page pulls the latest admin ops error logs from **both** cn and us in
+parallel, then merges them by `created_at`:
+
+- cn default: `https://sub2apicn.shujk.top/api/v1/admin/ops/errors`
+- us default: `https://sub2api.server4.shujk.top:19857/api/v1/admin/ops/errors`
+
+`sub2apius.shujk.top` is the Cloudflare-proxied public name for the same us
+backend, but Cloudflare often returns 1010 for non-browser clients, so the TUI
+defaults to the fixed server4 origin while still labeling rows as `us`.
+
+While the errors page is active it refreshes every 60 seconds by default
+(`--errors-refresh-seconds` / `SHUSUB2_ERRORS_REFRESH_SECONDS`). Leaving the
+page stops that interval. Query window defaults to `24h`
+(`--errors-time-range` / `SHUSUB2_ERRORS_TIME_RANGE`); each source uses
+`page=1` and `page_size` from `--errors-limit` (default 100, max 500).
+
+Columns:
+
+```text
+Node | Status | Key | Account | Model | Phase | Type | Owner | Time
+```
+
+Auth reuses the same admin API key as the logs page.
+
 ## Local Development
 
 ```bash

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "shusub2"
-version = "0.2.3"
+version = "0.2.4"
 description = "Terminal UI for Sub2API account quota and daily usage"
 readme = "README.md"
 requires-python = ">=3.11"

sub2api_quota_tui.py

@@ -13,22 +13,32 @@ import sys
 import urllib.parse
 import urllib.request
 import zlib
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
 from typing import Any
 
 
 APP_NAME = "shusub2"
-FALLBACK_VERSION = "0.2.3"
+FALLBACK_VERSION = "0.2.4"
 DEFAULT_API_URL = "http://127.0.0.1:18318/api/tui/accounts"
 DEFAULT_CONFIG_FILE = "~/.config/shusub2/api-url"
 DEFAULT_STATUS_CONFIG_FILE = "~/.config/shusub2/status-url"
 DEFAULT_LOGS_API_URL = "https://sub2apicn.shujk.top/api/v1/admin/usage"
 DEFAULT_LOGS_URL_CONFIG_FILE = "~/.config/shusub2/logs-url"
 DEFAULT_LOGS_TOKEN_CONFIG_FILE = "~/.config/shusub2/logs-token"
+# sub2apius.shujk.top is Cloudflare-proxied and often returns 1010 for non-browser clients.
+# Use the fixed server4 origin for machine clients while still labeling the source as us.
+DEFAULT_ERRORS_CN_URL = "https://sub2apicn.shujk.top/api/v1/admin/ops/errors"
+DEFAULT_ERRORS_US_URL = "https://sub2api.server4.shujk.top:19857/api/v1/admin/ops/errors"
+DEFAULT_ERRORS_CN_URL_CONFIG_FILE = "~/.config/shusub2/errors-cn-url"
+DEFAULT_ERRORS_US_URL_CONFIG_FILE = "~/.config/shusub2/errors-us-url"
 DEFAULT_VERSION_CHECK_URL = "https://gitea.shujk.top/shujakuin/shusub2/raw/branch/main/pyproject.toml"
 DEFAULT_REFRESH_SECONDS = 60
 DEFAULT_LOGS_REFRESH_SECONDS = 60
+DEFAULT_ERRORS_REFRESH_SECONDS = 60
 DEFAULT_LOGS_LIMIT = 100
+DEFAULT_ERRORS_LIMIT = 100
+DEFAULT_ERRORS_TIME_RANGE = "24h"
 DEFAULT_TIMEOUT_SECONDS = 10
 DEFAULT_VERSION_CHECK_TIMEOUT_SECONDS = 2
 MONITOR_OK_STATUSES = {"operational", "ok", "success"}
@@ -103,6 +113,30 @@ def logs_token_config_file_path() -> Path:
     return Path(os.environ.get("SHUSUB2_LOGS_TOKEN_FILE", DEFAULT_LOGS_TOKEN_CONFIG_FILE)).expanduser()
 
 
+def default_errors_cn_url() -> str:
+    return configured_url(
+        ("SHUSUB2_ERRORS_CN_URL",),
+        os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE", DEFAULT_ERRORS_CN_URL_CONFIG_FILE),
+        DEFAULT_ERRORS_CN_URL,
+    )
+
+
+def default_errors_us_url() -> str:
+    return configured_url(
+        ("SHUSUB2_ERRORS_US_URL",),
+        os.environ.get("SHUSUB2_ERRORS_US_URL_FILE", DEFAULT_ERRORS_US_URL_CONFIG_FILE),
+        DEFAULT_ERRORS_US_URL,
+    )
+
+
+def errors_cn_url_config_file_path() -> Path:
+    return Path(os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE", DEFAULT_ERRORS_CN_URL_CONFIG_FILE)).expanduser()
+
+
+def errors_us_url_config_file_path() -> Path:
+    return Path(os.environ.get("SHUSUB2_ERRORS_US_URL_FILE", DEFAULT_ERRORS_US_URL_CONFIG_FILE)).expanduser()
+
+
 def write_config_value(path: Path, value: str) -> Path:
     text = str(value or "").strip()
     if not text:
@@ -597,6 +631,265 @@ def print_logs_once(payload: dict[str, Any], filter_text: str = "") -> None:
         )
 
 
+def errors_request_url(errors_url: str, limit: int, time_range: str = DEFAULT_ERRORS_TIME_RANGE) -> str:
+    parsed = urllib.parse.urlparse(errors_url)
+    query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
+    query = [(key, value) for key, value in query if key not in {"page", "page_size", "limit", "time_range"}]
+    page_size = max(1, min(int(limit), 500))
+    query.extend(
+        (
+            ("page", "1"),
+            ("page_size", str(page_size)),
+            ("time_range", str(time_range or DEFAULT_ERRORS_TIME_RANGE).strip() or DEFAULT_ERRORS_TIME_RANGE),
+        )
+    )
+    return urllib.parse.urlunparse(parsed._replace(query=urllib.parse.urlencode(query)))
+
+
+def fetch_errors_payload(
+    errors_url: str,
+    token: str,
+    timeout: int,
+    limit: int = DEFAULT_ERRORS_LIMIT,
+    time_range: str = DEFAULT_ERRORS_TIME_RANGE,
+) -> dict[str, Any]:
+    headers = {"Accept": "application/json"}
+    if str(token or "").strip():
+        headers["x-api-key"] = str(token).strip()
+    req = urllib.request.Request(errors_request_url(errors_url, limit, time_range), headers=headers)
+    with urllib.request.urlopen(req, timeout=timeout) as response:
+        data = json.loads(response.read().decode("utf-8"))
+    if not isinstance(data, dict):
+        raise RuntimeError("errors API did not return a JSON object")
+    return data
+
+
+def error_items(payload: dict[str, Any]) -> list[dict[str, Any]]:
+    data = payload.get("data")
+    if isinstance(data, dict) and isinstance(data.get("items"), list):
+        items = data.get("items")
+    elif isinstance(payload.get("items"), list):
+        items = payload.get("items")
+    else:
+        items = []
+    return [item for item in items if isinstance(item, dict)]
+
+
+def error_total(payload: dict[str, Any]) -> int:
+    data = payload.get("data")
+    if isinstance(data, dict) and data.get("total") is not None:
+        return as_int(data.get("total"))
+    if payload.get("total") is not None:
+        return as_int(payload.get("total"))
+    return len(error_items(payload))
+
+
+def default_error_sources(cn_url: str, us_url: str) -> list[tuple[str, str]]:
+    sources: list[tuple[str, str]] = []
+    if str(cn_url or "").strip():
+        sources.append(("cn", str(cn_url).strip()))
+    if str(us_url or "").strip():
+        sources.append(("us", str(us_url).strip()))
+    return sources
+
+
+def fetch_merged_errors(
+    sources: list[tuple[str, str]],
+    token: str,
+    timeout: int,
+    limit: int = DEFAULT_ERRORS_LIMIT,
+    time_range: str = DEFAULT_ERRORS_TIME_RANGE,
+) -> dict[str, Any]:
+    source_status: dict[str, dict[str, Any]] = {}
+    merged_items: list[dict[str, Any]] = []
+
+    def _one(node: str, url: str) -> tuple[str, str, dict[str, Any] | None, str]:
+        try:
+            return node, url, fetch_errors_payload(url, token, timeout, limit, time_range), ""
+        except Exception as exc:
+            return node, url, None, str(exc)
+
+    if not sources:
+        return {"items": [], "sources": {}, "time_range": time_range, "limit": limit}
+
+    with ThreadPoolExecutor(max_workers=max(1, len(sources))) as pool:
+        futures = [pool.submit(_one, node, url) for node, url in sources]
+        for future in as_completed(futures):
+            node, url, payload, error = future.result()
+            if payload is None:
+                source_status[node] = {"ok": False, "url": url, "total": 0, "fetched": 0, "error": error}
+                continue
+            items = error_items(payload)
+            for item in items:
+                row = dict(item)
+                row["_node"] = node
+                row["_source_url"] = url
+                merged_items.append(row)
+            source_status[node] = {
+                "ok": True,
+                "url": url,
+                "total": error_total(payload),
+                "fetched": len(items),
+                "error": "",
+            }
+
+    merged_items.sort(
+        key=lambda item: (str(item.get("created_at") or ""), as_int(item.get("id")), str(item.get("_node") or "")),
+        reverse=True,
+    )
+    return {
+        "items": merged_items,
+        "sources": source_status,
+        "time_range": time_range,
+        "limit": limit,
+        "generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
+    }
+
+
+def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list[dict[str, Any]]:
+    needle = filter_text.strip().lower()
+    rows = []
+    for item in payload.get("items") or []:
+        if not isinstance(item, dict):
+            continue
+        node = str(item.get("_node") or item.get("node") or "-").strip() or "-"
+        key_name = str(item.get("api_key_name") or "").strip()
+        if not key_name and item.get("api_key_id") is not None:
+            key_name = f"#{as_int(item.get('api_key_id'))}"
+        if not key_name:
+            key_name = "-"
+        account_name = str(item.get("account_name") or "").strip()
+        if not account_name and item.get("account_id") is not None:
+            account_name = f"#{as_int(item.get('account_id'))}"
+        if not account_name:
+            account_name = "-"
+        user_email = str(item.get("user_email") or "").strip()
+        if not user_email and item.get("user_id") is not None:
+            user_email = f"#{as_int(item.get('user_id'))}"
+        model = str(item.get("requested_model") or item.get("model") or "-").strip() or "-"
+        upstream_model = str(item.get("upstream_model") or "").strip()
+        phase = str(item.get("phase") or "-").strip() or "-"
+        error_type = str(item.get("type") or "-").strip() or "-"
+        owner = str(item.get("error_owner") or "-").strip() or "-"
+        source = str(item.get("error_source") or "-").strip() or "-"
+        platform = str(item.get("platform") or "-").strip() or "-"
+        status_code = as_int(item.get("status_code"))
+        message = str(item.get("message") or "").strip()
+        request_id = str(item.get("request_id") or "").strip()
+        client_request_id = str(item.get("client_request_id") or "").strip()
+        group_name = str(item.get("group_name") or "").strip() or "-"
+        row = {
+            "id": as_int(item.get("id")),
+            "node": node,
+            "status_code": status_code,
+            "key": key_name,
+            "account": account_name,
+            "user": user_email or "-",
+            "model": model,
+            "upstream_model": upstream_model,
+            "phase": phase,
+            "type": error_type,
+            "owner": owner,
+            "source": source,
+            "platform": platform,
+            "group": group_name,
+            "message": message,
+            "request_id": request_id,
+            "client_request_id": client_request_id,
+            "created_at": str(item.get("created_at") or ""),
+            "time": short_time(item.get("created_at")),
+            "resolved": bool(item.get("resolved")),
+            "raw": item,
+        }
+        if needle:
+            haystack = " ".join(
+                (
+                    node,
+                    key_name,
+                    account_name,
+                    user_email,
+                    model,
+                    upstream_model,
+                    phase,
+                    error_type,
+                    owner,
+                    source,
+                    platform,
+                    group_name,
+                    message,
+                    request_id,
+                    client_request_id,
+                    str(status_code),
+                )
+            ).lower()
+            if needle not in haystack:
+                continue
+        rows.append(row)
+    rows.sort(key=lambda row: (row["created_at"], row["id"], row["node"]), reverse=True)
+    return rows
+
+
+def errors_summary_line(payload: dict[str, Any], shown: int) -> str:
+    sources = payload.get("sources") if isinstance(payload.get("sources"), dict) else {}
+    source_bits = []
+    for node in sorted(sources.keys()):
+        info = sources.get(node) if isinstance(sources.get(node), dict) else {}
+        if info.get("ok"):
+            source_bits.append(f"{node} {as_int(info.get('fetched'))}/{as_int(info.get('total'))}")
+        else:
+            source_bits.append(f"{node} err")
+    fetched = short_time(payload.get("generated_at") or dt.datetime.now(dt.timezone.utc).isoformat())
+    time_range = str(payload.get("time_range") or DEFAULT_ERRORS_TIME_RANGE)
+    source_text = ", ".join(source_bits) if source_bits else "no sources"
+    return f"{fetched} errors | showing {shown}/{len(payload.get('items') or [])} | {time_range} | {source_text}"
+
+
+def error_detail_line(row: dict[str, Any]) -> str:
+    model = row["model"]
+    if row["upstream_model"] and row["upstream_model"] != row["model"]:
+        model = f"{row['model']} -> {row['upstream_model']}"
+    message = row["message"] or "-"
+    if len(message) > 180:
+        message = message[:177] + "..."
+    detail = (
+        f"{row['time']} | {row['node']} | status {row['status_code']} | key {row['key']} | account {row['account']} | "
+        f"user {row['user']} | {model} | {row['platform']} | phase {row['phase']} | type {row['type']} | "
+        f"owner {row['owner']} | source {row['source']} | group {row['group']} | {message}"
+    )
+    if row["request_id"]:
+        detail += f" | {row['request_id']}"
+    if row["client_request_id"] and row["client_request_id"] != row["request_id"]:
+        detail += f" | client {row['client_request_id']}"
+    if row["resolved"]:
+        detail += " | resolved"
+    return detail
+
+
+def print_errors_once(payload: dict[str, Any], filter_text: str = "") -> None:
+    rows = normalize_error_rows(payload, filter_text)
+    print(errors_summary_line(payload, len(rows)))
+    sources = payload.get("sources") if isinstance(payload.get("sources"), dict) else {}
+    for node in sorted(sources.keys()):
+        info = sources.get(node) if isinstance(sources.get(node), dict) else {}
+        if info.get("ok"):
+            print(f"{node}: ok fetched {as_int(info.get('fetched'))} total {as_int(info.get('total'))} | {info.get('url') or '-'}")
+        else:
+            print(f"{node}: error {info.get('error') or 'unknown'} | {info.get('url') or '-'}")
+    print("node  status  key                   account               model                     phase      type           owner     time")
+    for row in rows:
+        print(
+            f"{row['node']:<5} "
+            f"{row['status_code']:<7} "
+            f"{row['key'][:20]:<21} "
+            f"{row['account'][:20]:<21} "
+            f"{row['model'][:24]:<25} "
+            f"{row['phase'][:9]:<10} "
+            f"{row['type'][:13]:<14} "
+            f"{row['owner'][:8]:<9} "
+            f"{row['time']}"
+        )
+
+
 def kind_label(kind: Any) -> str:
     return {
         "quota_limited": "quota",
@@ -869,10 +1162,15 @@ def run_textual(
     status_url: str,
     logs_url: str,
     logs_token: str,
+    errors_cn_url: str,
+    errors_us_url: str,
     refresh_seconds: int,
     logs_refresh_seconds: int,
+    errors_refresh_seconds: int,
     timeout: int,
     logs_limit: int,
+    errors_limit: int,
+    errors_time_range: str,
     version_message: str = "",
     start_page: str = "accounts",
 ) -> int:
@@ -891,6 +1189,7 @@ def run_textual(
             ("r", "refresh", "Refresh"),
             ("/", "focus_filter", "Filter"),
             ("l", "show_logs", "Logs"),
+            ("e", "show_errors", "Errors"),
         ]
 
         def __init__(self) -> None:
@@ -934,6 +1233,9 @@ def run_textual(
         def action_show_logs(self) -> None:
             self.app.switch_screen(LogsScreen())
 
+        def action_show_errors(self) -> None:
+            self.app.switch_screen(ErrorsScreen())
+
         def on_input_changed(self, event: Input.Changed) -> None:
             if event.input.id == "filter":
                 self.render_payload()
@@ -1040,6 +1342,7 @@ def run_textual(
             ("r", "refresh", "Refresh"),
             ("/", "focus_filter", "Filter"),
             ("a", "show_accounts", "Accounts"),
+            ("e", "show_errors", "Errors"),
         ]
 
         def __init__(self) -> None:
@@ -1074,6 +1377,9 @@ def run_textual(
         def action_show_accounts(self) -> None:
             self.app.switch_screen(AccountsScreen())
 
+        def action_show_errors(self) -> None:
+            self.app.switch_screen(ErrorsScreen())
+
         def on_input_changed(self, event: Input.Changed) -> None:
             if event.input.id == "filter":
                 self.render_payload()
@@ -1132,6 +1438,121 @@ def run_textual(
             if row:
                 self.query_one("#detail", Static).update(log_detail_line(row))
 
+    class ErrorsScreen(Screen[None]):
+        AUTO_FOCUS = "#errors"
+        BINDINGS = [
+            ("r", "refresh", "Refresh"),
+            ("/", "focus_filter", "Filter"),
+            ("a", "show_accounts", "Accounts"),
+            ("l", "show_logs", "Logs"),
+        ]
+
+        def __init__(self) -> None:
+            super().__init__()
+            self.payload: dict[str, Any] = {}
+            self.rows: list[dict[str, Any]] = []
+            self.row_by_key: dict[str, dict[str, Any]] = {}
+
+        def compose(self) -> ComposeResult:
+            yield Header(show_clock=True)
+            yield Static("", id="summary")
+            yield Input(placeholder="filter", id="filter")
+            yield DataTable(id="errors")
+            yield Static("", id="detail")
+            yield Static("", id="status")
+            yield Footer()
+
+        def on_mount(self) -> None:
+            table = self.query_one("#errors", DataTable)
+            table.cursor_type = "row"
+            table.zebra_stripes = True
+            table.add_columns("Node", "Status", "Key", "Account", "Model", "Phase", "Type", "Owner", "Time")
+            self.refresh_data()
+            self.set_interval(errors_refresh_seconds, self.refresh_data)
+
+        def action_refresh(self) -> None:
+            self.refresh_data()
+
+        def action_focus_filter(self) -> None:
+            self.query_one("#filter", Input).focus()
+
+        def action_show_accounts(self) -> None:
+            self.app.switch_screen(AccountsScreen())
+
+        def action_show_logs(self) -> None:
+            self.app.switch_screen(LogsScreen())
+
+        def on_input_changed(self, event: Input.Changed) -> None:
+            if event.input.id == "filter":
+                self.render_payload()
+
+        def on_input_submitted(self, event: Input.Submitted) -> None:
+            if event.input.id == "filter":
+                self.query_one("#errors", DataTable).focus()
+
+        def refresh_data(self) -> None:
+            status = self.query_one("#status", Static)
+            if not str(logs_token or "").strip():
+                status.update(logs_token_hint())
+                return
+            status.update("refreshing errors from cn + us...")
+            try:
+                self.payload = fetch_merged_errors(
+                    default_error_sources(errors_cn_url, errors_us_url),
+                    logs_token,
+                    timeout,
+                    errors_limit,
+                    errors_time_range,
+                )
+                self.render_payload()
+                sources = self.payload.get("sources") if isinstance(self.payload.get("sources"), dict) else {}
+                bits = []
+                for node in ("cn", "us"):
+                    info = sources.get(node) if isinstance(sources.get(node), dict) else {}
+                    if not info:
+                        continue
+                    if info.get("ok"):
+                        bits.append(f"{node} ok")
+                    else:
+                        bits.append(f"{node} {info.get('error') or 'error'}")
+                status_bits = [version_message, f"every {errors_refresh_seconds}s", " | ".join(bits)]
+                status.update(" | ".join(bit for bit in status_bits if bit))
+            except Exception as exc:
+                status.update(f"errors error: {exc}")
+
+        def render_payload(self) -> None:
+            filter_text = self.query_one("#filter", Input).value
+            self.rows = normalize_error_rows(self.payload, filter_text)
+            table = self.query_one("#errors", DataTable)
+            table.clear()
+            self.row_by_key = {}
+            for index, row in enumerate(self.rows):
+                key = f"{row['node']}-{row['id']}-{index}"
+                self.row_by_key[key] = row
+                table.add_row(
+                    row["node"],
+                    str(row["status_code"]),
+                    row["key"],
+                    row["account"],
+                    row["model"],
+                    row["phase"],
+                    row["type"],
+                    row["owner"],
+                    row["time"],
+                    key=key,
+                )
+            self.query_one("#summary", Static).update(errors_summary_line(self.payload, len(self.rows)))
+            if self.rows:
+                self.query_one("#detail", Static).update(error_detail_line(self.rows[0]))
+            else:
+                self.query_one("#detail", Static).update("no error logs")
+
+        def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
+            key = str(event.row_key.value)
+            row = self.row_by_key.get(key)
+            if row:
+                self.query_one("#detail", Static).update(error_detail_line(row))
+
     class Sub2APIQuotaApp(App[None]):
         CSS = """
         #summary { height: 1; padding: 0 1; color: $accent; }
@@ -1139,6 +1560,7 @@ def run_textual(
         #accounts { height: 2fr; }
         #keys { height: 1fr; border-top: solid $panel; }
         #logs { height: 1fr; }
+        #errors { height: 1fr; }
         #detail { height: 3; padding: 0 1; border-top: solid $panel; }
         #status { height: 1; padding: 0 1; color: $text-muted; }
         """
@@ -1147,7 +1569,12 @@ def run_textual(
         ]
 
         def on_mount(self) -> None:
-            self.push_screen(LogsScreen() if start_page == "logs" else AccountsScreen())
+            if start_page == "logs":
+                self.push_screen(LogsScreen())
+            elif start_page == "errors":
+                self.push_screen(ErrorsScreen())
+            else:
+                self.push_screen(AccountsScreen())
 
     Sub2APIQuotaApp().run()
     return 0
@@ -1165,8 +1592,14 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--api-url", default=default_api_url())
     parser.add_argument("--status-url", default=default_status_url(), help="optional sub2api-status /api/status URL for channel monitor health")
     parser.add_argument("--logs-url", default=default_logs_url(), help="Sub2API admin usage logs URL (default: sub2apicn /api/v1/admin/usage)")
-    parser.add_argument("--logs-token", default=default_logs_token(), help="Sub2API admin API key for the logs page (prefer SHUSUB2_LOGS_TOKEN or ~/.config/shusub2/logs-token)")
-    parser.add_argument("--save-config", action="store_true", help="persist --api-url/--logs-url/--logs-token to ~/.config/shusub2/ before running")
+    parser.add_argument("--logs-token", default=default_logs_token(), help="Sub2API admin API key for logs/errors pages (prefer SHUSUB2_LOGS_TOKEN or ~/.config/shusub2/logs-token)")
+    parser.add_argument("--errors-cn-url", default=default_errors_cn_url(), help="cn Sub2API admin ops errors URL (default: sub2apicn /api/v1/admin/ops/errors)")
+    parser.add_argument(
+        "--errors-us-url",
+        default=default_errors_us_url(),
+        help="us Sub2API admin ops errors URL (default: server4 fixed origin /api/v1/admin/ops/errors; CF sub2apius often blocks non-browser clients)",
+    )
+    parser.add_argument("--save-config", action="store_true", help="persist --api-url/--logs-url/--logs-token/--errors-*-url to ~/.config/shusub2/ before running")
     parser.add_argument("--install", action="store_true", help="persist config, install shusub2 as a uv tool, then exit")
     parser.add_argument("--version-check-url", default=os.environ.get("SHUSUB2_VERSION_CHECK_URL", DEFAULT_VERSION_CHECK_URL))
     parser.add_argument(
@@ -1185,14 +1618,30 @@ def build_parser() -> argparse.ArgumentParser:
         type=int,
         default=env_int("SHUSUB2_LOGS_REFRESH_SECONDS", DEFAULT_LOGS_REFRESH_SECONDS),
     )
+    parser.add_argument(
+        "--errors-refresh-seconds",
+        type=int,
+        default=env_int("SHUSUB2_ERRORS_REFRESH_SECONDS", DEFAULT_ERRORS_REFRESH_SECONDS),
+    )
     parser.add_argument(
         "--logs-limit",
         type=int,
         default=env_int("SHUSUB2_LOGS_LIMIT", DEFAULT_LOGS_LIMIT),
     )
+    parser.add_argument(
+        "--errors-limit",
+        type=int,
+        default=env_int("SHUSUB2_ERRORS_LIMIT", DEFAULT_ERRORS_LIMIT),
+    )
+    parser.add_argument(
+        "--errors-time-range",
+        default=os.environ.get("SHUSUB2_ERRORS_TIME_RANGE", DEFAULT_ERRORS_TIME_RANGE),
+        help="ops errors time_range query (5m/30m/1h/6h/24h/7d/30d, default 24h)",
+    )
     parser.add_argument("--timeout", type=int, default=env_int("SUB2API_QUOTA_TUI_TIMEOUT", DEFAULT_TIMEOUT_SECONDS))
     parser.add_argument("--once", action="store_true", help="print one snapshot and exit")
     parser.add_argument("--logs", action="store_true", help="start on the request logs page (with --once: print a logs snapshot)")
+    parser.add_argument("--errors", action="store_true", help="start on the merged cn+us errors page (with --once: print an errors snapshot)")
     parser.add_argument("--filter", default="", help="initial filter for --once output")
     return parser
 
@@ -1201,6 +1650,11 @@ def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     status_url = str(args.status_url or "").strip() or inferred_status_url(args.api_url)
     logs_limit = max(1, min(args.logs_limit, 1000))
+    errors_limit = max(1, min(args.errors_limit, 500))
+    errors_time_range = str(args.errors_time_range or DEFAULT_ERRORS_TIME_RANGE).strip() or DEFAULT_ERRORS_TIME_RANGE
+    if args.logs and args.errors:
+        print("choose only one of --logs or --errors", file=sys.stderr)
+        return 2
     if args.save_config or args.install:
         config_path = write_api_url_config(args.api_url)
         print(f"saved api url to {config_path}")
@@ -1210,6 +1664,12 @@ def main(argv: list[str] | None = None) -> int:
         if str(args.logs_token or "").strip():
             logs_token_path = write_config_value(logs_token_config_file_path(), args.logs_token)
             print(f"saved logs token to {logs_token_path}")
+        if str(args.errors_cn_url or "").strip():
+            errors_cn_path = write_config_value(errors_cn_url_config_file_path(), args.errors_cn_url)
+            print(f"saved errors cn url to {errors_cn_path}")
+        if str(args.errors_us_url or "").strip():
+            errors_us_path = write_config_value(errors_us_url_config_file_path(), args.errors_us_url)
+            print(f"saved errors us url to {errors_us_path}")
     if args.install:
         print("installing shusub2 with uv tool...")
         return run_install_command()
@@ -1227,6 +1687,21 @@ def main(argv: list[str] | None = None) -> int:
                 return 2
             print_logs_once(fetch_logs_payload(args.logs_url, args.logs_token, args.timeout, logs_limit), args.filter)
             return 0
+        if args.errors:
+            if not str(args.logs_token or "").strip():
+                print(logs_token_hint(), file=sys.stderr)
+                return 2
+            print_errors_once(
+                fetch_merged_errors(
+                    default_error_sources(args.errors_cn_url, args.errors_us_url),
+                    args.logs_token,
+                    args.timeout,
+                    errors_limit,
+                    errors_time_range,
+                ),
+                args.filter,
+            )
+            return 0
         status_payload, status_error = fetch_optional_payload(status_url, args.timeout)
         print_once(fetch_payload(args.api_url, args.timeout, refresh=True), args.filter, status_payload, status_error)
         if str(args.logs_token or "").strip():
@@ -1236,17 +1711,27 @@ def main(argv: list[str] | None = None) -> int:
             except Exception as exc:
                 print(f"keys error: {exc}", file=sys.stderr)
         return 0
+    start_page = "accounts"
+    if args.logs:
+        start_page = "logs"
+    elif args.errors:
+        start_page = "errors"
     return run_textual(
         args.api_url,
         status_url,
         args.logs_url,
         args.logs_token,
+        args.errors_cn_url,
+        args.errors_us_url,
         max(1, args.refresh_seconds),
         max(1, args.logs_refresh_seconds),
+        max(1, args.errors_refresh_seconds),
         max(1, args.timeout),
         logs_limit,
+        errors_limit,
+        errors_time_range,
         version_message,
-        start_page="logs" if args.logs else "accounts",
+        start_page=start_page,
     )
 
 

tests/test_payload.py

@@ -341,7 +341,6 @@ def sample_logs_payload() -> dict:
                     "user": {"id": 1, "name": "shujakuin"},
                     "model": "gpt-5.5",
                     "upstream_model": "gpt-5.5-codex",
-                    "reasoning_effort": "high",
                     "request_type": "stream",
                     "stream": True,
                     "input_tokens": 1200,
@@ -433,55 +432,6 @@ class Sub2APILogsTests(unittest.TestCase):
         capped = mod.logs_request_url("https://sub2apicn.shujk.top/api/v1/admin/usage", 9000)
         self.assertIn("page_size=1000", capped)
 
-    def test_log_tokens_per_second_uses_decode_window(self) -> None:
-        mod = load_module()
-
-        item = {"output_tokens": 340, "duration_ms": 5321, "first_token_ms": 800}
-        self.assertAlmostEqual(mod.log_tokens_per_second(item), 340 * 1000.0 / (5321 - 800))
-        self.assertEqual(mod.format_rate(mod.log_tokens_per_second(item)), "75.2/s")
-        # missing first token: falls back to the full duration
-        self.assertAlmostEqual(mod.log_tokens_per_second({"output_tokens": 100, "duration_ms": 2000}), 50.0)
-        # no output, zero duration, or first token >= duration -> no rate
-        self.assertEqual(mod.log_tokens_per_second({"output_tokens": 0, "duration_ms": 2000}), 0.0)
-        self.assertEqual(mod.log_tokens_per_second({"output_tokens": 10, "duration_ms": 0}), 0.0)
-        self.assertEqual(mod.log_tokens_per_second({"output_tokens": 10, "duration_ms": 500, "first_token_ms": 500}), 0.0)
-        self.assertEqual(mod.format_rate(0), "-")
-        self.assertEqual(mod.format_rate(123.4), "123/s")
-
-    def test_normalize_log_rows_carries_first_token_and_rate(self) -> None:
-        mod = load_module()
-
-        rows = mod.normalize_log_rows(sample_logs_payload())
-        oldest = rows[-1]
-        self.assertEqual(oldest["first_token_ms"], 800)
-        self.assertAlmostEqual(oldest["tokens_per_second"], 340 * 1000.0 / 4521)
-        self.assertEqual(oldest["effort"], "high")
-        self.assertEqual(rows[0]["effort"], "-")
-        detail = mod.log_detail_line(oldest)
-        self.assertIn("first 0.8s latency 5.3s", detail)
-        self.assertIn("(high)", detail)
-        self.assertIn("75.2/s", detail)
-
-    def test_format_seconds_converts_ms(self) -> None:
-        mod = load_module()
-
-        self.assertEqual(mod.format_seconds(5321), "5.3s")
-        self.assertEqual(mod.format_seconds(800), "0.8s")
-        self.assertEqual(mod.format_seconds(200735), "200.7s")
-        self.assertEqual(mod.format_seconds(0), "-")
-        self.assertEqual(mod.format_seconds(None), "-")
-
-    def test_key_color_is_stable_and_from_palette(self) -> None:
-        mod = load_module()
-
-        first = mod.key_color("codex-main")
-        self.assertIn(first, mod.KEY_COLOR_PALETTE)
-        self.assertEqual(first, mod.key_color("codex-main"))
-        self.assertEqual(mod.key_color(""), "")
-        self.assertEqual(mod.key_color("-"), "")
-        colors = {mod.key_color(name) for name in ("a", "b", "c", "d", "e", "f")}
-        self.assertGreater(len(colors), 1)
-
     def test_print_logs_once_renders_requested_columns(self) -> None:
         mod = load_module()
         out = io.StringIO()
@@ -497,15 +447,7 @@ class Sub2APILogsTests(unittest.TestCase):
         self.assertIn("stream", text)
         self.assertIn("5.7K", text)
         self.assertIn("$0.012", text)
-        header = text.splitlines()[1]
-        self.assertLess(header.index("model"), header.index("effort"))
-        self.assertLess(header.index("effort"), header.index("type"))
-        self.assertLess(header.index("first"), header.index("latency"))
-        self.assertIn("tok/s", header)
-        self.assertIn("high", text)
         self.assertIn("5.3s", text)
-        self.assertIn("0.8s", text)
-        self.assertIn("75.2/s", text)
         self.assertIn("total 2.3K records", text)
 
     def test_default_logs_token_reads_config_file(self) -> None:
@@ -562,84 +504,168 @@ class Sub2APILogsTests(unittest.TestCase):
             else:
                 os.environ["SHUSUB2_LOGS_URL_FILE"] = old_file
 
-
-def sample_key_usage_payload() -> dict:
-    return {
-        "date": "2026-07-21",
-        "trend": [
-            {"date": "2026-07-21", "api_key_id": 3, "key_name": "codex-main", "requests": 120, "tokens": 4_500_000},
-            {"date": "2026-07-21", "api_key_id": 4, "key_name": "claude-max", "requests": 40, "tokens": 9_100_000},
-            {"date": "2026-07-21", "api_key_id": 9, "key_name": "test", "requests": 3, "tokens": 250},
-        ],
-        "stats": {
-            "3": {"api_key_id": 3, "today_actual_cost": 12.5, "total_actual_cost": 100.0},
-            "4": {"api_key_id": 4, "today_actual_cost": 30.25, "total_actual_cost": 90.0},
-        },
-    }
-
-
-class Sub2APIKeyUsageTests(unittest.TestCase):
-    def test_admin_api_base_derived_from_logs_url(self) -> None:
+    def test_errors_request_url_forces_page_limit_and_time_range(self) -> None:
         mod = load_module()
 
-        self.assertEqual(
-            mod.admin_api_base("https://sub2apicn.shujk.top/api/v1/admin/usage"),
-            "https://sub2apicn.shujk.top/api/v1/admin",
+        url = mod.errors_request_url(
+            "https://sub2apicn.shujk.top/api/v1/admin/ops/errors?page=9&page_size=5&q=x",
+            100,
+            "6h",
         )
-        self.assertEqual(
-            mod.admin_api_base("https://sub2apicn.shujk.top/api/v1/admin/usage/"),
-            "https://sub2apicn.shujk.top/api/v1/admin",
-        )
-        self.assertEqual(
-            mod.admin_api_base("https://sub2apicn.shujk.top/api/v1/admin"),
-            "https://sub2apicn.shujk.top/api/v1/admin",
-        )
-        self.assertEqual(mod.admin_api_base("https://example.com/other/path"), "")
-        self.assertEqual(mod.admin_api_base("not a url"), "")
+        parsed = dict(part.split("=") for part in url.split("?", 1)[1].split("&"))
+        self.assertEqual(parsed["page"], "1")
+        self.assertEqual(parsed["page_size"], "100")
+        self.assertEqual(parsed["time_range"], "6h")
+        self.assertEqual(parsed["q"], "x")
 
-    def test_normalize_key_rows_merges_costs_and_sorts_by_cost(self) -> None:
-        mod = load_module()
+        capped = mod.errors_request_url("https://sub2apicn.shujk.top/api/v1/admin/ops/errors", 9000)
+        self.assertIn("page_size=500", capped)
+        self.assertIn("time_range=24h", capped)
 
-        rows = mod.normalize_key_rows(sample_key_usage_payload())
-
-        self.assertEqual([row["name"] for row in rows], ["claude-max", "codex-main", "test"])
-        top = rows[0]
-        self.assertEqual(top["cost"], 30.25)
-        self.assertEqual(top["tokens"], 9_100_000)
-        self.assertEqual(top["requests"], 40)
-        # key 9 has no cost stats -> cost 0, sorted last
-        self.assertEqual(rows[-1]["cost"], 0.0)
-        self.assertEqual(rows[-1]["requests"], 3)
-
-    def test_normalize_key_rows_aggregates_multiple_points_per_key(self) -> None:
+    def test_normalize_error_rows_merges_filters_and_sorts(self) -> None:
         mod = load_module()
         payload = {
-            "trend": [
-                {"api_key_id": 3, "key_name": "codex-main", "requests": 10, "tokens": 100},
-                {"api_key_id": 3, "key_name": "codex-main", "requests": 5, "tokens": 50},
-            ],
-            "stats": {},
+            "items": [
+                {
+                    "_node": "cn",
+                    "id": 1,
+                    "created_at": "2026-07-24T10:00:00+08:00",
+                    "status_code": 429,
+                    "api_key_name": "key-a",
+                    "account_name": "acc-a",
+                    "model": "gpt-5.5",
+                    "phase": "upstream",
+                    "type": "rate_limit",
+                    "error_owner": "provider",
+                    "error_source": "upstream_http",
+                    "message": "too many requests",
+                    "request_id": "req-a",
+                },
+                {
+                    "_node": "us",
+                    "id": 2,
+                    "created_at": "2026-07-24T11:00:00+08:00",
+                    "status_code": 500,
+                    "api_key_id": 9,
+                    "account_id": 12,
+                    "requested_model": "claude-opus",
+                    "phase": "internal",
+                    "type": "api_error",
+                    "error_owner": "platform",
+                    "error_source": "gateway",
+                    "message": "upstream failed",
+                    "request_id": "req-b",
+                },
+            ]
         }
 
-        rows = mod.normalize_key_rows(payload)
+        rows = mod.normalize_error_rows(payload)
+        self.assertEqual([row["id"] for row in rows], [2, 1])
+        self.assertEqual(rows[0]["node"], "us")
+        self.assertEqual(rows[0]["key"], "#9")
+        self.assertEqual(rows[0]["account"], "#12")
+        self.assertEqual(rows[0]["model"], "claude-opus")
+        self.assertEqual([row["id"] for row in mod.normalize_error_rows(payload, "rate_limit")], [1])
+        self.assertEqual([row["id"] for row in mod.normalize_error_rows(payload, "us")], [2])
+        self.assertEqual(mod.normalize_error_rows(payload, "no-match"), [])
 
-        self.assertEqual(len(rows), 1)
-        self.assertEqual(rows[0]["requests"], 15)
-        self.assertEqual(rows[0]["tokens"], 150)
-
-    def test_print_keys_once_renders_key_summary(self) -> None:
+    def test_fetch_merged_errors_labels_nodes_and_tolerates_partial_failure(self) -> None:
         mod = load_module()
-        out = io.StringIO()
 
-        with contextlib.redirect_stdout(out):
-            mod.print_keys_once(sample_key_usage_payload())
+        def fake_fetch(url: str, token: str, timeout: int, limit: int = 100, time_range: str = "24h"):
+            if "cn" in url:
+                return {
+                    "data": {
+                        "items": [
+                            {
+                                "id": 10,
+                                "created_at": "2026-07-24T12:00:00+08:00",
+                                "status_code": 404,
+                                "model": "gpt-a",
+                                "message": "cn miss",
+                            }
+                        ],
+                        "total": 3,
+                    }
+                }
+            raise RuntimeError("us down")
 
-        text = out.getvalue()
-        self.assertIn("keys today 2026-07-21 | 3 keys | $42.75", text)
-        self.assertIn("claude-max", text)
-        self.assertIn("$30.25", text)
-        self.assertIn("9.1M", text)
-        self.assertIn("codex-main", text)
+        old = mod.fetch_errors_payload
+        mod.fetch_errors_payload = fake_fetch
+        try:
+            payload = mod.fetch_merged_errors(
+                [("cn", "https://example/cn"), ("us", "https://example/us")],
+                "token",
+                5,
+                50,
+                "1h",
+            )
+        finally:
+            mod.fetch_errors_payload = old
+
+        self.assertEqual(len(payload["items"]), 1)
+        self.assertEqual(payload["items"][0]["_node"], "cn")
+        self.assertTrue(payload["sources"]["cn"]["ok"])
+        self.assertEqual(payload["sources"]["cn"]["total"], 3)
+        self.assertFalse(payload["sources"]["us"]["ok"])
+        self.assertIn("us down", payload["sources"]["us"]["error"])
+        self.assertIn("cn 1/3", mod.errors_summary_line(payload, 1))
+        self.assertIn("us err", mod.errors_summary_line(payload, 1))
+
+    def test_default_errors_urls(self) -> None:
+        mod = load_module()
+        old_cn = os.environ.pop("SHUSUB2_ERRORS_CN_URL", None)
+        old_us = os.environ.pop("SHUSUB2_ERRORS_US_URL", None)
+        old_cn_file = os.environ.get("SHUSUB2_ERRORS_CN_URL_FILE")
+        old_us_file = os.environ.get("SHUSUB2_ERRORS_US_URL_FILE")
+        os.environ["SHUSUB2_ERRORS_CN_URL_FILE"] = "/nonexistent/shusub2/errors-cn-url"
+        os.environ["SHUSUB2_ERRORS_US_URL_FILE"] = "/nonexistent/shusub2/errors-us-url"
+        try:
+            self.assertEqual(mod.default_errors_cn_url(), "https://sub2apicn.shujk.top/api/v1/admin/ops/errors")
+            self.assertEqual(
+                mod.default_errors_us_url(),
+                "https://sub2api.server4.shujk.top:19857/api/v1/admin/ops/errors",
+            )
+            self.assertEqual(
+                mod.default_error_sources(mod.default_errors_cn_url(), mod.default_errors_us_url()),
+                [
+                    ("cn", "https://sub2apicn.shujk.top/api/v1/admin/ops/errors"),
+                    ("us", "https://sub2api.server4.shujk.top:19857/api/v1/admin/ops/errors"),
+                ],
+            )
+        finally:
+            if old_cn is not None:
+                os.environ["SHUSUB2_ERRORS_CN_URL"] = old_cn
+            if old_us is not None:
+                os.environ["SHUSUB2_ERRORS_US_URL"] = old_us
+            if old_cn_file is None:
+                os.environ.pop("SHUSUB2_ERRORS_CN_URL_FILE", None)
+            else:
+                os.environ["SHUSUB2_ERRORS_CN_URL_FILE"] = old_cn_file
+            if old_us_file is None:
+                os.environ.pop("SHUSUB2_ERRORS_US_URL_FILE", None)
+            else:
+                os.environ["SHUSUB2_ERRORS_US_URL_FILE"] = old_us_file
+
+    def test_once_errors_without_token_prints_hint_and_exits_2(self) -> None:
+        mod = load_module()
+        old_token = os.environ.pop("SHUSUB2_LOGS_TOKEN", None)
+        old_file = os.environ.get("SHUSUB2_LOGS_TOKEN_FILE")
+        os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = "/nonexistent/shusub2/logs-token"
+        err = io.StringIO()
+        try:
+            with contextlib.redirect_stderr(err):
+                rc = mod.main(["--once", "--errors", "--no-version-check"])
+        finally:
+            if old_token is not None:
+                os.environ["SHUSUB2_LOGS_TOKEN"] = old_token
+            if old_file is None:
+                os.environ.pop("SHUSUB2_LOGS_TOKEN_FILE", None)
+            else:
+                os.environ["SHUSUB2_LOGS_TOKEN_FILE"] = old_file
+
+        self.assertEqual(rc, 2)
+        self.assertIn("logs token not configured", err.getvalue())
 
 
 if __name__ == "__main__":