feat(proxy,billing,pool): 增强隧道/流中断诊断日志、修复缓存 TTL 差异化计价

- stream_processor: 上游流中断时记录完整异常链与已传输 token 统计
- hub_transport: 断连影响 in-flight 流、STREAM_ERROR、超时场景补充 warning 日志；
  未启用时跳过重连循环，重连失败日志降频避免刷屏
- tunnel_manager: 流超时/错误/全部取消/STREAM_ERROR 增加诊断日志与字节统计
- billing_integration: 计费时自动补全 cache_ttl_minutes（从 provider key 查询
  或从 5m/1h 细分 token 回推），修复缓存 TTL 差异化计价缺失
- PoolManagement.vue: 移除重复的账号告警 Badge
- 新增 billing integration 单元测试

frontend/src/views/admin/PoolManagement.vue

@@ -484,14 +484,6 @@
                           {{ getKeyOAuthExpires(key)?.text }}
                         </span>
                       </template>
-                      <Badge
-                        v-if="getAccountAlertLabel(key)"
-                        variant="destructive"
-                        class="text-[9px] px-1 py-0 h-4 shrink-0"
-                        :title="getAccountAlertTitle(key)"
-                      >
-                        {{ getAccountAlertLabel(key) }}
-                      </Badge>
                       <Badge
                         v-if="key.oauth_plan_type"
                         variant="outline"
@@ -794,14 +786,6 @@
                       {{ getKeyOAuthExpires(key)?.text }}
                     </span>
                   </template>
-                  <Badge
-                    v-if="getAccountAlertLabel(key)"
-                    variant="destructive"
-                    class="text-[9px] px-1 py-0 h-4 shrink-0"
-                    :title="getAccountAlertTitle(key)"
-                  >
-                    {{ getAccountAlertLabel(key) }}
-                  </Badge>
                   <Badge
                     v-if="key.oauth_plan_type"
                     variant="outline"

src/api/handlers/base/stream_processor.py

@@ -1057,6 +1057,31 @@ class StreamProcessor:
         except GeneratorExit:
             raise
         except (httpx.StreamClosed, httpx.HTTPError) as exc:
+            # 记录上游流中断的详细诊断信息
+            elapsed = time.monotonic() - start_time if start_time else 0
+            exc_chain = []
+            seen: set[int] = set()
+            cause: BaseException | None = exc
+            while cause is not None and id(cause) not in seen:
+                seen.add(id(cause))
+                exc_chain.append(f"{type(cause).__name__}: {cause}")
+                cause = cause.__cause__ or cause.__context__
+            logger.warning(
+                "[{}] upstream stream error: provider={}, model={}, "
+                "yielded_any={}, has_completion={}, elapsed={:.1f}s, "
+                "input_tokens={}, output_tokens={}, "
+                "exception_chain=[{}]",
+                self.request_id,
+                ctx.provider_name,
+                ctx.model,
+                yielded_any,
+                ctx.has_completion,
+                elapsed,
+                ctx.input_tokens,
+                ctx.output_tokens,
+                " -> ".join(exc_chain),
+            )
+
             # 连接关闭/协议错误：best-effort flush 残留 SSE，避免丢失尾部 usage。
             try:
                 if buffer:

src/services/proxy_node/hub_transport.py

@@ -131,6 +131,8 @@ class HubConnectionManager:
     def _start_reconnect_loop(self) -> None:
         if self._closing:
             return
+        if not self._config.enabled:
+            return
         if self._reconnect_task is not None and not self._reconnect_task.done():
             return
         self._reconnect_task = asyncio.create_task(self._reconnect_loop())
@@ -151,7 +153,8 @@ class HubConnectionManager:
                     break
             except Exception as e:
                 attempt += 1
