feat: 双渠道共享节流器 + 流内错误全量日志

用户 PI 连续切换多模型仍偶发 400。日志时间线还原调度序列：
7 连成功 → TRAE 流内错误（非 4001/1005 码）累计 OTHER 达阈值 →
TRAE 冷却 10 分钟 → 后续请求只剩 CB → CB 频率风控 11128 →
双渠道同时不可用 → 400。该冷却全程无日志无记录。

- 流内 ERROR 的 OTHER 路径补 logger.warning（code/message 全记录），
  INVALID 分流补日志——调度决策完全可观测
- TRAE Provider 也接入 pacer（与 CB 共享同一实例）：
  两渠道请求共同保持最小间隔，任一渠道都不会被连续请求打爆
- 已重置凭证冷却/计数

复测：hy3 / qwen3.8-max / glm-5.3-flash / deepseek-v4-flash /
hy4-preview 五模型连续切换全部成功，零 11128。
后端 636 测试 / 覆盖 100%。

Author: RobbsLuo

src/engine/executor.py

@@ -126,8 +126,26 @@ async def stream(self, request: ChatRequest, *, username: str = "unknown"
                     credential_data, request.raw, self._upstream_model(provider_id, target.model)
                 ):
                     if event.kind is EventKind.ERROR:
+                        kind = _event_kind(event)
+                        if kind is ErrKind.INVALID:
+                            # 流内 4001 等参数/模型错误：换凭证没用，跳过该上游
+                            logger.warning(
+                                "上游 %s 流内拒绝模型 %s（凭证 %s 跳过）: code=%s %s",
+                                provider_id, target.model, credential_id,
+                                event.error_code, event.error_message)
+                            self._record_invalid(username, provider_id, credential_id,
+                                                 target.model, started, event)
+                            yield _error_frame(
+                                _reject_message(target.model, last_error,
+                                                self._suggestions(target.model)),
+                                "invalid_request")
+                            return
+                        logger.warning(
+                            "上游 %s 流内错误（凭证 %s，kind=%s）: code=%s %s",
+                            provider_id, credential_id, kind,
+                            event.error_code, event.error_message)
                         outcome = self._deps.scheduler.note_error(
-                            self._candidate(credential_id), _event_kind(event), int(time.time()))
+                            self._candidate(credential_id), kind, int(time.time()))
                         self._deps.credentials.save_error(credential_id, outcome)
                         last_error = UpstreamStreamError(event)
                         break

src/main.py

@@ -69,8 +69,10 @@ def build_app(settings: Settings | None = None, *, providers: dict | None = None
         else Pacer(config.codebuddy_chat_min_interval,
                    config.codebuddy_chat_min_interval)
     )
+    # TRAE/CB 共享同一 pacer：两渠道请求共同保持最小间隔，
+    # 避开各自的频率风控（CB 11128 / TRAE 流内错误）
     registry = providers if providers is not None else {
-        "trae": TraeProvider(),
+        "trae": TraeProvider(pacer=chat_pacer),
         "codebuddy": CodeBuddyProvider(pacer=chat_pacer),
     }
     # provider → {小写模型名: 上游原始 id}；playground_models 拉取后就地更新，

src/provider/trae/client.py

@@ -392,6 +392,7 @@ class TraeProvider:
     """Provider 协议实现（细接口，Q16=A）。"""
 
     client: TraeClient = field(default_factory=TraeClient)
+    pacer: Any | None = None
 
     id: str = "trae"
 
@@ -431,6 +432,8 @@ async def stream_chat(self, credential_data: dict, payload: dict,
                           model: str) -> AsyncIterator[Event]:
         """引擎调用入口：dict 凭证 → 上游流 → 中立事件。"""
         credential = TraeCredential.from_dict(credential_data)
+        if self.pacer is not None:
+            await self.pacer.wait_turn()
         async for event in self.client.stream_chat(credential, payload, model):
             yield event
 

tests/test_m1a_trae.py

