feat: show per-event traffic errors

README.md

@@ -69,9 +69,9 @@ Workspace 当前提供以下有界 projection：
 - `accounts`：账号摘要、quota window、当日用量和 provider/group/status 字段。
 - `status`：channel monitor 摘要。
 - `sources`：Pricing Monitor 已有的脱敏 source 余额与健康状态。
-- `traffic.requests`：最近的有界请求样本。
-- `traffic.errors`：有界错误聚合，保留聚合次数和状态链路。
-- `traffic.keys`：按 `Asia/Shanghai` 自然日聚合的 key 使用量，payload 同时携带统计日期和时区。
+- `traffic.requests`：仅 server6 的最近有界请求样本。
+- `traffic.errors`：server6/server4 的有界逐条错误事件，保留节点、时间、key/account、结构化错误类别和状态链路，不做错误分组或 per-row count。
+- `traffic.keys`：仅 server6 按 `Asia/Shanghai` 自然日聚合的 key 使用量，payload 同时携带统计日期和时区。
 
 投影允许展示 account/key 名称、稳定 ID、model、instance 和运维状态，但不包含 API
 key 原文、access/refresh token、cookie、密码、数据库凭据或请求/响应正文。ID 始终按
@@ -117,10 +117,10 @@ Accounts 只在 canonical 名称 `{family}-quota-{source}` 或
 provider、URL、display name 或模糊文本推断；source 为 error/stale 时保留名称和状态，
 但不伪造 CNY 数值。
 
-Requests 展示 key、account、model、token bucket、actual cost、first-token latency、
-duration 和 decode throughput。Errors 展示 instance、聚合次数、status path、key、
-account、model、phase/type/owner 和时间。Key usage、Requests、Errors 和对应 account
-当天用量都使用 workspace 服务器声明的 `Asia/Shanghai` 自然日，不依赖客户端本地日期或滚动
+Requests 展示 server6 的 key、account、model、token bucket、actual cost、first-token latency、
+duration 和 decode throughput。Errors 展示 server6/server4 的逐条 instance、事件时间、status path、key、
+account、model、phase/type/owner，不显示聚合次数。Key usage、Requests 和对应 account 当天用量只采用
+server6；Errors 保留双节点。所有 workspace Traffic 都使用服务器声明的 `Asia/Shanghai` 自然日，不依赖客户端本地日期或滚动
 24 小时窗口；每个 Traffic projection 显式携带日期、时区和 `[00:00, 次日 00:00)` 边界。
 
 ## Legacy Direct

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "shusub2"
-version = "0.3.1"
+version = "0.3.2"
 description = "Aggregated operations TUI for Sub2API"
 readme = "README.md"
 requires-python = ">=3.11"

sub2api_quota_tui.py

@@ -26,7 +26,7 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 
 APP_NAME = "shusub2"
-FALLBACK_VERSION = "0.3.1"
+FALLBACK_VERSION = "0.3.2"
 DEFAULT_WORKSPACE_URL = "https://price.tailbeb9ad.ts.net/api/ui-data?view=workspace"
 DEFAULT_WORKSPACE_URL_CONFIG_FILE = "~/.config/shusub2/workspace-url"
 DEFAULT_API_URL = "http://127.0.0.1:18318/api/tui/accounts"
@@ -58,7 +58,7 @@ MAX_WORKSPACE_MONITORS = 256
 MAX_WORKSPACE_SOURCES = 64
 MAX_WORKSPACE_INSTANCES = 16
 MAX_WORKSPACE_REQUESTS = 100
-MAX_WORKSPACE_ERRORS = 256
+MAX_WORKSPACE_ERROR_EVENTS = 256
 MAX_WORKSPACE_KEYS = 256
 USAGE_TIMEZONE = "Asia/Shanghai"
 MONITOR_OK_STATUSES = {"operational", "ok", "success"}
@@ -68,6 +68,13 @@ PRICING_ACCOUNT_SOURCE_RE = re.compile(
     r"^[a-z0-9]+-quota(?:only)?-(?P<source>[a-z0-9]+(?:-[a-z0-9]+)*)$",
     re.IGNORECASE,
 )