-                logger.debug("Hub reconnect attempt {} failed: {}", attempt, e)
+                if attempt <= 3 or attempt % 10 == 0 or attempt in (20, 50, 100):
+                    logger.debug("Hub reconnect attempt {} failed: {}", attempt, e)
 
     async def _handle_disconnect(
         self,
@@ -175,6 +178,15 @@ class HubConnectionManager:
                 pass
 
         if self._pending_streams:
+            affected_count = len(self._pending_streams)
+            affected_ids = list(self._pending_streams.keys())[:10]  # 最多记录 10 个
+            logger.warning(
+                "Hub disconnect affecting {} in-flight streams: reason={}, stream_ids={}{}",
+                affected_count,
+                reason,
+                affected_ids,
+                "..." if affected_count > 10 else "",
+            )
             for state in self._pending_streams.values():
                 state.set_error("hub disconnected")
             self._pending_streams.clear()
@@ -290,6 +302,11 @@ class HubConnectionManager:
                     message = (
                         frame.payload.decode(errors="replace") if frame.payload else "stream error"
                     )
+                    logger.warning(
+                        "Hub received STREAM_ERROR: stream_id={}, message={}",
+                        frame.stream_id,
+                        message[:500],
+                    )
                     stream.set_error(message)
 
             # -- connection-level frames --
@@ -599,11 +616,30 @@ class HubTunnelTransport(httpx.AsyncBaseTransport):
                 stream=HubResponseStream(manager, stream_state, timeout=self._timeout),
             )
         except TunnelStreamError as e:
+            stream_id = stream_state.stream_id if stream_state else None
+            has_headers = bool(stream_state and stream_state.status > 0)
+            logger.warning(
+                "HubTunnelTransport error: node_id={}, url={}, stream_id={}, "
+                "has_headers={}, error={}",
+                self._node_id,
+                str(request.url),
+                stream_id,
+                has_headers,
+                e,
+            )
             self._cleanup_stream(manager, stream_state)
-            if stream_state and stream_state.status > 0:
+            if has_headers:
                 raise httpx.ReadError(str(e)) from e
             raise httpx.ConnectError(str(e)) from e
         except asyncio.TimeoutError:
+            stream_id = stream_state.stream_id if stream_state else None
+            logger.warning(
+                "HubTunnelTransport timeout: node_id={}, url={}, stream_id={}, timeout={:.0f}s",
+                self._node_id,
+                str(request.url),
+                stream_id,
+                self._timeout,
+            )
             self._cleanup_stream(manager, stream_state)
             raise httpx.ReadTimeout("hub tunnel request timeout") from None
 

src/services/proxy_node/tunnel_manager.py

@@ -112,6 +112,13 @@ class TunnelConnection:
         raise TunnelStreamError("stream ID space exhausted")
 
     def cancel_all_streams(self) -> None:
+        if self._pending_streams:
+            logger.warning(
+                "tunnel cancel_all_streams: node_id={}, name={}, count={}",
+                self.node_id,
+                self.node_name,
+                len(self._pending_streams),
+            )
         for state in self._pending_streams.values():
             state.set_error("tunnel disconnected")
         self._pending_streams.clear()
@@ -169,17 +176,37 @@ class _StreamState:
             raise TunnelStreamError(self._error)
 
     async def iter_body(self, chunk_timeout: float = 60.0) -> AsyncGenerator[bytes, None]:
+        chunks_received = 0
+        total_bytes = 0
         while True:
             try:
                 chunk = await asyncio.wait_for(self._body_chunks.get(), timeout=chunk_timeout)
             except asyncio.TimeoutError:
                 self._error = "body chunk timeout"
                 self._done_event.set()
+                logger.warning(
+                    "tunnel stream body chunk timeout: stream_id={}, "
+                    "chunk_timeout={:.0f}s, chunks_received={}, total_bytes={}",
+                    self.stream_id,
+                    chunk_timeout,
+                    chunks_received,
+                    total_bytes,
+                )
                 raise TunnelStreamError("body chunk timeout")
             if chunk is None:
                 if self._error:
+                    logger.warning(
+                        "tunnel stream ended with error: stream_id={}, error={}, "
+                        "chunks_received={}, total_bytes={}",
+                        self.stream_id,
+                        self._error,
+                        chunks_received,
+                        total_bytes,
+                    )
                     raise TunnelStreamError(self._error)
                 return
+            chunks_received += 1
+            total_bytes += len(chunk)
             yield chunk
 
 
@@ -439,6 +466,12 @@ class TunnelManager:
         elif frame.msg_type == MsgType.STREAM_ERROR:
             if stream:
                 msg = frame.payload.decode(errors="replace") if frame.payload else "stream error"
+                logger.warning(
+                    "tunnel received STREAM_ERROR: node={}, stream_id={}, message={}",
+                    conn.node_name,
+                    frame.stream_id,
+                    msg[:500],
+                )
                 stream.set_error(msg)
                 conn.remove_stream(frame.stream_id)
 

