fix: 用量记录保留空 headers/body，格式转换支持同 normalizer 透传

- 用量记录和遥测写入中将 truthiness 检查改为 is not None，
  避免空 dict/list 被误丢弃
- FormatConversionRegistry 新增 _same_normalizer() 方法，
  共享 normalizer 的格式（如 claude:chat/claude:cli）视为同格式透传
- 前端请求详情抽屉按数据可用性智能选择默认数据源
- 视频任务 finalize 支持 headers/original_headers 等别名字段

frontend/src/features/usage/components/RequestDetailDrawer.vue

@@ -709,12 +709,7 @@ watch(activeTab, (newTab) => {
   if (!['request-body', 'response-body'].includes(newTab)) {
     contentViewMode.value = 'json'
   }
-  // 请求头/体默认选中服务端，响应头/体默认选中客户端
-  if (['request-headers', 'request-body'].includes(newTab)) {
-    dataSource.value = 'provider'
-  } else if (['response-headers', 'response-body'].includes(newTab)) {
-    dataSource.value = 'client'
-  }
+  dataSource.value = getDefaultDataSourceForTab(newTab)
 })
 
 // 检测暗色模式
@@ -728,10 +723,14 @@ const hasProviderHeaders = computed(() => {
          Object.keys(detail.value.provider_request_headers).length > 0)
 })
 