+ERROR_DISPLAY_LABEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._/+\-]{0,95}$")
+ERROR_URL_RE = re.compile(r"(?i)\b(?:https?|wss?)://|\bwww\.")
+ERROR_EMAIL_RE = re.compile(r"(?i)\b[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}\b")
+ERROR_SECRET_RE = re.compile(
+    r"(?i)\b(?:authorization|bearer|api[_ -]?key|token|password|secret|cookie|session(?:[_ -]?id)?)\b"
+)
+ERROR_OPAQUE_RE = re.compile(r"(?i)\b(?:sk|rk|pk|eyj)[_-]?[A-Z0-9]{12,}\b")
 INSTALL_COMMAND = "uv tool install --force git+https://gitea.shujk.top/shujakuin/shusub2.git"
 INSTALL_COMMAND_ARGS = ["uv", "tool", "install", "--force", "git+https://gitea.shujk.top/shujakuin/shusub2.git"]
 
@@ -610,7 +617,7 @@ def fetch_workspace_payload(workspace_url: str, timeout: int) -> dict[str, Any]:
         ),
         ("instances", traffic.get("instances"), MAX_WORKSPACE_INSTANCES),
         ("requests", traffic.get("requests"), MAX_WORKSPACE_REQUESTS),
-        ("errors", traffic.get("errors"), MAX_WORKSPACE_ERRORS),
+        ("errors", traffic.get("errors"), MAX_WORKSPACE_ERROR_EVENTS),
         ("keys", traffic.get("keys"), MAX_WORKSPACE_KEYS),
     )
     for label, rows, maximum in bounded_fields:
@@ -618,6 +625,10 @@ def fetch_workspace_payload(workspace_url: str, timeout: int) -> dict[str, Any]:
             not isinstance(row, dict) for row in rows
         ):
             raise RuntimeError(f"Pricing Monitor returned invalid workspace {label}")
+    for label in ("requests", "keys"):
+        for row in traffic.get(label) or []:
+            if str(row.get("instance") or row.get("node") or "").strip() != "server6":
+                raise RuntimeError(f"Pricing Monitor returned noncanonical workspace {label}")
     if len(payload["sources"]) > MAX_WORKSPACE_SOURCES or not all(
         isinstance(source, dict) for source in payload["sources"]
     ):
@@ -651,8 +662,11 @@ def workspace_logs_payload(payload: dict[str, Any]) -> dict[str, Any]:
     for raw in raw_items:
         if not isinstance(raw, dict):
             continue
+        instance = str(raw.get("instance") or raw.get("node") or "").strip()
+        if instance != "server6":
+            continue
         item = dict(raw)