src/services/usage/_billing_integration.py

@@ -68,6 +68,41 @@ class UsageBillingIntegrationMixin:
         from src.services.billing.service import BillingService
 
         request_count = 0 if is_failed_request else 1
+        has_cache_tokens = bool(
+            params.cache_creation_input_tokens > 0 or params.cache_read_input_tokens > 0
+        )
+        effective_cache_ttl_minutes = params.cache_ttl_minutes
+
+        # 主链路很多场景不会显式传 cache_ttl_minutes，这里补全以确保 1h/5m TTL 差异化计价生效。
+        if effective_cache_ttl_minutes is None and has_cache_tokens and params.provider_api_key_id:
+            try:
+                from src.models.database import ProviderAPIKey
+
+                key_ttl = (
+                    params.db.query(ProviderAPIKey.cache_ttl_minutes)
+                    .filter(ProviderAPIKey.id == params.provider_api_key_id)
+                    .scalar()
+                )
+                if key_ttl is not None:
+                    key_ttl_int = int(key_ttl)
+                    if key_ttl_int >= 0:
+                        effective_cache_ttl_minutes = key_ttl_int
+            except Exception:
+                # Best-effort fallback below.
+                pass
+
+        # 无法从 key 获取时，尽量从 5m/1h 细分回推（主要覆盖 Claude cache_creation）。
+        if effective_cache_ttl_minutes is None and has_cache_tokens:
+            t5m = int(params.cache_creation_input_tokens_5m or 0)
+            t1h = int(params.cache_creation_input_tokens_1h or 0)
+            if t1h > 0 and t5m == 0:
+                effective_cache_ttl_minutes = 60
+            elif t5m > 0 and t1h == 0:
+                effective_cache_ttl_minutes = 5
+            elif t1h > 0:
+                # 混合场景优先按长 TTL 计，避免 1h 缓存被按 5m 误计。
+                effective_cache_ttl_minutes = 60
+
         dims: dict[str, Any] = {
             "input_tokens": input_tokens_for_billing,
             "output_tokens": params.output_tokens,
@@ -75,8 +110,8 @@ class UsageBillingIntegrationMixin:
             "cache_read_input_tokens": params.cache_read_input_tokens,
             "request_count": request_count,
         }
-        if params.cache_ttl_minutes is not None:
-            dims["cache_ttl_minutes"] = params.cache_ttl_minutes
+        if effective_cache_ttl_minutes is not None:
+            dims["cache_ttl_minutes"] = effective_cache_ttl_minutes
         # If tiered pricing is disabled, force first tier by using tier-key=0.
         if not params.use_tiered_pricing:
             dims["total_input_context"] = 0