-// 检测是否有提供商请求体（格式转换后的）
-const hasProviderBody = computed(() => {
-  return !!(detail.value?.provider_request_body && hasContent(detail.value.provider_request_body))
-    || !!(detail.value?.client_response_body && hasContent(detail.value.client_response_body))
+// 请求体：仅当 provider_request_body 存在时才展示来源切换
+const hasProviderRequestBody = computed(() => {
+  return hasContent(detail.value?.provider_request_body)
+})
+
+// 响应体：只有客户端侧和 provider 侧都存在时才展示来源切换
+const hasProviderResponseBody = computed(() => {
+  return hasContent(detail.value?.response_body) && hasContent(detail.value?.client_response_body)
 })
 
 // 检测是否有两套响应头（客户端侧 + 提供商侧）
@@ -746,7 +745,8 @@ const hasProviderResponseHeaders = computed(() => {
 const showDataSourceToggle = computed(() => {
   if (activeTab.value === 'request-headers') return hasProviderHeaders.value
   if (activeTab.value === 'response-headers') return hasProviderResponseHeaders.value
-  if (['request-body', 'response-body'].includes(activeTab.value)) return hasProviderBody.value
+  if (activeTab.value === 'request-body') return hasProviderRequestBody.value
+  if (activeTab.value === 'response-body') return hasProviderResponseBody.value
   return false
 })
 
@@ -785,9 +785,17 @@ const currentResponseBody = computed(() => {
 // 获取当前数据源的请求头数据
 const currentHeaderData = computed(() => {
   if (!detail.value) return null
-  return dataSource.value === 'client'
-    ? detail.value.request_headers
-    : detail.value.provider_request_headers
+  if (dataSource.value === 'client' && hasContent(detail.value.request_headers)) {
+    return detail.value.request_headers
+  }
+  if (dataSource.value === 'provider' && hasContent(detail.value.provider_request_headers)) {
+    return detail.value.provider_request_headers
+  }
+  // 回退：优先 client，再 provider
+  if (hasContent(detail.value.request_headers)) {
+    return detail.value.request_headers
+  }
+  return detail.value.provider_request_headers
 })
 
 // 请求体渲染结果
@@ -945,6 +953,40 @@ function hasContent(data: unknown): boolean {
   return true
 }
 
+function getDefaultDataSourceForTab(tab: string): 'client' | 'provider' {
+  if (!detail.value) {
+    if (['request-headers', 'request-body'].includes(tab)) return 'provider'
+    if (['response-headers', 'response-body'].includes(tab)) return 'client'
+    return dataSource.value
+  }
+
+  if (tab === 'request-headers') {
+    if (hasContent(detail.value.provider_request_headers)) return 'provider'
+    if (hasContent(detail.value.request_headers)) return 'client'
+    return 'provider'
+  }
+
+  if (tab === 'request-body') {
+    if (hasContent(detail.value.provider_request_body)) return 'provider'
+    if (hasContent(detail.value.request_body)) return 'client'
+    return 'provider'
+  }
+
+  if (tab === 'response-headers') {
+    if (hasContent(detail.value.client_response_headers)) return 'client'
+    if (hasContent(detail.value.response_headers)) return 'provider'
+    return 'client'
+  }
+
+  if (tab === 'response-body') {
+    if (hasContent(detail.value.client_response_body)) return 'client'
+    if (hasContent(detail.value.response_body)) return 'provider'
+    return 'client'
+  }
+
+  return dataSource.value
+}
+
 // 获取当前数据源的响应头数据
 const currentResponseHeaderData = computed(() => {
   if (!detail.value) return null
@@ -1018,11 +1060,8 @@ async function loadDetail(id: string, silent = false) {
       }
     }
 
-    // 根据数据可用性自动选择请求头数据源
-    // provider_request_headers 在 streaming 完成后才写入，pending/streaming 期间为空
-    const hasProviderReqHeaders = detail.value.provider_request_headers &&
-      Object.keys(detail.value.provider_request_headers).length > 0
-    dataSource.value = hasProviderReqHeaders ? 'provider' : 'client'
+    // 根据当前 Tab 的数据可用性自动选择默认数据源
+    dataSource.value = getDefaultDataSourceForTab(activeTab.value)
 
     // 使用请求记录中保存的历史价格
     if (detail.value.input_price_per_1m || detail.value.output_price_per_1m || detail.value.price_per_request) {

src/core/api_format/conversion/registry.py

@@ -77,6 +77,16 @@ class FormatConversionRegistry:
             raise FormatConversionError(format_id, format_id, f"未注册 Normalizer: {format_id}")
         return normalizer
 
+    def _same_normalizer(self, source_format: str, target_format: str) -> bool:
+        """判断两个 format_id 是否解析到同一个 normalizer 实例（即底层数据格式相同，可直接透传）。
+        例如 claude:chat / claude:cli 共享 ClaudeNormalizer，gemini:chat / gemini:cli 共享 GeminiNormalizer。
+        """
+        if str(source_format).upper() == str(target_format).upper():
+            return True
+        src = self.get_normalizer(source_format)
+        tgt = self.get_normalizer(target_format)
+        return src is not None and src is tgt
+
     def _repair_internal_tool_call_ids(self, internal: InternalRequest) -> None:
         """修复 InternalRequest 中空的 tool id/tool_use_id，避免上游校验报错。"""
 
@@ -122,11 +132,11 @@ class FormatConversionRegistry:
         target_variant: str | None = None,
         output_limit: int | None = None,
     ) -> dict[str, Any]:
-        if str(source_format).upper() == str(target_format).upper() and not target_variant:
+        if self._same_normalizer(source_format, target_format) and not target_variant:
             return request
 
-        # 同格式 + variant: 优先尝试轻量补丁（跳过 internal 转换）
-        if str(source_format).upper() == str(target_format).upper() and target_variant:
+        # 同 normalizer + variant: 优先尝试轻量补丁（跳过 internal 转换）
+        if self._same_normalizer(source_format, target_format) and target_variant:
             normalizer = self._require_normalizer(source_format)
             with _track_conversion_metrics(
                 "request_patch", str(source_format).upper(), str(target_format).upper()
@@ -159,11 +169,11 @@ class FormatConversionRegistry:
         output_limit: int | None = None,
     ) -> dict[str, Any]:
         """异步版本的 convert_request，在 internal 阶段执行图片 URL 下载等异步操作。"""
-        if str(source_format).upper() == str(target_format).upper() and not target_variant:
+        if self._same_normalizer(source_format, target_format) and not target_variant:
             return request
 
-        # 同格式 + variant: 优先尝试轻量补丁（跳过 internal 转换）
-        if str(source_format).upper() == str(target_format).upper() and target_variant:
+        # 同 normalizer + variant: 优先尝试轻量补丁（跳过 internal 转换）
+        if self._same_normalizer(source_format, target_format) and target_variant:
             normalizer = self._require_normalizer(source_format)
             with _track_conversion_metrics(
                 "request_patch", str(source_format).upper(), str(target_format).upper()
@@ -208,7 +218,7 @@ class FormatConversionRegistry:
                             如果提供，响应中的 model 字段将使用此值，
                             而不是上游返回的映射后模型名。
         """
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             # 即使格式相同，也需要替换 model 字段
             if requested_model and isinstance(response, dict):
                 response = dict(response)  # 避免修改原始响应
@@ -237,7 +247,7 @@ class FormatConversionRegistry:
         source_format: str,
         target_format: str,
     ) -> dict[str, Any]:
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             return error_response
 
         src = self._require_normalizer(source_format)
@@ -375,7 +385,7 @@ class FormatConversionRegistry:
         target_format: str,
         state: StreamState | None = None,
     ) -> list[dict[str, Any]]:
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             return [chunk]
 
         src = self._require_normalizer(source_format)
@@ -412,7 +422,7 @@ class FormatConversionRegistry:
     # ==================== 能力查询 ====================
 
     def can_convert_request(self, source_format: str, target_format: str) -> bool:
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             return True
         return (
             self.get_normalizer(source_format) is not None
@@ -423,7 +433,7 @@ class FormatConversionRegistry:
         return self.can_convert_request(source_format, target_format)
 
     def can_convert_stream(self, source_format: str, target_format: str) -> bool:
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             return True
         src = self.get_normalizer(source_format)
         tgt = self.get_normalizer(target_format)
@@ -432,7 +442,7 @@ class FormatConversionRegistry:
         return bool(src.capabilities.supports_stream and tgt.capabilities.supports_stream)
 
     def can_convert_error(self, source_format: str, target_format: str) -> bool:
-        if str(source_format).upper() == str(target_format).upper():
+        if self._same_normalizer(source_format, target_format):
             return True
         src = self.get_normalizer(source_format)
         tgt = self.get_normalizer(target_format)

src/services/task/service.py

@@ -1704,6 +1704,15 @@ class TaskService:
             delta = task.completed_at - task.submitted_at
             response_time_ms = int(delta.total_seconds() * 1000)
 
+        request_headers: dict[str, Any] | None = None
+        if isinstance(getattr(task, "request_metadata", None), dict):
+            task_meta = task.request_metadata
+            for header_key in ("request_headers", "headers", "original_headers"):
+                raw_headers = task_meta.get(header_key)
+                if isinstance(raw_headers, dict):
+                    request_headers = dict(raw_headers)
+                    break
+
         try:
             await UsageService.record_usage_with_custom_cost(
                 db=self.db,
@@ -1734,11 +1743,7 @@ class TaskService:
                     "fallback_created": True,
                     "video_task_id": task.id,
                 },
-                request_headers=(
-                    (task.request_metadata or {}).get("request_headers")
-                    if isinstance(task.request_metadata, dict)
-                    else None
-                ),
+                request_headers=request_headers,
                 request_body=getattr(task, "original_request_body", None),
                 provider_request_headers=None,
                 response_headers=None,

src/services/usage/_recording_helpers.py

@@ -108,12 +108,12 @@ def build_usage_params(
 
     # 处理请求头（可能需要脱敏）
     processed_request_headers = None
-    if should_log_headers and request_headers:
+    if should_log_headers and request_headers is not None:
         processed_request_headers = SystemConfigService.mask_sensitive_headers(db, request_headers)
 
     # 处理提供商请求头（可能需要脱敏）
     processed_provider_request_headers = None
-    if should_log_headers and provider_request_headers:
+    if should_log_headers and provider_request_headers is not None:
         processed_provider_request_headers = SystemConfigService.mask_sensitive_headers(
             db, provider_request_headers
         )
@@ -124,33 +124,33 @@ def build_usage_params(
     processed_response_body = None
     processed_client_response_body = None
     if should_log_body:
-        if request_body:
+        if request_body is not None:
             processed_request_body = SystemConfigService.truncate_body(
                 db, request_body, is_request=True
             )
-        if provider_request_body:
+        if provider_request_body is not None:
             processed_provider_request_body = SystemConfigService.truncate_body(
                 db, provider_request_body, is_request=True
             )
-        if response_body:
+        if response_body is not None:
             processed_response_body = SystemConfigService.truncate_body(
                 db, response_body, is_request=False
             )
-        if client_response_body:
+        if client_response_body is not None:
             processed_client_response_body = SystemConfigService.truncate_body(
                 db, client_response_body, is_request=False
             )
 
     # 处理响应头
     processed_response_headers = None
-    if should_log_headers and response_headers:
+    if should_log_headers and response_headers is not None:
         processed_response_headers = SystemConfigService.mask_sensitive_headers(
             db, response_headers
         )
 
     # 处理返回给客户端的响应头
     processed_client_response_headers = None
-    if should_log_headers and client_response_headers:
+    if should_log_headers and client_response_headers is not None:
         processed_client_response_headers = SystemConfigService.mask_sensitive_headers(
             db, client_response_headers
         )

src/services/usage/lifecycle.py

@@ -45,14 +45,14 @@ class UsageLifecycleMixin:
 
         # 处理请求头
         processed_request_headers = None
-        if should_log_headers and request_headers:
+        if should_log_headers and request_headers is not None:
             processed_request_headers = SystemConfigService.mask_sensitive_headers(
                 db, request_headers
             )
 
         # 处理请求体
         processed_request_body = None
-        if should_log_body and request_body:
+        if should_log_body and request_body is not None:
             processed_request_body = SystemConfigService.truncate_body(
                 db, request_body, is_request=True
             )
@@ -265,17 +265,17 @@ class UsageLifecycleMixin:
         should_log_body = SystemConfigService.should_log_body(db)
 
         processed_provider_headers = None
-        if should_log_headers and provider_request_headers:
+        if should_log_headers and provider_request_headers is not None:
             processed_provider_headers = SystemConfigService.mask_sensitive_headers(
                 db, provider_request_headers
             )
 
         processed_response_headers = None
-        if should_log_headers and response_headers:
+        if should_log_headers and response_headers is not None:
             processed_response_headers = dict(response_headers)
 
         processed_response_body = None
-        if should_log_body and response_body:
+        if should_log_body and response_body is not None:
             processed_response_body = SystemConfigService.truncate_body(
                 db, response_body, is_request=False
             )

src/services/usage/telemetry_writer.py

@@ -252,18 +252,14 @@ class QueueTelemetryWriter(TelemetryWriter):
 
         # Optional: Headers (masked)
         if self.include_headers:
-            if kwargs.get("request_headers"):
-                data["request_headers"] = self._mask_headers(kwargs["request_headers"])
-            if kwargs.get("provider_request_headers"):
-                data["provider_request_headers"] = self._mask_headers(
-                    kwargs["provider_request_headers"]
-                )
-            if kwargs.get("response_headers"):
-                data["response_headers"] = self._mask_headers(kwargs["response_headers"])
-            if kwargs.get("client_response_headers"):
-                data["client_response_headers"] = self._mask_headers(
-                    kwargs["client_response_headers"]
-                )
+            for _hdr_key in (
+                "request_headers",
+                "provider_request_headers",
+                "response_headers",
+                "client_response_headers",
+            ):
+                if kwargs.get(_hdr_key) is not None:
+                    data[_hdr_key] = self._mask_headers(kwargs[_hdr_key])
 
         # Optional: Bodies (truncated)
         if self.include_bodies:

tests/services/test_usage_queue_events.py

@@ -338,6 +338,37 @@ async def test_queue_writer_include_headers_bodies(monkeypatch: Any) -> None:
     assert "response_body" in event.data
 
 
+@pytest.mark.asyncio
+async def test_queue_writer_failure_preserves_empty_request_headers(monkeypatch: Any) -> None:
+    """失败事件在传入空请求头时也应保留 request_headers 字段。"""
+    dummy = DummyRedis()
+
+    async def _get_redis_client(require_redis: bool = False) -> Any:
+        return dummy
+
+    monkeypatch.setattr("src.services.usage.telemetry_writer.get_redis_client", _get_redis_client)
+
+    writer = QueueTelemetryWriter(
+        request_id="req-empty-hdr",
+        user_id="user-1",
+        api_key_id="key-1",
+        log_level="full",
+    )
+    await writer.record_failure(
+        provider="test",
+        model="model",
+        status_code=500,
+        error_message="boom",
+        request_headers={},
+        request_body={"message": "x"},
+    )
+
+    event = UsageEvent.from_stream_fields(dummy.calls[0][1])
+    assert "request_headers" in event.data
+    assert event.data["request_headers"] == {}
+    assert event.data["request_body"] == {"message": "x"}
+
+
 @pytest.mark.asyncio
 async def test_event_to_record_body_deserialization(monkeypatch: Any) -> None:
     """测试 _event_to_record 正确反序列化 body 字符串为 dict"""

tests/services/test_video_finalize.py

@@ -189,3 +189,40 @@ async def test_video_finalize_strict_mode_missing_required_marks_failed(
     kwargs = update_settled.call_args.kwargs
     assert kwargs["total_cost_usd"] == 0.0
     assert kwargs["status"] == "failed"
+
+
+@pytest.mark.asyncio
+async def test_video_finalize_fallback_reads_headers_alias(monkeypatch: pytest.MonkeyPatch) -> None:
+    usage_query = MagicMock()
+    usage_query.filter.return_value.first.return_value = None
+    user_query = MagicMock()
+    user_query.filter.return_value.first.return_value = SimpleNamespace(id="u1")
+    api_key_query = MagicMock()
+    api_key_query.filter.return_value.first.return_value = SimpleNamespace(id="ak1")
+    provider_query = MagicMock()
+    provider_query.filter.return_value.first.return_value = SimpleNamespace(
+        id="p1",
+        name="prov1",
+    )
+
+    db = MagicMock()
+    db.query.side_effect = [usage_query, user_query, api_key_query, provider_query]
+
+    task = _make_task(
+        status=VideoStatus.FAILED.value,
+        error_message="boom",
+        request_metadata={"headers": {"x-test-header": "1"}},
+    )
+
+    record_usage = AsyncMock(return_value=SimpleNamespace(id="usage-1"))
+    monkeypatch.setattr(
+        "src.services.usage.service.UsageService.record_usage_with_custom_cost",
+        record_usage,
+    )
+
+    svc = TaskService(db)
+    ok = await svc.finalize_video_task(task)
+
+    assert ok is True
+    kwargs = record_usage.call_args.kwargs
+    assert kwargs["request_headers"] == {"x-test-header": "1"}