refactor: 收敛重复样板（错误类型/统计 scope/DB 事务）

- UpstreamHTTPError 提到 provider/base.py：两个 provider 此前逐字重复
  status/body/kind()，子类只绑 classify_status
- admin_stats：五个端点重复的 `username if is_admin else principal.username`
  收进 _scope()
- Database.transaction() 上下文：repo/collector 里 14 处
  connect→execute→commit 样板收敛，多语句写入（set_pinned）异常时回滚，
  此前中途抛错会留下半完成状态；新增提交/回滚两个分支的测试
- TECHNICAL §4：可选能力是「不实现即不定义 + 调用方 getattr 探测 → 400」，
  不是原文写的「trae 抛 NotImplementedError」（与实现不符）
- TECHNICAL §7：写入统一走 transaction()，无应用层写锁

Author: RobbsLuo

TECHNICAL.md

@@ -194,7 +194,7 @@ class Provider(Protocol):
     def import_credential(self, raw: dict) -> Credential: ...
     def refresh(self, cred: DecryptedCred) -> None: ...
 
-    # 可选能力（CB 独有；trae 抛 NotImplementedError）
+    # 可选能力（CB 独有：多账号切换）
     def list_accounts(self, cred: DecryptedCred) -> list[Account]: ...
     def switch_account(self, cred: DecryptedCred, account_id: str) -> None: ...
 
@@ -213,6 +213,11 @@ class Provider(Protocol):
 约定：
 - `DecryptedCred` = db 取出 `data_enc` → Fernet 解密后的 dataclass；provider 不接触 sqlite
 - `stream_chat` 只产出 `Event`，产出前先做 HTTP 状态码检查；`classify` 由 executor 调用
+- 可选能力**不实现即不定义**（不是抛 `NotImplementedError`）：调用方用
+  `getattr`/`hasattr` 探测，缺失时返回 400「该凭证不支持此操作」，而不是 500。
+  当前可选集：`start_auth`/`poll_auth`（仅支持 poll 的 provider 才有）、
+  `complete_callback`（仅 TRAE）、`list_accounts`/`switch_account`（仅 CodeBuddy）、
+  `credential_from`/`checkin_scope`（刷新与签到任务的能力探测）
 - 两个 provider 共用 `engine/sse.py` 的帧解析器（SSE 规范层），事件语义各自映射
 
 ---
@@ -266,7 +271,7 @@ class Scheduler:
     def pin(self, credential_id: str | None) -> None: ...
 ```
 
-状态全部落 `credentials` 表（`cooling_until` / `err_count` / `health` / `disabled` / `quota_expiry_ladder`），进程重启不丢冷却状态。写路径无应用层锁：每个写方法直接走当前线程的连接，并发写靠 SQLite WAL + `busy_timeout=5000` 串行化。
+状态全部落 `credentials` 表（`cooling_until` / `err_count` / `health` / `disabled` / `quota_expiry_ladder`），进程重启不丢冷却状态。写路径无应用层锁：并发写靠 SQLite WAL + `busy_timeout=5000` 串行化；每个写方法走 `Database.transaction()` 上下文（正常提交、异常回滚），不再散落 `connect()/commit()` 样板。
 
 到期积分只算一处：`expiring_credits()`。选号走 `Candidate.expiry_credits()`，管理台列表走 `GET /api/credentials` 的 `quota_expiring_credits`（窗口值随响应返回 `expiry_window_seconds`），两处共用同一实现，界面数字与选号顺序不会漂移；渠道无到期信息时返回 `null`（不显示），窗口关闭或确实无积分临近过期时返回 `0`（同样不显示）。
 
@@ -322,7 +327,7 @@ PRAGMA busy_timeout = 5000;
 PRAGMA foreign_keys = ON;      -- api_keys 之外无外键（users.txt 无表）
 ```
 
