feat: show channel monitor health

README.md

@@ -19,13 +19,15 @@ uv tool install git+https://gitea.shujk.top/shujakuin/shusub2.git
 shusub2 --api-url https://example.com/api/tui/accounts
 ```
 
-Configure a default API URL for `shusub2`:
+Configure default API and optional status URLs for `shusub2`:
 
 ```bash
 mkdir -p ~/.config/shusub2
 chmod 700 ~/.config/shusub2
 printf '%s\n' 'https://example.com/api/tui/accounts' > ~/.config/shusub2/api-url
+printf '%s\n' 'https://example.com/api/status' > ~/.config/shusub2/status-url
 chmod 600 ~/.config/shusub2/api-url
+chmod 600 ~/.config/shusub2/status-url
 shusub2
 ```
 
@@ -34,6 +36,8 @@ Environment variables override the config file:
 - `SHUSUB2_API_URL`
 - `SUB2API_QUOTA_TUI_API_URL`
 - `SHUSUB2_API_URL_FILE`
+- `SHUSUB2_STATUS_URL`
+- `SHUSUB2_STATUS_URL_FILE`
 
 ## Local Development
 
@@ -52,11 +56,13 @@ zellij action new-pane --name sub2api-quota -- \
 Configuration:
 
 - `--api-url` / `SUB2API_QUOTA_TUI_API_URL`
+- `--status-url` / `SHUSUB2_STATUS_URL`
 - `--refresh-seconds` / `SUB2API_QUOTA_TUI_REFRESH_SECONDS`
 - `--timeout` / `SUB2API_QUOTA_TUI_TIMEOUT`
 
-The TUI reads only `/api/tui/accounts`; it does not need SSH, database access,
+The TUI reads `/api/tui/accounts` and can optionally read `sub2api-status`
-API keys, OAuth credentials, or plaintext env files.
+`/api/status` for channel monitor health. It does not need SSH, database
+access, API keys, OAuth credentials, or plaintext env files.
 
 Columns are ordered for scanning inside zellij:
 
@@ -73,3 +79,7 @@ Name | Provider | Group | Daily | Today | Tokens | Req | Kind | 5h | 7d | Reset
 `Group` is the derived Sub2API tier alias group. Higher tiers win when multiple
 aliases exist: `id < slow < fast < sfast`. The table shows `sfast` first, then
 `fast`, `slow`, `id`, and ungrouped accounts.
+
+When `--status-url` is configured, the status line shows channel monitor
+health, and the selected account detail shows the matching monitor status when
+one exists.

pyproject.toml

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

sub2api_quota_tui.py

@@ -15,8 +15,12 @@ from typing import Any
 
 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_REFRESH_SECONDS = 60
 DEFAULT_TIMEOUT_SECONDS = 10
+MONITOR_OK_STATUSES = {"operational", "ok", "success"}
+MONITOR_FAILED_STATUSES = {"error", "failed", "failure"}
+MONITOR_STOPWORDS = {"response", "responses", "monitor"}
 
 
 def env_int(name: str, default: int, *, minimum: int = 1) -> int:
@@ -26,12 +30,12 @@ def env_int(name: str, default: int, *, minimum: int = 1) -> int:
         return default
 
 
-def default_api_url() -> str:
+def configured_url(env_names: tuple[str, ...], config_path: str, default: str = "") -> str:
-    for name in ("SHUSUB2_API_URL", "SUB2API_QUOTA_TUI_API_URL"):
+    for name in env_names:
         value = os.environ.get(name, "").strip()
         if value:
             return value
-    config_file = Path(os.environ.get("SHUSUB2_API_URL_FILE", DEFAULT_CONFIG_FILE)).expanduser()
+    config_file = Path(config_path).expanduser()
     try:
         for line in config_file.read_text(encoding="utf-8").splitlines():
             value = line.strip()
@@ -39,7 +43,22 @@ def default_api_url() -> str:
                 return value
     except OSError:
         pass
-    return DEFAULT_API_URL
+    return default
+
+
+def default_api_url() -> str:
+    return configured_url(
+        ("SHUSUB2_API_URL", "SUB2API_QUOTA_TUI_API_URL"),
+        os.environ.get("SHUSUB2_API_URL_FILE", DEFAULT_CONFIG_FILE),
+        DEFAULT_API_URL,
+    )
+
+
+def default_status_url() -> str:
+    return configured_url(
+        ("SHUSUB2_STATUS_URL",),
+        os.environ.get("SHUSUB2_STATUS_URL_FILE", DEFAULT_STATUS_CONFIG_FILE),
+    )
 
 
 def as_float(value: Any) -> float:
@@ -127,6 +146,15 @@ def fetch_payload(api_url: str, timeout: int, *, refresh: bool = False) -> dict[
     return data
 
 
+def fetch_optional_payload(url: str, timeout: int) -> tuple[dict[str, Any], str]:
+    if not str(url or "").strip():
+        return {}, ""
+    try:
+        return fetch_payload(url, timeout), ""
+    except Exception as exc:
+        return {}, str(exc)
+
+
 def kind_label(kind: Any) -> str:
     return {
         "quota_limited": "quota",
@@ -152,6 +180,84 @@ def routing_group_sort_rank(value: Any) -> int:
     return {"sfast": 0, "fast": 1, "slow": 2, "id": 3, "": 4, "-": 4}.get(str(value or "").strip().lower(), 4)
 
 
+def status_kind(value: Any) -> str:
+    text = str(value or "").strip().lower()
+    if text in MONITOR_OK_STATUSES:
+        return "ok"
+    if text in MONITOR_FAILED_STATUSES:
+        return "failed"
+    return "unknown"
+
+
+def format_latency(value: Any) -> str:
+    number = as_int(value)
+    return "-" if number <= 0 else f"{number}ms"
+
+
+def monitor_items(status_payload: dict[str, Any]) -> list[dict[str, Any]]:
+    monitors = status_payload.get("channel_monitors") if isinstance(status_payload.get("channel_monitors"), dict) else {}
+    items = []
+    for item in monitors.get("items") or []:
+        if isinstance(item, dict):
+            items.append(item)
+    return items
+
+
+def monitor_summary(status_payload: dict[str, Any], error: str = "") -> str:
+    if error:
+        return f"monitors error: {error}"
+    monitors = status_payload.get("channel_monitors") if isinstance(status_payload.get("channel_monitors"), dict) else {}
+    if not monitors:
+        return "monitors -"
+    enabled = as_int(monitors.get("enabled"))
+    ok = as_int(monitors.get("latest_ok"))
+    failed = as_int(monitors.get("latest_failed"))
+    unknown = as_int(monitors.get("latest_unknown"))
+    checked = short_time(monitors.get("latest_checked_max"))
+    return f"monitors {ok} ok / {failed} failed / {unknown} unknown ({enabled} enabled, {checked})"
+
+
+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 token not in MONITOR_STOPWORDS and not token.isdigit():
+            tokens.add(token)
+    return tokens
+
+
+def matching_monitor(row: dict[str, Any], status_payload: dict[str, Any]) -> dict[str, Any] | None:
+    account_tokens = text_tokens(row.get("name"))
+    if not account_tokens:
+        return None
+    best: tuple[int, dict[str, Any]] | None = None
+    for item in monitor_items(status_payload):
+        monitor_tokens = text_tokens(item.get("name"))
+        if not monitor_tokens:
+            continue
+        overlap = account_tokens & monitor_tokens
+        if not overlap:
+            continue
+        subset_bonus = 10 if account_tokens <= monitor_tokens or monitor_tokens <= account_tokens else 0
+        score = subset_bonus + len(overlap)
+        if best is None or score > best[0]:
+            best = (score, item)
+    return None if best is None else best[1]
+
+
+def monitor_detail(row: dict[str, Any], status_payload: dict[str, Any]) -> str:
+    item = matching_monitor(row, status_payload)
+    if not item:
+        return "monitor -"
+    status = str(item.get("latest_status") or "unknown")
+    latency = format_latency(item.get("latency_ms"))
+    checked = short_time(item.get("checked_at"))
+    message = str(item.get("message") or "").strip()
+    detail = f"monitor {status} {latency} checked {checked}"
+    if message:
+        detail += f" {message}"
+    return detail
+
+
 def window_for(account: dict[str, Any], window_id: str) -> dict[str, Any]:
     for window in account.get("windows") or []:
         if isinstance(window, dict) and window.get("id") == window_id:
@@ -272,8 +378,10 @@ def summary_line(payload: dict[str, Any]) -> str:
     )
 
 
-def print_once(payload: dict[str, Any], filter_text: str = "") -> None:
+def print_once(payload: dict[str, Any], filter_text: str = "", status_payload: dict[str, Any] | None = None, status_error: str = "") -> None:
     print(summary_line(payload))
+    if status_payload or status_error:
+        print(monitor_summary(status_payload or {}, status_error))
     print("name                            provider  group  daily      today      tokens   req   kind      5h        7d        reset       status")
     for row in normalize_account_rows(payload, filter_text):
         print(
@@ -292,7 +400,7 @@ def print_once(payload: dict[str, Any], filter_text: str = "") -> None:
         )
 
 
-def run_textual(api_url: str, refresh_seconds: int, timeout: int) -> int:
+def run_textual(api_url: str, status_url: str, refresh_seconds: int, timeout: int) -> int:
     try:
         from textual.app import App, ComposeResult
         from textual.widgets import DataTable, Footer, Header, Input, Static
@@ -317,6 +425,8 @@ def run_textual(api_url: str, refresh_seconds: int, timeout: int) -> int:
         def __init__(self) -> None:
             super().__init__()
             self.payload: dict[str, Any] = {}
+            self.status_payload: dict[str, Any] = {}
+            self.status_error = ""
             self.rows: list[dict[str, Any]] = []
             self.row_by_key: dict[str, dict[str, Any]] = {}
 
@@ -352,8 +462,10 @@ def run_textual(api_url: str, refresh_seconds: int, timeout: int) -> int:
             status.update("refreshing...")
             try:
                 self.payload = fetch_payload(api_url, timeout, refresh=refresh)
+                self.status_payload, self.status_error = fetch_optional_payload(status_url, timeout)
                 self.render_payload()
-                status.update(f"source {self.payload.get('source_name') or '-'} | {api_url}")
+                status_bits = [monitor_summary(self.status_payload, self.status_error), f"source {self.payload.get('source_name') or '-'}"]
+                status.update(" | ".join(bit for bit in status_bits if bit) + f" | {api_url}")
             except Exception as exc:
                 status.update(f"error: {exc}")
 
@@ -401,7 +513,8 @@ def run_textual(api_url: str, refresh_seconds: int, timeout: int) -> int:
                 f"{row['daily_quota_remaining'] if row['daily_quota_remaining'] is not None else '-'} left) | "
                 f"today {format_cost(row['today_cost_usd'])}, {format_count(row['today_tokens'])} tokens, "
                 f"{format_count(row['today_requests'])} req | latest {row['latest_usage_at']} | "
-                f"priority {row['priority'] if row['priority'] is not None else '-'}"
+                f"priority {row['priority'] if row['priority'] is not None else '-'} | "
+                f"{monitor_detail(row, self.status_payload)}"
             )
             error = str(raw.get("error") or "").strip()
             if error:
@@ -415,6 +528,7 @@ def run_textual(api_url: str, refresh_seconds: int, timeout: int) -> int:
 def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Sub2API quota and daily usage TUI")
     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(
         "--refresh-seconds",
         type=int,
@@ -429,9 +543,10 @@ def build_parser() -> argparse.ArgumentParser:
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     if args.once:
-        print_once(fetch_payload(args.api_url, args.timeout, refresh=True), args.filter)
+        status_payload, status_error = fetch_optional_payload(args.status_url, args.timeout)
+        print_once(fetch_payload(args.api_url, args.timeout, refresh=True), args.filter, status_payload, status_error)
         return 0
-    return run_textual(args.api_url, max(1, args.refresh_seconds), max(1, args.timeout))
+    return run_textual(args.api_url, args.status_url, max(1, args.refresh_seconds), max(1, args.timeout))
 
 
 if __name__ == "__main__":

tests/test_payload.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import contextlib
+import io
 from pathlib import Path
 import importlib.util
 import sys
@@ -134,6 +136,77 @@ class Sub2APIQuotaTUITests(unittest.TestCase):
 
         self.assertEqual([row["name"] for row in rows], ["sfast", "fast", "slow", "manual"])
 
+    def test_monitor_summary_and_account_match(self) -> None:
+        mod = load_module()
+        status_payload = {
+            "channel_monitors": {
+                "enabled": 6,
+                "latest_ok": 5,
+                "latest_failed": 0,
+                "latest_unknown": 1,
+                "latest_checked_max": "2026-06-09 15:07:40+08:00",
+                "items": [
+                    {
+                        "name": "input responses",
+                        "provider": "openai",
+                        "enabled": True,
+                        "primary_model": "gpt-5.5",
+                        "latest_status": "operational",
+                        "latency_ms": 2606,
+                        "checked_at": "2026-06-09 15:03:28+08:00",
+                        "message": "",
+                    },
+                    {
+                        "name": "kedaya responses",
+                        "provider": "openai",
+                        "enabled": True,
+                        "primary_model": "gpt-5.5",
+                        "latest_status": "degraded",
+                        "latency_ms": 25923,
+                        "checked_at": "2026-06-09 15:07:40+08:00",
+                        "message": "slow response: 25923ms",
+                    },
+                ],
+            }
+        }
+        row = {"name": "input 300"}
+
+        self.assertIn("monitors 5 ok / 0 failed / 1 unknown", mod.monitor_summary(status_payload))
+        self.assertIn("monitor operational 2606ms", mod.monitor_detail(row, status_payload))
+
+    def test_monitor_error_is_non_blocking_in_once_output(self) -> None:
+        mod = load_module()
+        payload = {
+            "generated_at": "2026-06-09T12:00:00+08:00",
+            "totals": {"total_accounts": 1, "usable_accounts": 1, "today_cost_usd": 0, "today_tokens": 0, "today_requests": 0},
+            "accounts": [
+                {"id": 1, "name": "alpha", "routing_group": "slow", "provider": "openai", "kind": "pay_as_you_go", "status": "ok"}
+            ],
+        }
+        out = io.StringIO()
+
+        with contextlib.redirect_stdout(out):
+            mod.print_once(payload, status_error="timeout")
+
+        text = out.getvalue()
+        self.assertIn("monitors error: timeout", text)
+        self.assertIn("alpha", text)
+
+    def test_once_output_includes_monitor_summary(self) -> None:
+        mod = load_module()
+        payload = {
+            "generated_at": "2026-06-09T12:00:00+08:00",
+            "totals": {"total_accounts": 0, "usable_accounts": 0, "today_cost_usd": 0, "today_tokens": 0, "today_requests": 0},
+            "accounts": [],
+        }
+        status_payload = {"channel_monitors": {"enabled": 1, "latest_ok": 1, "latest_failed": 0, "latest_unknown": 0}}
+        out = io.StringIO()
+
+        with contextlib.redirect_stdout(out):
+            mod.print_once(payload, status_payload=status_payload)
+
+        self.assertIn("monitors 1 ok / 0 failed / 0 unknown", out.getvalue())
+
 
 if __name__ == "__main__":
     unittest.main()

uv.lock

@@ -85,7 +85,7 @@ wheels = [
 
 [[package]]
 name = "shusub2"
-version = "0.1.1"
+version = "0.1.2"
 source = { editable = "." }
 dependencies = [
     { name = "textual" },