tests/services/test_usage_billing_integration.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.services.usage._billing_integration import UsageBillingIntegrationMixin
+from src.services.usage._types import UsageRecordParams
+
+
+class _TestUsageBillingIntegration(UsageBillingIntegrationMixin):
+    @classmethod
+    async def _get_rate_multiplier_and_free_tier(
+        cls,
+        db: Any,  # noqa: ARG003
+        provider_api_key_id: str | None,  # noqa: ARG003
+        provider_id: str | None,  # noqa: ARG003
+        api_format: str | None = None,  # noqa: ARG003
+    ) -> tuple[float, bool]:
+        return 1.0, False
+
+
+class _DummyBillingService:
+    last_dimensions: dict[str, Any] | None = None
+
+    def __init__(self, db: Any) -> None:  # noqa: D107, ARG002
+        pass
+
+    def calculate(
+        self,
+        *,
+        task_type: str,  # noqa: ARG002
+        model: str,  # noqa: ARG002
+        provider_id: str,  # noqa: ARG002
+        dimensions: dict[str, Any],
+        strict_mode: bool | None,  # noqa: ARG002
+    ) -> Any:
+        _DummyBillingService.last_dimensions = dict(dimensions)
+        snapshot = SimpleNamespace(
+            cost_breakdown={
+                "input_cost": 0.0,
+                "output_cost": 0.0,
+                "cache_creation_cost": 0.0,
+                "cache_read_cost": 0.0,
+                "request_cost": 0.0,
+            },
+            total_cost=0.0,
+            resolved_variables={},
+            to_dict=lambda: {},
+        )
+        return SimpleNamespace(snapshot=snapshot)
+
+
+def _build_params(
+    db: Any,
+    *,
+    cache_creation_input_tokens: int = 0,
+    cache_read_input_tokens: int = 0,
+    cache_creation_input_tokens_5m: int = 0,
+    cache_creation_input_tokens_1h: int = 0,
+    cache_ttl_minutes: int | None = None,
+    provider_api_key_id: str | None = "pak-test",
+) -> UsageRecordParams:
+    return UsageRecordParams(
+        db=db,
+        user=None,
+        api_key=None,
+        provider="provider-x",
+        model="claude-sonnet",
+        input_tokens=100,
+        output_tokens=50,
+        cache_creation_input_tokens=cache_creation_input_tokens,
+        cache_read_input_tokens=cache_read_input_tokens,
+        cache_creation_input_tokens_5m=cache_creation_input_tokens_5m,
+        cache_creation_input_tokens_1h=cache_creation_input_tokens_1h,
+        request_type="chat",
+        api_format="claude:chat",
+        api_family="claude",
+        endpoint_kind="chat",
+        endpoint_api_format="claude:chat",
+        has_format_conversion=False,
+        is_stream=False,
+        response_time_ms=123,
+        first_byte_time_ms=None,
+        status_code=200,
+        error_message=None,
+        metadata={},
+        request_headers=None,
+        request_body=None,
+        provider_request_headers=None,
+        provider_request_body=None,
+        response_headers=None,
+        client_response_headers=None,
+        response_body=None,
+        client_response_body=None,
+        request_id="req-test",
+        provider_id="provider-id",
+        provider_endpoint_id="endpoint-id",
+        provider_api_key_id=provider_api_key_id,
+        status="completed",
+        cache_ttl_minutes=cache_ttl_minutes,
+        use_tiered_pricing=True,
+        target_model=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_prepare_usage_record_uses_provider_key_ttl_for_cache_read(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    db = MagicMock()
+    db.query.return_value.filter.return_value.scalar.return_value = 60
+
+    monkeypatch.setattr("src.services.billing.service.BillingService", _DummyBillingService)
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.sanitize_request_metadata",
+        lambda metadata: metadata,
+    )
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.build_usage_params",
+        lambda **kwargs: {"total_cost_usd": 0.0, "actual_total_cost_usd": 0.0},
+    )
+
+    params = _build_params(db, cache_read_input_tokens=321)
+    await _TestUsageBillingIntegration._prepare_usage_record(params)
+
+    assert _DummyBillingService.last_dimensions is not None
+    assert _DummyBillingService.last_dimensions.get("cache_ttl_minutes") == 60
+
+
+@pytest.mark.asyncio
+async def test_prepare_usage_record_prefers_explicit_cache_ttl(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    db = MagicMock()
+    db.query.return_value.filter.return_value.scalar.return_value = 60
+
+    monkeypatch.setattr("src.services.billing.service.BillingService", _DummyBillingService)
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.sanitize_request_metadata",
+        lambda metadata: metadata,
+    )
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.build_usage_params",
+        lambda **kwargs: {"total_cost_usd": 0.0, "actual_total_cost_usd": 0.0},
+    )
+
+    params = _build_params(db, cache_read_input_tokens=123, cache_ttl_minutes=5)
+    await _TestUsageBillingIntegration._prepare_usage_record(params)
+
+    assert _DummyBillingService.last_dimensions is not None
+    assert _DummyBillingService.last_dimensions.get("cache_ttl_minutes") == 5
+
+
+@pytest.mark.asyncio
+async def test_prepare_usage_record_infers_ttl_from_1h_cache_split(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    db = MagicMock()
+
+    monkeypatch.setattr("src.services.billing.service.BillingService", _DummyBillingService)
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.sanitize_request_metadata",
+        lambda metadata: metadata,
+    )
+    monkeypatch.setattr(
+        "src.services.usage._billing_integration.build_usage_params",
+        lambda **kwargs: {"total_cost_usd": 0.0, "actual_total_cost_usd": 0.0},
+    )
+
+    params = _build_params(
+        db,
+        provider_api_key_id=None,
+        cache_creation_input_tokens=1000,
+        cache_creation_input_tokens_1h=1000,
+    )
+    await _TestUsageBillingIntegration._prepare_usage_record(params)
+
+    assert _DummyBillingService.last_dimensions is not None
+    assert _DummyBillingService.last_dimensions.get("cache_ttl_minutes") == 60