-        item["_node"] = str(raw.get("instance") or raw.get("node") or "-")
+        item["_node"] = instance or "-"
         if "total_cost" not in item:
             item["total_cost"] = (
                 raw.get("cost")
@@ -670,8 +684,15 @@ def workspace_logs_payload(payload: dict[str, Any]) -> dict[str, Any]:
 def workspace_keys_payload(payload: dict[str, Any]) -> dict[str, Any]:
     traffic = payload.get("traffic") if isinstance(payload.get("traffic"), dict) else {}
     period = workspace_traffic_period(traffic) or {}
+    raw_items = traffic.get("keys") if isinstance(traffic.get("keys"), list) else []
+    items = [
+        dict(item)
+        for item in raw_items
+        if isinstance(item, dict)
+        and str(item.get("instance") or item.get("node") or "").strip() == "server6"
+    ]
     return {
-        "items": traffic.get("keys") if isinstance(traffic.get("keys"), list) else [],
+        "items": items,
         "generated_at": traffic.get("generated_at") or payload.get("generated_at"),
         "period_kind": period.get("period_kind", ""),
         "date": period.get("date", ""),
@@ -687,26 +708,37 @@ def workspace_errors_payload(payload: dict[str, Any]) -> dict[str, Any]:
     for index, raw in enumerate(raw_items):
         if not isinstance(raw, dict):
             continue
-        item = dict(raw)
-        item.setdefault("id", index + 1)
-        item["_node"] = str(raw.get("instance") or raw.get("node") or "-")
-        item.setdefault("created_at", raw.get("latest_at"))
-        item.setdefault("requested_model", raw.get("model"))
-        item.setdefault("phase", raw.get("error_source"))
-        item.setdefault("type", raw.get("error_type"))
-        item.setdefault("message", raw.get("error_detail") or "")
-        items.append(item)
+        items.append(
+            {
+                "id": raw.get("id", index + 1),
+                "_node": raw.get("instance") or raw.get("node") or "-",
+                "created_at": raw.get("created_at") or raw.get("latest_at"),
+                "status_code": raw.get("status_code"),
+                "inbound_status_code": raw.get("inbound_status_code"),
+                "upstream_status_code": raw.get("upstream_status_code"),
+                "api_key_id": raw.get("api_key_id"),
+                "api_key_name": raw.get("api_key_name"),
+                "account_id": raw.get("account_id"),
+                "account_name": raw.get("account_name"),
+                "requested_model": raw.get("model"),
+                "upstream_model": raw.get("upstream_model"),
+                "phase": raw.get("error_source"),
+                "type": raw.get("error_type"),
+                "error_source": raw.get("error_source"),
+                "error_type": raw.get("error_type"),
+            }
+        )
     instances = traffic.get("instances") if isinstance(traffic.get("instances"), list) else []
     sources = {}
     for raw in instances:
         if not isinstance(raw, dict):
             continue
-        node = str(raw.get("instance") or raw.get("name") or "-")
+        node = safe_error_label(raw.get("instance") or raw.get("name"), "-")
         sources[node] = {
             "ok": raw.get("ok") is not False,
             "total": as_int(raw.get("error_total")),
-            "fetched": sum(1 for item in items if item.get("_node") == node),
-            "error": str(raw.get("error") or ""),
+            "fetched": sum(1 for item in items if safe_error_label(item.get("_node"), "-") == node),
+            "error": "unavailable" if raw.get("error") else "",
         }
     return {
         "items": items,
@@ -1214,17 +1246,15 @@ def fetch_merged_errors(
         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}
+                source_status[node] = {"ok": False, "total": 0, "fetched": 0, "error": "unavailable"}
                 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": "",
@@ -1243,48 +1273,62 @@ def fetch_merged_errors(
     }
 
 
+def safe_error_label(value: Any, fallback: str = "-") -> str:
+    text = re.sub(r"[\x00-\x1f\x7f]+", " ", str(value or "")).strip()
+    if (
+        not text
+        or not ERROR_DISPLAY_LABEL_RE.fullmatch(text)
+        or ERROR_URL_RE.search(text)
+        or ERROR_EMAIL_RE.search(text)
+        or ERROR_SECRET_RE.search(text)
+        or ERROR_OPAQUE_RE.search(text)
+    ):
+        return fallback
+    return text
+
+
+def safe_error_timestamp(value: Any) -> str:
+    parsed = parse_time(value)
+    if parsed is None:
+        return ""
+    return parsed.astimezone(dt.timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+
+
 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 item.get("error_source") or "-").strip() or "-"
-        error_type = str(item.get("type") or item.get("error_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 item.get("error_detail") 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 "-"
+        key_id = as_int(item.get("api_key_id"))
+        account_id = as_int(item.get("account_id"))
+        node = safe_error_label(item.get("_node") or item.get("node"), "-")
+        key_name = safe_error_label(
+            item.get("api_key_name"),
+            f"#{key_id}" if key_id else "-",
+        )
+        account_name = safe_error_label(
+            item.get("account_name"),
+            f"#{account_id}" if account_id else "-",
+        )
+        model = safe_error_label(item.get("requested_model") or item.get("model"), "-")
+        upstream_model = safe_error_label(item.get("upstream_model"), "")
+        phase = safe_error_label(item.get("phase") or item.get("error_source"), "-")
+        error_type = safe_error_label(item.get("type") or item.get("error_type"), "-")
+        source = safe_error_label(item.get("error_source"), "-")
+        owner = safe_error_label(item.get("error_owner"), "-")
+        platform = safe_error_label(item.get("platform"), "-")
+        group_name = safe_error_label(item.get("group_name"), "-")
+        created_at = safe_error_timestamp(item.get("created_at"))
         row = {
             "id": as_int(item.get("id")),
             "node": node,
-            "status_code": status_code,
-            "count": max(1, as_int(item.get("error_count")) or 1),
+            "status_code": as_int(item.get("status_code")),
             "inbound_status_code": as_int(item.get("inbound_status_code")),
             "upstream_status_code": as_int(item.get("upstream_status_code")),
             "key": key_name,
             "account": account_name,
-            "user": user_email or "-",
+            "user": "-",
             "model": model,
             "upstream_model": upstream_model,
             "phase": phase,
@@ -1293,14 +1337,13 @@ def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list
             "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")),
-            "age": relative_age(item.get("created_at")),
+            "message": "",
+            "request_id": "",
+            "client_request_id": "",
+            "created_at": created_at,
+            "time": short_time(created_at),
+            "age": relative_age(created_at),
             "resolved": bool(item.get("resolved")),
-            "raw": item,
         }
         if needle:
             haystack = " ".join(
@@ -1308,7 +1351,6 @@ def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list
                     node,
                     key_name,
                     account_name,
-                    user_email,
                     model,
                     upstream_model,
                     phase,
@@ -1317,10 +1359,7 @@ def normalize_error_rows(payload: dict[str, Any], filter_text: str = "") -> list
                     source,
                     platform,
                     group_name,
-                    message,
-                    request_id,
-                    client_request_id,
-                    str(status_code),
+                    str(row["status_code"]),
                 )
             ).lower()
             if needle not in haystack:
@@ -1359,14 +1398,10 @@ def error_detail_line(row: dict[str, Any]) -> str:
     ]
     status_path_text = "/".join(value for value in status_path if value) or "-"
     detail = (
-        f"{row['time']} | {row['node']} | count {row.get('count', 1)} | {status_path_text} | key {row['key']} | account {row['account']} | "
+        f"{row['time']} | {row['node']} | {status_path_text} | 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
@@ -1379,14 +1414,13 @@ def print_errors_once(payload: dict[str, Any], filter_text: str = "") -> None:
     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 '-'}")
+            print(f"{node}: ok fetched {as_int(info.get('fetched'))} total {as_int(info.get('total'))}")
         else:
-            print(f"{node}: error {info.get('error') or 'unknown'} | {info.get('url') or '-'}")
-    print("node  count  status  key                   account               model                     phase      type           owner     time         age")
+            print(f"{node}: error {info.get('error') or 'unknown'}")
+    print("node  status  key                   account               model                     phase      type           owner     time         age")
     for row in rows:
         print(
             f"{row['node']:<5} "
-            f"{format_count(row.get('count', 1)):<6} "
             f"{row['status_code']:<7} "
             f"{row['key'][:20]:<21} "
             f"{row['account'][:20]:<21} "
@@ -1972,7 +2006,7 @@ def run_textual(
             )
             self.configure_table(
                 self.query_one("#errors", DataTable),
-                ("Node", "Count", "Status", "Key", "Account", "Model", "Time", "Age"),
+                ("Node", "Status", "Key", "Account", "Model", "Time", "Age"),
             )
             self.refresh_all(refresh=legacy_direct)
             self.set_interval(refresh_seconds, self.refresh_accounts)
@@ -2177,7 +2211,6 @@ def run_textual(
                 self.error_by_key[key] = row
                 table.add_row(
                     row["node"],
-                    format_count(row.get("count", 1)),
                     str(row["status_code"]),
                     row["key"],
                     row["account"],
@@ -2785,7 +2818,7 @@ def run_textual(
             table = self.query_one("#errors", DataTable)
             table.cursor_type = "row"
             table.zebra_stripes = True
-            table.add_columns("Node", "Count", "Status", "Key", "Account", "Model", "Phase", "Type", "Owner", "Time", "Age")
+            table.add_columns("Node", "Status", "Key", "Account", "Model", "Phase", "Type", "Owner", "Time", "Age")
             self.refresh_data()
             self.set_interval(errors_refresh_seconds, self.refresh_data)
 
@@ -2852,7 +2885,6 @@ def run_textual(
                 self.row_by_key[key] = row
                 table.add_row(
                     row["node"],
-                    format_count(row.get("count", 1)),
                     str(row["status_code"]),
                     row["key"],
                     row["account"],

tests/test_payload.py

@@ -4,6 +4,7 @@ import contextlib
 import copy
 import datetime as dt
 import io
+import json
 import os
 from pathlib import Path
 import importlib.util
@@ -113,8 +114,8 @@ def workspace_payload_fixture() -> dict[str, object]:
             "errors": [
                 {
                     "instance": "server6",
-                    "latest_at": "2026-08-03T11:58:00Z",
-                    "error_count": 4,
+                    "id": "21",
+                    "created_at": "2026-08-03T11:58:00Z",
                     "status_code": 502,
                     "inbound_status_code": 500,
                     "upstream_status_code": 502,
@@ -762,6 +763,27 @@ class WorkspaceTests(unittest.TestCase):
     def test_workspace_adapters_preserve_ids_and_map_compact_traffic(self) -> None:
         mod = load_module()
         payload = workspace_payload_fixture()
+        payload["traffic"]["requests"].append(
+            {"instance": "server4", "id": "71", "created_at": "2026-08-03T11:57:00Z"}
+        )
+        payload["traffic"]["keys"].append(
+            {"instance": "server4", "api_key_id": "71", "api_key_name": "server4-copy"}
+        )
+        payload["traffic"]["errors"].append(
+            {
+                "instance": "server4",
+                "id": "22",
+                "created_at": "2026-08-03T11:57:00Z",
+                "status_code": 503,
+                "api_key_id": "8",
+                "account_id": "10",
+                "error_type": "upstream",
+                "error_source": "upstream_http",
+                "error_detail": "forbidden-detail",
+                "request_id": "forbidden-request-id",
+                "user_email": "forbidden@example.test",
+            }
+        )
 
         accounts = mod.workspace_accounts_payload(payload)
         status = mod.workspace_status_payload(payload)
@@ -773,6 +795,7 @@ class WorkspaceTests(unittest.TestCase):
         self.assertEqual(accounts["accounts"][0]["id"], "9007199254740993")
         self.assertEqual(status["channel_monitors"]["items"][0]["id"], "9007199254740995")
         self.assertEqual(pricing["sources"][0]["name"], "code-plan")
+        self.assertEqual(len(logs["data"]["items"]), 1)
         self.assertEqual(logs["data"]["items"][0]["id"], "9007199254740997")
         self.assertEqual(logs["data"]["items"][0]["instance"], "server6")
         normalized_accounts = mod.normalize_account_rows(accounts, pricing_payload=pricing)
@@ -781,6 +804,7 @@ class WorkspaceTests(unittest.TestCase):
         self.assertEqual(normalized_logs[0]["id"], 9007199254740997)
         self.assertEqual(normalized_logs[0]["cost"], 0.25)
         self.assertEqual(mod.as_int("9007199254740999"), 9007199254740999)
+        self.assertEqual(len(keys["items"]), 1)
         self.assertEqual(mod.normalize_key_rows(keys)[0]["name"], "wmy")
         self.assertEqual(mod.normalize_key_rows(keys)[0]["cost"], 0.25)
         self.assertEqual(keys["period_kind"], "calendar_day")
@@ -788,7 +812,13 @@ class WorkspaceTests(unittest.TestCase):
         self.assertEqual(keys["timezone"], "Asia/Shanghai")
         self.assertIn(f"today {keys['date']} (Asia/Shanghai)", mod.usage_period_label(keys))
         self.assertEqual(errors["time_range"], f"today {keys['date']} (Asia/Shanghai)")
-        self.assertEqual(errors["items"][0]["error_count"], 4)
+        self.assertEqual({item["_node"] for item in errors["items"]}, {"server6", "server4"})
+        self.assertNotIn("forbidden-detail", json.dumps(errors))
+        self.assertNotIn("forbidden-request-id", json.dumps(errors))
+        self.assertNotIn("forbidden@example.test", json.dumps(errors))
+        self.assertEqual(errors["items"][0]["id"], "21")
+        self.assertEqual(errors["items"][0]["created_at"], "2026-08-03T11:58:00Z")
+        self.assertNotIn("error_count", errors["items"][0])
         self.assertEqual(errors["items"][0]["phase"], "upstream_http")
         self.assertEqual(errors["sources"]["server6"]["total"], 4)
         self.assertEqual(errors["sources"]["server4"]["total"], 0)
@@ -805,8 +835,8 @@ class WorkspaceTests(unittest.TestCase):
         self.assertEqual(fallback_row["key"], "#9007199254740999")
         self.assertEqual(fallback_row["account"], "#9007199254740998")
         normalized_errors = mod.normalize_error_rows(errors)
-        self.assertEqual(normalized_errors[0]["count"], 4)
-        self.assertIn("count 4", mod.error_detail_line(normalized_errors[0]))
+        self.assertNotIn("count", normalized_errors[0])
+        self.assertNotIn("| count ", mod.error_detail_line(normalized_errors[0]))
 
     def test_workspace_payload_validation_rejects_nonfinite_and_oversized_rows(self) -> None:
         mod = load_module()
@@ -846,6 +876,18 @@ class WorkspaceTests(unittest.TestCase):
             with self.assertRaisesRegex(RuntimeError, "invalid workspace traffic period"):
                 mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
 
+        noncanonical_request = workspace_payload_fixture()
+        noncanonical_request["traffic"]["requests"][0]["instance"] = "server4"
+        with mock.patch.object(mod, "fetch_payload", return_value=noncanonical_request):
+            with self.assertRaisesRegex(RuntimeError, "noncanonical workspace requests"):
+                mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
+
+        missing_key_provenance = workspace_payload_fixture()
+        missing_key_provenance["traffic"]["keys"][0].pop("instance")
+        with mock.patch.object(mod, "fetch_payload", return_value=missing_key_provenance):
+            with self.assertRaisesRegex(RuntimeError, "noncanonical workspace keys"):
+                mod.fetch_workspace_payload("https://workspace.example.test/data", 3)
+
     def test_workspace_cache_coalesces_reads_and_returns_copies(self) -> None:
         mod = load_module()
         calls: list[str] = []
@@ -1200,6 +1242,39 @@ class Sub2APILogsTests(unittest.TestCase):
         self.assertEqual([row["id"] for row in mod.normalize_error_rows(payload, "us")], [2])
         self.assertEqual(mod.normalize_error_rows(payload, "no-match"), [])
 
+        unsafe = {
+            "items": [
+                {
+                    "id": 3,
+                    "_node": "server6",
+                    "created_at": "2026-07-24T12:00:00+08:00",
+                    "status_code": 502,
+                    "api_key_name": "safe-key",
+                    "account_name": "safe-account",
+                    "error_type": "upstream",
+                    "error_source": "upstream_http",
+                    "error_detail": "private detail token=redacted-test-value",
+                    "message": "private message",
+                    "request_id": "private-request-id",
+                    "client_request_id": "private-client-request-id",
+                    "user_email": "private@example.test",
+                }
+            ]
+        }
+        safe_row = mod.normalize_error_rows(unsafe)[0]
+        serialized = json.dumps(safe_row)
+        for private_value in (
+            "redacted-test-value",
+            "private message",
+            "private-request-id",
+            "private-client-request-id",
+            "private@example.test",
+        ):
+            self.assertNotIn(private_value, serialized)
+        self.assertEqual(safe_row["message"], "")
+        self.assertEqual(safe_row["request_id"], "")
+        self.assertEqual(safe_row["user"], "-")
+
     def test_fetch_merged_errors_labels_nodes_and_tolerates_partial_failure(self) -> None:
         mod = load_module()
 
@@ -1239,7 +1314,7 @@ class Sub2APILogsTests(unittest.TestCase):
         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.assertEqual(payload["sources"]["us"]["error"], "unavailable")
         self.assertIn("cn 1/3", mod.errors_summary_line(payload, 1))
         self.assertIn("us err", mod.errors_summary_line(payload, 1))
 
@@ -1515,7 +1590,8 @@ class DashboardLayoutTests(unittest.IsolatedAsyncioTestCase):
                 self.assertEqual(type(screen).__name__, "DashboardScreen")
                 self.assertEqual(app.sub_title, "Dashboard")
                 self.assertEqual(screen.account_rows[0]["name"], "oai-quota-code-plan")
-                self.assertEqual(screen.error_rows[0]["count"], 4)
+                self.assertEqual(screen.error_rows[0]["key"], "wmy")
+                self.assertNotIn("count", screen.error_rows[0])
                 self.assertEqual(len(fetch.call_args_list), 1)
 
                 await pilot.press("p")

uv.lock

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