fix(trae): 9074 不是限流而是设备标识格式——签到设备号改 16 位数字

TRAE 那个号连续签到失败 205 次（两天），日志一直刷「当前参与用户太多」。
此前把它当随机限流做软失败重试，所以永远好不了。

实测对照（同一账号、同一 token，只改 X-Device-Id）：

  deviceId = hex32（登录 URL 那对 machine/device id） → 9074 限流
  deviceId = 16 位纯数字                              → code:0 success
  deviceId = 随机 hex32                               → code:0 success
  deviceId = 空串                                     → 9004 参数错误

结论：9074 是设备标识格式不符（http 200 + 业务码），不是限流。长度/字符集
受限，具体校验规则未知（16 位数字与随机 hex 都过），实现按公开逆向的派生法取
16 位数字：sha256(uid[#genN]) mod 10^16 补零。派生值确定性，重试不会换号。

- ug_headers 新增 device_id 参数；默认取派生值，uid 为空时退回凭证自带 device_id
- fetch_checkin_status / claim_checkin 支持 generation
- checkin 支持 9074 时轮换代次重试（CHECKIN_DEVICE_GENERATIONS=3），
  全部代次都失败才如实返回软失败（不误报成功、不无限重试）
- machine_id 保持登录时那一对（非签到校验项，换掉可能与登录态不匹配）

验证：958 测试 / 100% 行+分支覆盖；ruff 通过。真实账号实测签到成功
（status.checked_in 由 false 变 true，重放 claim 幂等返回 code:0）。

Author: RobbsLuo

PROPOSAL.md

@@ -83,6 +83,7 @@
 - 认证：浏览器登录 → 302 回调 `/authorize` → `ExchangeToken` → `GetUserInfo`
 - Token 刷新：`POST /cloudide/api/v3/trae/oauth/ExchangeToken`（refreshToken 轮换）
 - 签到：`/trae/api/v2/ug/checkin_credits/{status,claim}`；额度：`/trae/api/v2/pay/ide_user_ent_usage`
+- **签到设备头 `X-Device-Id` 必须是 16 位纯数字**：用登录 URL 那套 hex32 deviceId 调 claim 会稳定返回 `9074 当前参与用户太多`——看着像限流，实际是设备标识格式不符。实测对照：hex32 → 9074；16 位数字 / 随机 hex → `code:0`；空串 → 9004。本项目的取法是 sha256(uid[#genN]) mod 10^16 补零（确定性，重试不会换号），9074 时换代次再试
 - SSE 事件序列：`metadata` → `timing_cost` → `output`×N → `extra_info` → `token_usage` → `done`
 - `token_usage` 含缓存字段 `cache_read_input_tokens` / `cache_creation_input_tokens`（未命中为 0，非缺失），映射为统计的 `cached_tokens`；**无 per-request credit**
 - 错误码 `1005` = 权益不足；仅流式，非流式需聚合

TECHNICAL.md

@@ -62,7 +62,7 @@ coding2api/
 │   │   ├── trae/
 │   │   │   ├── client.py        # SOLO 上游 + 双 httpx 客户端 + 额度探测
 │   │   │   ├── events.py        # 自定义 SSE → Event
-│   │   │   ├── credential.py    # 凭证解析（嵌套/扁平）+ 原子写回
+│   │   │   ├── credential.py    # 凭证解析（嵌套/扁平）+ 原子写回 + 签到设备号派生
 │   │   │   └── callback.py      # 登录 URL 构造 + 回调解析
 │   │   └── fixtures/            # fixture 清单
 │   │       ├── codebuddy/*.sse

src/provider/trae/client.py

@@ -31,7 +31,7 @@
     new_machine_identity,
     parse_callback_url,
 )
-from .credential import TraeCredential, parse_credential
+from .credential import TraeCredential, checkin_device_id, parse_credential
 from .events import (
     AGENT_HOST,
     APP_ID,
@@ -50,6 +50,8 @@
     UpstreamProtocolViolation,
 )
 
+# 9074 时的设备号轮换次数（同账号换不同的派生设备号，上游按 uid 记账不影响发放）
+CHECKIN_DEVICE_GENERATIONS = 3
 STREAM_TIMEOUT = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)
 SHORT_TIMEOUT = httpx.Timeout(30.0)
 
@@ -230,12 +232,17 @@ def solo_headers(credential: TraeCredential, *, stream: bool = True) -> dict[str
     return headers
 
 
-def ug_headers(credential: TraeCredential) -> dict[str, str]:
-    """签到/积分端点头（api.trae.cn）。对照原实现 UgHeaders。
+def ug_headers(credential: TraeCredential, *, device_id: str = "") -> dict[str, str]:
+    """签到/积分端点头（api.trae.cn）。
 
-    论坛实测（topic/180147）：设备头是签到 API 的隐藏必填项。
-    缺 X-Machine-Id / X-Device-Id 时 status 直接返回 9004（参数错误）；
-    补齐后请求格式正确，只剩 9074（服务器繁忙/限流）。uid 对应 auth.userId。
+    设备头是签到 API 的隐藏必填项（论坛实测 topic/180147）：缺 X-Device-Id 时
+    status 直接返回 9004（参数错误）。
+
+    **X-Device-Id 必须是 16 位纯数字**。用凭证自带的 deviceId（登录 URL 用的
+    hex32）调 claim 会稳定得到 9074「当前参与用户太多」——这个码看起来像限流，
+    实际是设备标识格式不符（实测：hex32 → 9074；16 位数字 / 随机 hex → code:0；
+    空串 → 9004）。默认取 `checkin_device_id(uid)` 的派生值，调用方可在 9074 时
+    传入轮换后的代次值。
     """
     headers = {
         "Content-Type": "application/json",
@@ -244,10 +251,12 @@ def ug_headers(credential: TraeCredential) -> dict[str, str]:
         "Authorization": f"Cloud-IDE-JWT {credential.access_token}",
         "X-User-Region": "CN",
     }
+    # machine_id 保持登录时那一对（它不是签到的校验项，换掉反而可能与登录态不匹配）
     if credential.machine_id:
         headers["X-Machine-Id"] = credential.machine_id
-    if credential.device_id:
-        headers["X-Device-Id"] = credential.device_id
+    effective = device_id or checkin_device_id(credential.uid) or credential.device_id
+    if effective:
+        headers["X-Device-Id"] = effective
     if credential.uid:
         headers["X-Uid"] = credential.uid
     return headers
@@ -388,20 +397,24 @@ 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]:
+    async def fetch_checkin_status(self, credential: TraeCredential,
+                                   *, generation: int = 0) -> 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))
+            f"{self.ug_host}{trae_events.EP_CHECKIN_STATUS}", {},
+            ug_headers(credential, device_id=checkin_device_id(credential.uid, generation)))
         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:
+    async def claim_checkin(self, credential: TraeCredential,
+                            *, generation: int = 0) -> dict[str, Any] | None:
         """checkin_credits/claim：领取当日积分。"""
         return await self._post_json(
-            f"{self.ug_host}{trae_events.EP_CHECKIN_CLAIM}", {}, ug_headers(credential))
+            f"{self.ug_host}{trae_events.EP_CHECKIN_CLAIM}", {},
+            ug_headers(credential, device_id=checkin_device_id(credential.uid, generation)))
 
     async def refresh_token(self, credential: TraeCredential) -> TraeCredential:
         """ExchangeToken；失败不改写原凭证字段。"""
@@ -603,25 +616,36 @@ async def checkin(self, credential_data: dict) -> CheckinResult:
         """TRAE 签到：先查状态，未签且可签才领取。
 
         上游没有独立的「已签到」错误码，status.checked_in 就是已签语义。
-        claim 可能返回 HTTP 200 + 业务码非 0 的软失败（如实测 9074
-        「当前参与用户太多」），必须解析 code，否则会误报成功。
+        claim 可能返回 HTTP 200 + 业务码非 0 的软失败，必须解析 code，否则会误报成功。
+
+        9074「当前参与用户太多」**不是限流，而是设备标识格式不符**（实测：hex32
+        → 9074，16 位数字 → success）。默认已经用派生的 16 位数字设备号，所以
+        正常不该再遇到；这里保留一次轮换代次重试，应对「派生值恰好在服务端侧不可用」
+        这种未知情况（同账号换设备号不影响领取结果，上游按 uid 记账）。
         """
         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="当前账号不可签到")
-        claim = await self.client.claim_checkin(credential)  # _post_json 保证 dict
-        code = claim.get("code")
-        if code not in (0, None):
+        for generation in range(CHECKIN_DEVICE_GENERATIONS):
+            status = await self.client.fetch_checkin_status(
+                credential, generation=generation)
+            if status["checked_in"]:
+                return CheckinResult(ok=True, credit=None, message="今天已签到",
+                                     already_checked_in=True)
+            if not status["enable"]:
+                return CheckinResult(ok=False, message="当前账号不可签到")
+            claim = await self.client.claim_checkin(credential, generation=generation)
+            code = claim.get("code")
+            if code in (0, None):
+                # 领取成功 → 回查 status 带回当前积分总额
+                latest = await self.client.fetch_checkin_status(
+                    credential, generation=generation)
+                return CheckinResult(ok=True, credit=latest.get("credits"), code=0)
+            if code == 9074 and generation + 1 < CHECKIN_DEVICE_GENERATIONS:
+                continue          # 换一个派生设备号再试（同账号其它代次）
             # 软失败：不抛异常（后台任务按失败计数，当日不封账，下轮重试）
             return CheckinResult(ok=False, code=int(code),
                                  message=str(claim.get("message") or "claim 失败"))
-        # 领取成功 → 回查 status 带回当前积分总额
-        latest = await self.client.fetch_checkin_status(credential)
-        return CheckinResult(ok=True, credit=latest.get("credits"), code=0)
+        # 循环必然在每轮内返回；保留兜底让静态检查满意
+        return CheckinResult(ok=False, message="签到未完成")  # pragma: no cover
 
     # ------------------------------------------------- callback 轨道（Q17=C）
 

src/provider/trae/credential.py

@@ -85,3 +85,25 @@ def parse_credential(raw: bytes | dict[str, Any]) -> TraeCredential:
     if not credential.access_token:
         raise UpstreamProtocolViolation("credential missing accessToken")
     return credential
+
+
+# 签到设备标识：必须是 **16 位纯数字**。用 hex32（登录 URL 里的 machine/device id）
+# 调 claim 会稳定得到 9074「当前参与用户太多」——这个码看起来像限流，实际是设备
+# 标识格式不符（实测：hex32 → 9074，16 位数字 / 随机 hex → code:0；空串 → 9004）。
+# 派生规则来自公开逆向（trae2api-more 的 CheckinDeviceID）：sha256(identity) 取模
+# 10^16 后补零成 16 位；generation > 0 时把代数并入摘要，用于 9074 时轮换设备。
+_CHECKIN_DEVICE_MODULUS = 10 ** 16
+
+
+def checkin_device_id(identity: str, generation: int = 0) -> str:
+    """派生签到的 X-Device-Id：16 位纯数字，由账号身份确定性地导出。
+
+    identity 为空时返回空串（调用方会退回凭证自带的 device_id）。
+    """
+    import hashlib
+
+    if not identity:
+        return ""
+    material = identity if generation <= 0 else f"{identity}#gen{generation}"
+    digest = hashlib.sha256(material.encode("utf-8")).digest()
+    return f"{int.from_bytes(digest, 'big') % _CHECKIN_DEVICE_MODULUS:016d}"

tests/test_m15_operations.py

@@ -3314,9 +3314,11 @@ async def test_trae_checkin_claim_soft_failure_and_success_paths():
     from src.provider.trae.client import TraeClient, TraeProvider
 
     seen: list[str] = []
+    device_ids: list[str] = []
 
     async def handler(request: httpx.Request) -> httpx.Response:
         seen.append(request.url.path)
+        device_ids.append(request.headers.get("x-device-id", ""))
         if request.url.path.endswith("status"):
             # 只有第二次（成功的）claim 之后才算已签
             checked = len([p for p in seen if p.endswith("claim")]) >= 2
@@ -3335,12 +3337,78 @@ async def handler(request: httpx.Request) -> httpx.Response:
     client = TraeClient(stream_client=_httpx.AsyncClient(transport=transport, timeout=None),
                         short_client=_httpx.AsyncClient(transport=transport, timeout=None))
     provider = TraeProvider(client=client)
-    data = {"bearer_token": "t", "device_id": "d"}
+    data = {"bearer_token": "t", "device_id": "d", "uid": "u"}
 
+    # 单轮内 9074 会轮换代次设备号重试，所以一轮就应成功；期间用的都是 16 位数字
     first = await provider.checkin(data)
-    assert not first.ok and first.code == 9074
-    assert "当前参与用户太多" in first.message
+    assert first.ok and first.code == 0 and first.credit == 200
+    # gen0: status + claim(9074)；gen1: status + claim(成功) + 回查 status
+    assert len(device_ids) == 5
+    assert len(set(device_ids)) == 2     # 轮换过一次设备号
+    assert all(d.isdigit() and len(d) == 16 for d in device_ids)
 
-    second = await provider.checkin(data)
-    assert second.ok and second.code == 0 and second.credit == 200
 
+
+async def test_trae_checkin_client_returns_soft_failure_after_all_generations():
+    """所有代次设备号都 9074 → 如实返回软失败（不误报成功、不无限重试）。"""
+    from src.provider.trae.client import TraeClient
+
+    claims: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("status"):
+            return httpx.Response(200, json={"checked_in": False, "credits": 150,
+                                             "enable": True, "code": 0})
+        claims.append(request.headers.get("x-device-id", ""))
+        return httpx.Response(200, json={"code": 9074, "message": "当前参与用户太多"})
+
+    import httpx as _httpx
+    transport = _httpx.MockTransport(handler)
+    client = TraeClient(stream_client=_httpx.AsyncClient(transport=transport, timeout=None),
+                        short_client=_httpx.AsyncClient(transport=transport, timeout=None))
+    from src.provider.trae.client import CHECKIN_DEVICE_GENERATIONS, TraeProvider
+
+    provider = TraeProvider(client=client)
+    result = await provider.checkin({"bearer_token": "t", "device_id": "d", "uid": "u"})
+    assert result.ok is False and result.code == 9074
+    assert "当前参与用户太多" in result.message
+    assert len(claims) == CHECKIN_DEVICE_GENERATIONS      # 每代各试一次就收手
+    assert len(set(claims)) == CHECKIN_DEVICE_GENERATIONS  # 每次换设备号
+
+
+async def test_trae_checkin_not_enabled_short_circuits():
+    """enable=False → 直接返回不可签到，不发 claim。"""
+    from src.provider.trae.client import TraeClient
+
+    claims: list[str] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("status"):
+            return httpx.Response(200, json={"checked_in": False, "credits": 0,
+                                             "enable": False, "code": 0})
+        claims.append(request.url.path)
+        return httpx.Response(200, json={"code": 0})
+
+    import httpx as _httpx
+    transport = _httpx.MockTransport(handler)
+    client = TraeClient(stream_client=_httpx.AsyncClient(transport=transport, timeout=None),
+                        short_client=_httpx.AsyncClient(transport=transport, timeout=None))
+    from src.provider.trae.client import TraeProvider
+
+    result = await TraeProvider(client=client).checkin({"bearer_token": "t", "uid": "u"})
+    assert result.ok is False and result.message == "当前账号不可签到"
+    assert claims == []
+
+
+def test_checkin_device_id_is_deterministic_and_generation_sensitive():
+    """设备号派生：确定性、16 位数字、按代次变化、身份为空返回空串。"""
+    from src.provider.trae.credential import checkin_device_id
+
+    first = checkin_device_id("u1")
+    assert first == checkin_device_id("u1")               # 确定性（重试不会换号）
+    assert first.isdigit() and len(first) == 16
+    assert checkin_device_id("u1", 0) == first            # gen0 等价
+    assert checkin_device_id("u1", 1) != first            # 代次不同则设备号不同
+    assert checkin_device_id("u2") != first
+    assert checkin_device_id("", 0) == ""
+    assert checkin_device_id("u1", -1) == first           # 负代次按 0 处理

tests/test_m1a_gaps.py

@@ -21,6 +21,7 @@
     TraeCredential,
     prepare_body,
 )
+from src.provider.trae.credential import checkin_device_id
 from src.provider.trae.events import UpstreamProtocolViolation
 from tests.conftest import SECRET
 
@@ -247,10 +248,20 @@ def test_ug_headers_carry_region_and_auth():
                                         machine_id="m", uid="u"))
     assert headers["Authorization"] == "Cloud-IDE-JWT a"
     assert headers["X-User-Region"] == "CN"
-    assert headers["X-Device-Id"] == "d"
     assert headers["X-Machine-Id"] == "m"
     assert headers["X-Uid"] == "u"
     assert "X-Cloudide-Token" not in headers
+    # X-Device-Id 必须是 16 位纯数字：凭证自带的 hex32 会让 claim 返回 9074
+    # （实测 hex32 → 9074，16 位数字 → code:0），所以默认用派生值而非 device_id
+    assert headers["X-Device-Id"] == checkin_device_id("u", 0)
+    assert headers["X-Device-Id"].isdigit() and len(headers["X-Device-Id"]) == 16
+    # 显式传入时以传入值为准（供 9074 轮换代次使用）
+    assert ug_headers(TraeCredential(access_token="a", uid="u"),
+                      device_id="123")["X-Device-Id"] == "123"
+    # uid 为空 → 退回凭证自带的 device_id
+    assert ug_headers(TraeCredential(access_token="a", device_id="d"))["X-Device-Id"] == "d"
+    # 两者都空 → 不带该头（上游会返回 9004，不会静默当成有效请求）
+    assert "X-Device-Id" not in ug_headers(TraeCredential(access_token="a"))
 
 
 async def test_client_aclose_is_idempotent():
@@ -700,8 +711,8 @@ def handler(request: httpx.Request) -> httpx.Response:
         # httpx 内部存储全小写
         assert headers.get("x-user-region") == "CN"
         assert headers.get("authorization") == "Cloud-IDE-JWT a"
-        # 论坛实测（topic/180147）：设备头是签到 API 的隐藏必填项
-        assert headers.get("x-device-id") == "d"
+        # 设备头是隐藏必填项；值必须是 16 位数字（hex32 会触发 9074）
+        assert headers.get("x-device-id") == checkin_device_id("u", 0)
         assert headers.get("x-machine-id") == "m"
         assert headers.get("x-uid") == "u"