feat: TRAE 签到接入（此前「credential does not support checkin」）

M1.5 只实现了 CodeBuddy 的签到，TRAE 的签到端点从未接入——
界面上给所有凭证都渲染了「签到」按钮，但 TRAE 点了必然报
「credential does not support checkin」。这是「按钮与能力不匹配」。

按原实现接入（trae2api-web 的 CheckinStatus/CheckinClaim）：
- 先查 /trae/api/v2/ug/checkin_credits/status（checked_in/credits/enable）
- 未签且 enable=true 才调 claim；checked_in=true 视为成功（already_checked_in）
- enable=false → 明确提示「当前账号不可签到」
- 请求头走 UgHeaders（Authorization + X-User-Region: CN + X-Device-Id）

顺带统一：CheckinResult 从 codebuddy/checkin.py 上移到 provider/base.py，
两个 provider 共用同一类型（此前 TRAE 缺签到正是这个类型归属混乱的表现）。

CheckinTask 无需改动——它只依赖 provider.checkin/checkin_scope 协议。

后端 600 测试 / 覆盖 100%。真实凭证验证：
TRAE 签到 ok=true，额度探测随之刷新（4500 → 4499.92）。

Author: RobbsLuo

src/provider/base.py

@@ -74,6 +74,17 @@ def health_score(quota: Quota | None) -> int | None:
     return max(0, min(100, round(remaining / quota.total * 100)))
 
 
+@dataclass(slots=True)
+class CheckinResult:
+    """签到结果。already_checked_in 表示当日已签（不算错误）。"""
+
+    ok: bool
+    credit: float | None = None
+    code: int | None = None
+    message: str = ""
+    already_checked_in: bool = False
+
+
 @dataclass(slots=True)
 class Model:
     id: str

src/provider/codebuddy/checkin.py

@@ -9,26 +9,17 @@
 from __future__ import annotations
 
 import math
-from dataclasses import dataclass
 from typing import Any
 
 import httpx
 
+from ...provider.base import CheckinResult
 from .client import build_headers
 from .credential import CodeBuddyCredential
 from .events import UpstreamProtocolViolation
 from .headers import EP_DAILY_CHECKIN
 
 
-@dataclass(slots=True)
-class CheckinResult:
-    ok: bool
-    credit: float | None = None
-    code: int | None = None
-    message: str = ""
-    already_checked_in: bool = False
-
-
 def parse_checkin_response(body: Any) -> CheckinResult:
     """严格按上游语义解析：code=0 且 credit 为有限数值才算成功。"""
     if not isinstance(body, dict):

src/provider/trae/client.py

@@ -13,7 +13,7 @@
 import httpx
 
 from ...engine.sse import iter_frames