-- 连接：`threading.local()` 每线程一个 `sqlite3.Connection(row_factory=sqlite3.Row)`；引擎与 FastAPI 线程池各自持有自己的连接。没有应用层写锁，写入并发由 SQLite 自身串行化（WAL + `busy_timeout=5000` 下短写足够；确需多语句原子性时用显式事务）
+- 连接：`threading.local()` 每线程一个 `sqlite3.Connection(row_factory=sqlite3.Row)`；引擎与 FastAPI 线程池各自持有自己的连接。写入统一走 `Database.transaction()`（`conn.commit()` / 异常 `rollback()`），没有应用层写锁，并发由 SQLite 自身串行化（WAL + `busy_timeout=5000` 下短写足够）
 - 加密：`Fernet(base64.urlsafe_b64encode(sha256(APP_SECRET).digest()))`；APP_SECRET 丢失 = 凭证全部不可解，只能重录（Q13 已明示）
 - migration：启动时读 `schema.sql` 逐条 `CREATE TABLE IF NOT EXISTS`（只加不改，列注释可改）；新增列写进 `migrate._MIGRATION_COLUMNS` 走 `ALTER TABLE ... ADD COLUMN`（重复列名忽略，老库幂等补列），删表写进 `migrate._MIGRATION_DROPS` 走 `DROP TABLE IF EXISTS`（`CREATE TABLE IF NOT EXISTS` 对老库无效，不给删会遗留死表），同时 `SCHEMA_VERSION + 1`，版本记在 `PRAGMA user_version`
 

src/api/admin_stats.py

@@ -4,46 +4,50 @@
 
 from fastapi import APIRouter, Depends
 
+from ..auth.rbac import Principal
 from .deps import Services, principal_from_request
 
 
+def _scope(principal: Principal, username: str | None) -> str | None:
+    """admin 传 username 可查全局/指定人；普通用户永远只能看自己。"""
+    return username if principal.is_admin else principal.username
+
+
 def create_router(services: Services) -> APIRouter:
     router = APIRouter()
     stats_query = services.stats_query
 
     @router.get("/api/stats/overview")
     async def stats_overview(principal=Depends(principal_from_request),
                              username: str | None = None, since: int | None = None):
-        target = username if principal.is_admin else principal.username
-        return stats_query.overview(username=target, since=since)
+        return stats_query.overview(username=_scope(principal, username), since=since)
 
     @router.get("/api/stats/by-provider")
     async def stats_by_provider(principal=Depends(principal_from_request),
                                 username: str | None = None, since: int | None = None):
-        target = username if principal.is_admin else principal.username
-        return {"providers": stats_query.by_provider(username=target, since=since)}
+        return {"providers": stats_query.by_provider(
+            username=_scope(principal, username), since=since)}
 
     @router.get("/api/stats/timeline")
     async def stats_timeline(principal=Depends(principal_from_request),
                              username: str | None = None, since: int | None = None,
                              metric: str = "requests"):
-        target = username if principal.is_admin else principal.username
-        return {"points": stats_query.timeline(username=target, since=since, metric=metric)}
+        return {"points": stats_query.timeline(
+            username=_scope(principal, username), since=since, metric=metric)}
 
     @router.get("/api/stats/model-timeline")
     async def stats_model_timeline(principal=Depends(principal_from_request),
                                    username: str | None = None, since: int | None = None,
                                    metric: str = "requests"):
-        target = username if principal.is_admin else principal.username
-        return stats_query.model_timeline(username=target, since=since, metric=metric)
+        return stats_query.model_timeline(
+            username=_scope(principal, username), since=since, metric=metric)
 
     @router.get("/api/stats/events")
     async def stats_events(principal=Depends(principal_from_request),
                            username: str | None = None, since: int | None = None,
                            before: int | None = None, limit: int = 50):
         # 明细保留 90 天；单页上限 200，防止一次拉爆
-        target = username if principal.is_admin else principal.username
-        return stats_query.events(username=target, since=since, before=before,
-                                  limit=max(1, min(limit, 200)))
+        return stats_query.events(username=_scope(principal, username), since=since,
+                                  before=before, limit=max(1, min(limit, 200)))
 
     return router