@@ -929,3 +929,29 @@ def test_prepare_body_tools_edge_cases():
         {"type": "function", "function": {"name": "f", "parameters": "{}"}},
     ]}, "m")
     assert body["tools"][0]["function"]["parameters"] == "{}"
+
+
+async def test_trae_pacer_wait_and_disable():
+    """TRAE stream_chat 前等待 pacer；None 不等待（与 CB 对称）。"""
+    import time as _time
+
+    from src.provider.trae.client import TraeProvider
+
+    waited = []
+
+    class FakePacer:
+        async def wait_turn(self):
+            waited.append(_time.monotonic())
+
+    class FakeClient:
+        async def stream_chat(self, cred, payload, model):
+            yield Event(kind=EventKind.CONTENT, content="ok")
+
+    provider = TraeProvider(client=FakeClient(), pacer=FakePacer())
+    events = [e async for e in provider.stream_chat({"accessToken": "a"}, {}, "m")]
+    assert events and waited
+
+    provider2 = TraeProvider(client=FakeClient(), pacer=None)
+    waited.clear()
+    _ = [e async for e in provider2.stream_chat({"accessToken": "a"}, {}, "m")]
+    assert not waited

tests/test_m1b_codebuddy.py

@@ -1101,3 +1101,67 @@ async def stream_chat(self, cred, payload, model):
     waited.clear()
     _ = [e async for e in provider2.stream_chat({"bearer_token": "t"}, {}, "m")]
     assert not waited
+
+
+async def test_stream_inner_error_other_is_logged_and_counted(dual_repo, caplog):
+    """流内非 4001/1005 错误 → 记冷却并打日志（可观测性回归）。"""
+    import logging
+
+    from src.provider.base import Event
+
+    repo, _db = dual_repo
+    repo.add(provider="codebuddy", credential_data={"bearer_token": "cb"})
+    calls = {"n": 0}
+
+    async def gen(_cred, _payload, _model):
+        calls["n"] += 1
+        yield Event(kind=EventKind.ERROR, error_code=41291, error_message="rate limited")
+
+    class P:
+        id = "codebuddy"
+        stream_chat = staticmethod(gen)
+
+    executor = Executor(ExecutorDeps(
+        providers={"codebuddy": P()}, credentials=repo,
+        scheduler=Scheduler(max_rotate=1), default_model="m"))
+    frames = [f async for f in executor.stream(_request("m"), username="u")]
+    assert any(b"error" in f for f in frames)
+    assert calls["n"] == 1
+    assert any("流内错误" in r.getMessage() for r in caplog.records
+               if r.levelno == logging.WARNING)
+
+
+async def test_stream_inner_4001_yields_invalid_frame(dual_repo, caplog):
+    """流内 4001 → INVALID：跳过上游 + invalid_request 帧。"""
+    import json as _json
+    import logging
+
+    from src.provider.base import Event
+
+    repo, db = dual_repo
+    repo.add(provider="codebuddy", credential_data={"bearer_token": "cb"})
+    calls = {"n": 0}
+
+    async def gen(_cred, _payload, _model):
+        calls["n"] += 1
+        yield Event(kind=EventKind.ERROR, error_code=4001, error_message="param invalid")
+
+    class P:
+        id = "codebuddy"
+        stream_chat = staticmethod(gen)
+
+    executor = Executor(ExecutorDeps(
+        providers={"codebuddy": P()}, credentials=repo,
+        scheduler=Scheduler(max_rotate=1), default_model="m",
+        model_suggestions=lambda _n: ["alt-model"]))
+
+    with caplog.at_level(logging.WARNING):
+        frames = [f async for f in executor.stream(_request("m"), username="u")]
+    payload = _json.loads(frames[0].split(b"\n\n")[0].removeprefix(b"data: "))
+    assert payload["error"]["code"] == "invalid_request"
+    assert "alt-model" in payload["error"]["message"]
+    assert calls["n"] == 1                              # 不轮换
+    row = db.connect().execute(
+        "SELECT err_count, cooling_until FROM credentials").fetchone()
+    assert tuple(row) == (0, None)                      # 不冷却
+    assert any("流内拒绝模型" in r.getMessage() for r in caplog.records)