-from ...provider.base import AuthSession, ErrKind, Event, Model, Quota
+from ...provider.base import AuthSession, CheckinResult, ErrKind, Event, Model, Quota
 from . import events as trae_events
 from .callback import (
     CallbackInfo,
@@ -270,6 +270,21 @@ async def fetch_quota(self, credential: TraeCredential) -> Quota:
             used += float(pack_used) if isinstance(pack_used, (int, float)) else 0.0
         return Quota(remaining=max(0.0, limit - used), total=limit, probed_at=int(time.time()))
 
+    async def fetch_checkin_status(self, credential: TraeCredential) -> dict[str, Any]:
+        """checkin_credits/status：checked_in / credits / enable。"""
+        data = await self._post_json(
+            f"{self.ug_host}{trae_events.EP_CHECKIN_STATUS}", {}, ug_headers(credential))
+        return {
+            "checked_in": bool(data.get("checked_in")),
+            "credits": data.get("credits"),
+            "enable": bool(data.get("enable")),
+        }
+
+    async def claim_checkin(self, credential: TraeCredential) -> dict[str, Any] | None:
+        """checkin_credits/claim：领取当日积分。"""
+        return await self._post_json(
+            f"{self.ug_host}{trae_events.EP_CHECKIN_CLAIM}", {}, ug_headers(credential))
+
     async def refresh_token(self, credential: TraeCredential) -> TraeCredential:
         """ExchangeToken；失败不改写原凭证字段。"""
         host = credential.api_host or self.oauth_host
@@ -372,6 +387,25 @@ async def aclose(self) -> None:
         """释放内部 HTTP 连接池。"""
         await self.client.aclose()
 
+    async def checkin(self, credential_data: dict) -> CheckinResult:
+        """TRAE 签到：先查状态，未签且可签才领取。
+
+        上游没有独立的「已签到」错误码，status.checked_in 就是已签语义。
+        """
+        credential = TraeCredential.from_dict(credential_data)
+        status = await self.client.fetch_checkin_status(credential)
+        if status["checked_in"]:
+            return CheckinResult(ok=True, credit=None, message="今天已签到",
+                                 already_checked_in=True)
+        if not status["enable"]:
+            return CheckinResult(ok=False, message="当前账号不可签到")
+        await self.client.claim_checkin(credential)
+        return CheckinResult(ok=True, credit=None)
+
+    def checkin_scope(self, credential_data: dict) -> str:
+        """同上游账号的多凭证共享一次签到。"""
+        return f"trae|{credential_data.get('uid', '')}"
+
     # ------------------------------------------------- callback 轨道（Q17=C）
 
     def parse_login_url(self, raw_url: str) -> CallbackInfo:

src/provider/trae/events.py

@@ -28,6 +28,8 @@
 EP_MODELS = "/api/ide/v1/get_detail_param"
 EP_EXCHANGE = "/cloudide/api/v3/trae/oauth/ExchangeToken"
 EP_USER_INFO = "/cloudide/api/v3/trae/GetUserInfo"
+EP_CHECKIN_STATUS = "/trae/api/v2/ug/checkin_credits/status"
+EP_CHECKIN_CLAIM = "/trae/api/v2/ug/checkin_credits/claim"
 
 
 class UpstreamProtocolViolation(ValueError):

tests/test_m15_operations.py

@@ -797,14 +797,45 @@ async def test_checkin_dedupes_same_upstream_account(repo):
     assert provider.checkin_calls == 1
 
 
-async def test_checkin_skips_disabled_and_unsupported(repo):
+async def test_checkin_skips_disabled(repo):
     credentials, _db = repo
     credential_id = credentials.add(provider="codebuddy", credential_data={"bearer_token": "a"})
     credentials.save_error(credential_id, _disabled_outcome())
-    credentials.add(provider="trae", credential_data={"accessToken": "t"})
-    task = CheckinTask(credentials, {"codebuddy": ProbeProvider(), "trae": TraeProvider()})
+    task = CheckinTask(credentials, {"codebuddy": ProbeProvider()})
     report = await task.run_once()
     assert report.attempted == 0
+    assert report.skipped == 1
+
+
+async def test_checkin_supports_both_providers(repo):
+    """TRAE 与 CodeBuddy 都实现签到后，任务要能覆盖两个上游。"""
+    credentials, _db = repo
+    credentials.add(provider="codebuddy", credential_data={"bearer_token": "a",
+                                                          "account_uid": "cb"})
+    credentials.add(provider="trae", credential_data={"accessToken": "t", "uid": "tr"})
+
+    class TraeCheckinProvider:
+        id = "trae"
+
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def checkin(self, _data):
+            self.calls += 1
+            from src.provider.base import CheckinResult
+
+            return CheckinResult(ok=True, credit=None, message="今天已签到",
+                                 already_checked_in=True)
+
+        def checkin_scope(self, data):
+            return f"trae|{data.get('uid', '')}"
+
+    trae = TraeCheckinProvider()
+    task = CheckinTask(credentials, {"codebuddy": ProbeProvider(), "trae": trae})
+    report = await task.run_once()
+    assert report.attempted == 2
+    assert report.succeeded == 2
+    assert trae.calls == 1
 
 
 def _disabled_outcome():
@@ -2079,7 +2110,6 @@ def test_schedule_probe_returns_early_when_credential_unreadable(admin_client):
 
 def test_trae_start_auth_builds_login_url_with_public_callback():
     from src.main import resolve_public_callback_url
-    from src.provider.trae.client import TraeProvider
 
     settings = Settings(_env_file=None, APP_SECRET="s", PUBLIC_BASE_URL="https://gw.example")
     provider = TraeProvider()
@@ -2096,7 +2126,7 @@ async def test_trae_complete_callback_exchanges_token():
     """回调链接必须真的换 token，而不是只存 refreshToken。"""
     import httpx as _httpx
 
-    from src.provider.trae.client import TraeClient, TraeProvider
+    from src.provider.trae.client import TraeClient
 
     def handler(request: _httpx.Request) -> _httpx.Response:
         if request.url.path.endswith("ExchangeToken"):
@@ -2123,7 +2153,6 @@ def handler(request: _httpx.Request) -> _httpx.Response:
 
 
 async def test_trae_complete_callback_rejects_bad_state():
-    from src.provider.trae.client import TraeProvider
     from src.provider.trae.events import UpstreamProtocolViolation
 
     provider = TraeProvider()
@@ -2135,7 +2164,7 @@ def test_authorize_completes_trae_login_end_to_end(tmp_path):
     """完整闭环：start → 浏览器回调 → 凭证入库 → 立即探测。"""
     import httpx as _httpx
 
-    from src.provider.trae.client import TraeClient, TraeProvider
+    from src.provider.trae.client import TraeClient
 
     def handler(request: _httpx.Request) -> _httpx.Response:
         if request.url.path.endswith("ExchangeToken"):
@@ -2214,7 +2243,7 @@ async def test_complete_callback_does_not_use_refresh_token_as_access_token():
     """ExchangeToken 失败时不得把 refreshToken 当 accessToken 塞进池子。"""
     import httpx as _httpx
 
-    from src.provider.trae.client import TraeClient, TraeProvider
+    from src.provider.trae.client import TraeClient
     from src.provider.trae.events import UpstreamProtocolViolation
 
     def handler(request: _httpx.Request) -> _httpx.Response:
@@ -2236,7 +2265,7 @@ def handler(request: _httpx.Request) -> _httpx.Response:
 async def test_complete_callback_raises_when_no_token_available():
     import httpx as _httpx
 
-    from src.provider.trae.client import TraeClient, TraeProvider
+    from src.provider.trae.client import TraeClient
     from src.provider.trae.events import UpstreamProtocolViolation
 
     def handler(_request: _httpx.Request) -> _httpx.Response:
@@ -2398,7 +2427,7 @@ def test_authorize_accepts_real_trae_callback_without_state(tmp_path):
     """
     import httpx as _httpx
 
-    from src.provider.trae.client import TraeClient, TraeProvider
+    from src.provider.trae.client import TraeClient
 
     def handler(request: _httpx.Request) -> _httpx.Response:
         if request.url.path.endswith("ExchangeToken"):

tests/test_m1a_gaps.py

@@ -654,3 +654,114 @@ def test_solo_headers_include_uid_when_present():
     assert headers["X-Uid"] == "u-9"
     no_uid = solo_headers(TraeCredential(access_token="a"))
     assert "X-Uid" not in no_uid
+
+
+# ------------------------------------------- TRAE 签到（三态 + 真实请求）
+
+def _trae_client(handler) -> TraeClient:
+    import httpx as _httpx
+
+    transport = _httpx.MockTransport(handler)
+    return TraeClient(stream_client=_httpx.AsyncClient(transport=transport, timeout=None),
+                      short_client=_httpx.AsyncClient(transport=transport, timeout=None))
+
+
+async def test_trae_checkin_status_and_claim_use_ug_headers():
+    """签到两端点必须走 ug_headers（含 X-User-Region），且用 POST。"""
+    seen: list[tuple[str, dict]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.append((request.url.path, dict(request.headers)))
+        if request.url.path.endswith("checkin_credits/status"):
+            return httpx.Response(200, json={"checked_in": False, "credits": 0,
+                                             "enable": True})
+        return httpx.Response(200, json={"credits": 200})
+
+    client = _trae_client(handler)
+    status = await client.fetch_checkin_status(TraeCredential(access_token="a", device_id="d"))
+    assert status == {"checked_in": False, "credits": 0, "enable": True}
+
+    claim = await client.claim_checkin(TraeCredential(access_token="a", device_id="d"))
+    assert claim == {"credits": 200}
+
+    assert len(seen) == 2
+    for path, headers in seen:
+        assert "checkin_credits" in path
+        # httpx 内部存储全小写
+        assert headers.get("x-user-region") == "CN"
+        assert headers.get("authorization") == "Cloud-IDE-JWT a"
+
+
+async def test_trae_provider_checkin_already_is_success():
+    """已签到（status.checked_in=true）→ ok=true + already_checked_in。"""
+    from src.provider.trae.client import TraeProvider
+
+    async def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"checked_in": True, "credits": 0, "enable": True})
+
+    provider = TraeProvider(client=_trae_client(handler))
+    result = await provider.checkin({"accessToken": "a"})
+    assert result.ok is True and result.already_checked_in is True
+    assert result.message == "今天已签到"
+
+
+async def test_trae_provider_checkin_disabled_reports_not_ok():
+    """status.enable=false → 不可签到。"""
+    from src.provider.trae.client import TraeProvider
+
+    async def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, json={"checked_in": False, "credits": 0, "enable": False})
+
+    provider = TraeProvider(client=_trae_client(handler))
+    result = await provider.checkin({"accessToken": "a"})
+    assert result.ok is False
+
+
+async def test_trae_provider_checkin_claims_when_eligible():
+    """未签且可签 → 调 claim 并成功。"""
+    paths: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        paths.append(request.url.path)
+        if request.url.path.endswith("checkin_credits/status"):
+            return httpx.Response(200, json={"checked_in": False, "credits": 0,
+                                             "enable": True})
+        return httpx.Response(200, json={"credits": 500})
+
+    from src.provider.trae.client import TraeProvider
+
+    provider = TraeProvider(client=_trae_client(handler))
+    result = await provider.checkin({"accessToken": "a"})
+    assert result.ok is True and result.already_checked_in is False
+    assert any(p.endswith("checkin_credits/claim") for p in paths)
+
+
+def test_trae_checkin_scope_uses_uid():
+    from src.provider.trae.client import TraeProvider
+
+    provider = TraeProvider()
+    assert provider.checkin_scope({"uid": "u1"}) == "trae|u1"
+    assert provider.checkin_scope({}) == "trae|"
+
+
+async def test_checkin_task_skips_provider_without_scope(tmp_path):
+    """provider 没有 checkin_scope 时跳过（background 148-150）。"""
+    from src.db.conn import Database
+    from src.db.crypto import CredentialCipher
+    from src.db.migrate import apply_schema
+    from src.db.repo import CredentialRepository
+
+    db = Database(tmp_path / "ns.sqlite3")
+    apply_schema(db.connect())
+    credentials = CredentialRepository(db, CredentialCipher("s"))
+    credentials.add(provider="codebuddy", credential_data={"bearer_token": "t"})
+
+    from src.tasks.background import CheckinTask
+
+    class NoScope:
+        id = "codebuddy"
+
+    task = CheckinTask(credentials, {"codebuddy": NoScope()})
+    report = await task.run_once()
+    assert report.skipped == 1
+    db.close()