src/db/conn.py

@@ -7,6 +7,8 @@
 
 import sqlite3
 import threading
+from collections.abc import Iterator
+from contextlib import contextmanager
 from pathlib import Path
 
 PRAGMAS = (
@@ -42,3 +44,19 @@ def close(self) -> None:
         if conn is not None:
             conn.close()
             self._local.conn = None
+
+    @contextmanager
+    def transaction(self) -> Iterator[sqlite3.Connection]:
+        """写操作事务：正常退出提交，异常回滚。
+
+        多语句写入必须走这个上下文（如「先清 pin 再置 pin」），
+        否则中途抛错会留下半完成状态；单语句写入也可用，
+        让「connect → execute → commit」的样板只出现一次。
+        """
+        conn = self.connect()
+        try:
+            yield conn
+        except BaseException:
+            conn.rollback()
+            raise
+        conn.commit()

src/db/repo.py

@@ -55,86 +55,81 @@ def add(self, *, provider: str, credential_data: dict, nickname: str = "",
             added_by: str = "", now: int | None = None) -> str:
         credential_id = _new_id("cred")
         payload = json.dumps(credential_data, ensure_ascii=False).encode("utf-8")
-        self._db.connect().execute(
-            "INSERT INTO credentials (id, provider, nickname, data_enc, created_at, added_by) "
-            "VALUES (?,?,?,?,?,?)",
-            (credential_id, provider, nickname, self._cipher.encrypt(payload),
-             int(now if now is not None else time.time()), added_by),
-        )
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            conn.execute(
+                "INSERT INTO credentials (id, provider, nickname, data_enc, created_at, added_by) "
+                "VALUES (?,?,?,?,?,?)",
+                (credential_id, provider, nickname, self._cipher.encrypt(payload),
+                 int(now if now is not None else time.time()), added_by),
+            )
         return credential_id
 
     def delete(self, credential_id: str) -> bool:
-        cursor = self._db.connect().execute("DELETE FROM credentials WHERE id = ?",
-                                           (credential_id,))
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            cursor = conn.execute("DELETE FROM credentials WHERE id = ?", (credential_id,))
         return cursor.rowcount > 0
 
     def set_enabled(self, credential_id: str, enabled: bool) -> bool:
-        cursor = self._db.connect().execute(
-            "UPDATE credentials SET enabled = ? WHERE id = ?", (1 if enabled else 0, credential_id))
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            cursor = conn.execute(
+                "UPDATE credentials SET enabled = ? WHERE id = ?",
+                (1 if enabled else 0, credential_id))
         return cursor.rowcount > 0
 
     def set_pinned(self, credential_id: str | None) -> None:
-        conn = self._db.connect()
-        conn.execute("UPDATE credentials SET pinned = 0")
-        if credential_id is not None:
-            conn.execute("UPDATE credentials SET pinned = 1 WHERE id = ?", (credential_id,))
-        conn.commit()
+        with self._db.transaction() as conn:
+            conn.execute("UPDATE credentials SET pinned = 0")
+            if credential_id is not None:
+                conn.execute("UPDATE credentials SET pinned = 1 WHERE id = ?", (credential_id,))
 
     def revive(self, credential_id: str) -> bool:
         """解除硬禁用（session 死亡）与冷却，允许凭证重新参与调度。
 
         没有这个入口时，凭证一旦因 session 失效被硬禁用就只能删除重建，
         重新登录后也无法复用同一条记录。
         """
-        cursor = self._db.connect().execute(
-            "UPDATE credentials SET disabled = 0, disabled_reason = NULL, cooling_until = NULL, "
-            "err_count = 0 WHERE id = ?", (credential_id,))
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            cursor = conn.execute(
+                "UPDATE credentials SET disabled = 0, disabled_reason = NULL, "
+                "cooling_until = NULL, err_count = 0 WHERE id = ?", (credential_id,))
         return cursor.rowcount > 0
 
     def save_error(self, credential_id: str, outcome: ErrorOutcome) -> None:
-        conn = self._db.connect()
-        if outcome.disabled:
-            conn.execute(
-                "UPDATE credentials SET disabled = 1, disabled_reason = ?, err_count = 0, "
-                "cooling_until = NULL WHERE id = ?", ("session dead", credential_id))
-        else:
-            conn.execute(
-                "UPDATE credentials SET cooling_until = ?, err_count = ? WHERE id = ?",
-                (outcome.cooling_until, outcome.err_count, credential_id))
-        conn.commit()
+        with self._db.transaction() as conn:
+            if outcome.disabled:
+                conn.execute(
+                    "UPDATE credentials SET disabled = 1, disabled_reason = ?, err_count = 0, "
+                    "cooling_until = NULL WHERE id = ?", ("session dead", credential_id))
+            else:
+                conn.execute(
+                    "UPDATE credentials SET cooling_until = ?, err_count = ? WHERE id = ?",
+                    (outcome.cooling_until, outcome.err_count, credential_id))
 
     def save_success(self, credential_id: str) -> None:
-        conn = self._db.connect()
-        conn.execute("UPDATE credentials SET err_count = 0 WHERE id = ?", (credential_id,))
-        conn.commit()
+        with self._db.transaction() as conn:
+            conn.execute("UPDATE credentials SET err_count = 0 WHERE id = ?", (credential_id,))
 
     def save_credential_data(self, credential_id: str, credential_data: dict) -> None:
         payload = json.dumps(credential_data, ensure_ascii=False).encode("utf-8")
-        conn = self._db.connect()
-        conn.execute("UPDATE credentials SET data_enc = ? WHERE id = ?",
-                     (self._cipher.encrypt(payload), credential_id))
-        conn.commit()
+        with self._db.transaction() as conn:
+            conn.execute("UPDATE credentials SET data_enc = ? WHERE id = ?",
+                         (self._cipher.encrypt(payload), credential_id))
 
     def save_quota(self, credential_id: str, quota: Quota) -> None:
-        conn = self._db.connect()
-        conn.execute(
-            "UPDATE credentials SET quota_remaining = ?, quota_total = ?, quota_cycle_end = ?, "
-            "quota_expiry_ladder = ?, quota_probed_at = ?, health = ? WHERE id = ?",
-            (quota.remaining, quota.total, quota.cycle_end,
-             _ladder_text(quota.expiry_ladder), quota.probed_at,
-             health_score(quota), credential_id),
-        )
-        conn.commit()
+        with self._db.transaction() as conn:
+            conn.execute(
+                "UPDATE credentials SET quota_remaining = ?, quota_total = ?, "
+                "quota_cycle_end = ?, quota_expiry_ladder = ?, quota_probed_at = ?, health = ? "
+                "WHERE id = ?",
+                (quota.remaining, quota.total, quota.cycle_end,
+                 _ladder_text(quota.expiry_ladder), quota.probed_at,
+                 health_score(quota), credential_id),
+            )
 
     def mark_probe_failed(self, credential_id: str, now: int | None = None) -> None:
-        conn = self._db.connect()
-        conn.execute("UPDATE credentials SET quota_probed_at = ?, health = NULL WHERE id = ?",
-                     (int(now if now is not None else time.time()), credential_id))
-        conn.commit()
+        with self._db.transaction() as conn:
+            conn.execute("UPDATE credentials SET quota_probed_at = ?, health = NULL WHERE id = ?",
+                         (int(now if now is not None else time.time()), credential_id))
 
     # ------------------------------------------------------------- 读取
 
@@ -210,13 +205,13 @@ def create(self, username: str, name: str = "", now: int | None = None) -> dict[
         plaintext = generate_api_key()
         key_id = _new_id("key")
         created_at = int(now if now is not None else time.time())
-        self._db.connect().execute(
-            "INSERT INTO api_keys (id, username, name, key_digest, preview, created_at) "
-            "VALUES (?,?,?,?,?,?)",
-            (key_id, username, name, digest_api_key(plaintext), preview_api_key(plaintext),
-             created_at),
-        )
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            conn.execute(
+                "INSERT INTO api_keys (id, username, name, key_digest, preview, created_at) "
+                "VALUES (?,?,?,?,?,?)",
+                (key_id, username, name, digest_api_key(plaintext), preview_api_key(plaintext),
+                 created_at),
+            )
         return {"id": key_id, "username": username, "name": name, "api_key": plaintext,
                 "preview": preview_api_key(plaintext), "created_at": created_at}
 
@@ -226,9 +221,9 @@ def verify(self, api_key: str) -> str | None:
             "SELECT id, username FROM api_keys WHERE key_digest = ?", (digest,)).fetchone()
         if row is None:
             return None
-        self._db.connect().execute("UPDATE api_keys SET last_used_at = ? WHERE id = ?",
-                                   (int(time.time()), row["id"]))
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            conn.execute("UPDATE api_keys SET last_used_at = ? WHERE id = ?",
+                         (int(time.time()), row["id"]))
         return row["username"]
 
     def list_for(self, username: str) -> list[dict[str, Any]]:
@@ -238,7 +233,7 @@ def list_for(self, username: str) -> list[dict[str, Any]]:
         return [dict(row) for row in rows]
 
     def delete(self, key_id: str, username: str) -> bool:
-        cursor = self._db.connect().execute(
-            "DELETE FROM api_keys WHERE id = ? AND username = ?", (key_id, username))
-        self._db.connect().commit()
+        with self._db.transaction() as conn:
+            cursor = conn.execute(
+                "DELETE FROM api_keys WHERE id = ? AND username = ?", (key_id, username))
         return cursor.rowcount > 0

src/provider/base.py

@@ -6,9 +6,10 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass
 from enum import StrEnum
-from typing import Protocol, runtime_checkable
+from typing import ClassVar, Protocol, runtime_checkable
 
 
 class EventKind(StrEnum):
@@ -39,6 +40,24 @@ def body_hint(body: bytes, limit: int = 160) -> str:
     return text[:limit]
 
 
+class UpstreamHTTPError(Exception):
+    """上游非 2xx；两个 provider 共用同一形状（status + 原始 body）。
+
+    `kind()` 由子类绑定各自的 classify_status，因为 1005 等业务码的
+    判定规则是上游协议私有的。
+    """
+
+    classify_status: ClassVar[Callable[[int, bytes], ErrKind]]
+
+    def __init__(self, status: int, body: bytes) -> None:
+        self.status = status
+        self.body = body
+        super().__init__(f"upstream http {status}: {body_hint(body)}")
+
+    def kind(self) -> ErrKind:
+        return self.classify_status(self.status, self.body)
+
+
 @dataclass(slots=True)
 class Usage:
     input_tokens: int | None = None

src/provider/codebuddy/client.py

@@ -17,7 +17,8 @@
 import httpx
 
 from ...engine.sse import iter_frames
-from ...provider.base import ErrKind, Event, Model, Quota, body_hint
+from ...provider import base
+from ...provider.base import ErrKind, Event, Model, Quota
 from . import events as cb_events
 from .credential import CodeBuddyCredential, parse_credential
 from .events import UpstreamProtocolViolation
@@ -435,14 +436,10 @@ def _cached_refresh(client: CodeBuddyClient):
 CODEBUDDY_IDE_VERSION = "1.42.0"
 
 
-class UpstreamHTTPError(Exception):
-    def __init__(self, status: int, body: bytes) -> None:
-        self.status = status
-        self.body = body
-        super().__init__(f"upstream http {status}: {body_hint(body)}")
+class UpstreamHTTPError(base.UpstreamHTTPError):
+    """CodeBuddy 上游非 2xx；kind() 走 CB 的 1005/400 规则。"""
 
-    def kind(self) -> ErrKind:
-        return cb_events.classify_status(self.status, self.body)
+    classify_status = staticmethod(cb_events.classify_status)
 
 
 @dataclass(slots=True)