feat: OAuth Token 自动刷新调度、OAuth 对话框优化及 OpenAI CLI normalizer 重构

- 系统设置新增 OAuth Token 自动刷新任务开关，支持动态调度
- OAuth 授权对话框简化步骤布局，移除编号圆圈样式
- 重构 OpenAI CLI normalizer：将 if-else 链替换为事件处理器映射表，
  将 stream_event_from_internal 拆分为独立方法
- 维护调度器在禁用刷新时移除已调度的 job
- .gitignore 新增 CLIProxyAPI/ 和 sub2api/ 排除项
- 补充 noop/unknown 事件处理器的测试覆盖

.gitignore

@@ -1,6 +1,9 @@
 # Created by https://www.toptal.com/developers/gitignore/api/python
 # Edit at https://www.toptal.com/developers/gitignore?templates=python
 
+CLIProxyAPI/
+sub2api/
+
 # AI Assistant Configuration
 .claude/
 .serena/

frontend/src/features/providers/components/OAuthAccountDialog.vue

@@ -6,7 +6,7 @@
     size="md"
     @update:model-value="handleDialogUpdate"
   >
-    <div class="space-y-6">
+    <div class="space-y-5">
       <!-- 加载中 -->
       <div
         v-if="oauth.starting && !oauth.authorization_url"
@@ -21,22 +21,14 @@
       <!-- 授权流程 -->
       <template v-else-if="oauth.authorization_url">
         <!-- 步骤 1: 打开授权链接 -->
-        <div class="space-y-3">
+        <div class="space-y-2">
-          <div class="flex items-center gap-2">
+          <p class="text-xs font-medium text-muted-foreground uppercase tracking-wider">
-            <div class="w-5 h-5 rounded-full bg-primary text-primary-foreground flex items-center justify-center text-xs font-medium shrink-0">
+            第一步 · 前往授权
-              1
+          </p>
-            </div>
+          <p class="text-xs text-muted-foreground">
-            <span class="text-sm font-medium">打开授权链接</span>
-          </div>
-          <p class="text-xs text-muted-foreground pl-7">
             点击下方按钮在浏览器中完成登录授权
           </p>
-          <div class="ml-7 p-2.5 rounded-md bg-muted/50 border border-border/50">
+          <div class="flex gap-2 pt-1">
-            <p class="text-xs font-mono text-muted-foreground break-all line-clamp-3 leading-relaxed">
-              {{ oauth.authorization_url }}
-            </p>
-          </div>
-          <div class="flex gap-2 pl-7">
             <Button
               size="sm"
               :disabled="oauthBusy"
@@ -57,28 +49,26 @@
           </div>
         </div>
 
+        <Separator />
+
         <!-- 步骤 2: 粘贴回调地址 -->
-        <div class="space-y-3">
+        <div class="space-y-2">
-          <div class="flex items-center gap-2">
+          <p class="text-xs font-medium text-muted-foreground uppercase tracking-wider">
-            <div class="w-5 h-5 rounded-full bg-muted text-muted-foreground flex items-center justify-center text-xs font-medium shrink-0">
+            第二步 · 粘贴回调
-              2
+          </p>
-            </div>
+          <p class="text-xs text-muted-foreground">
-            <span class="text-sm font-medium">粘贴回调地址</span>
-          </div>
-          <p class="text-xs text-muted-foreground pl-7">
             授权完成后，复制浏览器地址栏的完整 URL 并粘贴到下方
           </p>
-          <div class="pl-7">
+          <div class="pt-1">
             <Textarea
               v-model="oauth.callback_url"
               :disabled="oauthBusy"
               placeholder="http://localhost:xxx/callback?code=..."
-              class="min-h-[80px] text-xs font-mono resize-none"
+              class="min-h-[80px] text-xs font-mono break-all !rounded-xl"
               spellcheck="false"
             />
           </div>
         </div>
-
       </template>
     </div>
 
@@ -93,7 +83,7 @@
         :disabled="!canCompleteOAuth"
         @click="handleCompleteOAuth"
       >
-        {{ oauth.completing ? '验证中...' : '完成授权' }}
+        {{ oauth.completing ? '验证中...' : '验证' }}
       </Button>
     </template>
   </Dialog>
@@ -101,7 +91,7 @@
 
 <script setup lang="ts">
 import { ref, computed, watch } from 'vue'
-import { Dialog, Button, Textarea } from '@/components/ui'
+import { Dialog, Button, Textarea, Separator } from '@/components/ui'
 import { UserPlus, Copy, ExternalLink } from 'lucide-vue-next'
 import { useToast } from '@/composables/useToast'
 import { useClipboard } from '@/composables/useClipboard'

frontend/src/views/admin/SystemSettings.vue

@@ -534,7 +534,7 @@
 
                 <!-- 右侧：时间选择器 + 保存按钮 -->
                 <div
-                  v-if="task.enabled"
+                  v-if="task.enabled && task.hasTimeConfig"
                   class="flex items-center gap-2 shrink-0"
                 >
                   <Clock class="w-4 h-4 text-muted-foreground" />
@@ -995,7 +995,7 @@
 
 <script setup lang="ts">
 import { ref, computed, onMounted } from 'vue'
-import { Download, Upload, CalendarCheck, RotateCcw, Clock, Check, Loader2 } from 'lucide-vue-next'
+import { Download, Upload, CalendarCheck, RotateCcw, RefreshCw, Clock, Check, Loader2 } from 'lucide-vue-next'
 import Button from '@/components/ui/button.vue'
 import Input from '@/components/ui/input.vue'
 import Label from '@/components/ui/label.vue'
@@ -1044,6 +1044,7 @@ interface SystemConfig {
   enable_user_quota_reset: boolean
   user_quota_reset_time: string
   user_quota_reset_interval_days: number
+  enable_oauth_token_refresh: boolean
 }
 
 const basicConfigLoading = ref(false)
@@ -1104,6 +1105,7 @@ const systemConfig = ref<SystemConfig>({
   enable_user_quota_reset: false,
   user_quota_reset_time: '05:00',
   user_quota_reset_interval_days: 1,
+  enable_oauth_token_refresh: true,
 })
 
 // 原始配置值（用于检测变动）
@@ -1215,6 +1217,7 @@ async function loadSystemConfig() {
       'enable_user_quota_reset',
       'user_quota_reset_time',
       'user_quota_reset_interval_days',
+      'enable_oauth_token_refresh',
     ]
 
     for (const key of configs) {
@@ -1423,6 +1426,28 @@ async function handleUserQuotaResetToggle(enabled: boolean) {
   }
 }
 
+async function handleOAuthTokenRefreshToggle(enabled: boolean) {
+  const previousValue = systemConfig.value.enable_oauth_token_refresh
+  systemConfig.value.enable_oauth_token_refresh = enabled
+  try {
+    await adminApi.updateSystemConfig(
+      'enable_oauth_token_refresh',
+      enabled,
+      '是否启用 OAuth Token 自动刷新任务'
+    )
+    success(enabled ? '已启用 OAuth Token 自动刷新' : '已禁用 OAuth Token 自动刷新')
+  } catch (err) {
+    error('保存配置失败')
+    log.error('保存 OAuth Token 自动刷新配置失败:', err)
+    // 回滚状态
+    systemConfig.value.enable_oauth_token_refresh = previousValue
+  }
+}
+
+// OAuth Token 刷新（无时间配置，占位 ref）
+const oauthRefreshHourSelectOpen = ref(false)
+const oauthRefreshMinuteSelectOpen = ref(false)
+
 // 用户配额重置时间相关
 const previousUserQuotaResetTime = ref('')
 const userQuotaResetHourSelectOpen = ref(false)
@@ -1465,6 +1490,7 @@ const scheduledTasks = computed(() => [
     title: 'Provider 自动签到',
     description: '自动执行已配置 Provider 的签到任务',
     enabled: systemConfig.value.enable_provider_checkin,
+    hasTimeConfig: true,
     hour: checkinHour.value,
     minute: checkinMinute.value,
     hourSelectOpen: checkinHourSelectOpen,
@@ -1481,6 +1507,7 @@ const scheduledTasks = computed(() => [
     title: '用户配额自动重置',
     description: '定时将用户已使用配额重置为零',
     enabled: systemConfig.value.enable_user_quota_reset,
+    hasTimeConfig: true,
     hour: userQuotaResetHour.value,
     minute: userQuotaResetMinute.value,
     hourSelectOpen: userQuotaResetHourSelectOpen,
@@ -1491,6 +1518,23 @@ const scheduledTasks = computed(() => [
     onToggle: handleUserQuotaResetToggle,
     onSave: handleQuotaResetConfigSave,
   },
+  {
+    id: 'oauth-token-refresh',
+    icon: RefreshCw,
+    title: 'OAuth Token 自动刷新',
+    description: '主动刷新即将过期的 OAuth Token（动态调度）',
+    enabled: systemConfig.value.enable_oauth_token_refresh,
+    hasTimeConfig: false,
+    hour: '',
+    minute: '',
+    hourSelectOpen: oauthRefreshHourSelectOpen,
+    minuteSelectOpen: oauthRefreshMinuteSelectOpen,
+    updateTime: () => {},
+    hasChanges: false,
+    loading: false,
+    onToggle: handleOAuthTokenRefreshToggle,
+    onSave: () => {},
+  },
 ])
 
 // Provider 签到时间保存

src/api/admin/system.py

@@ -634,6 +634,15 @@ class AdminSetSystemConfigAdapter(AdminApiAdapter):
             except Exception as e:
                 logger.warning(f"更新用户配额重置任务时间失败: {e}")
 
+        # 如果更新的是 OAuth Token 自动刷新开关，触发调度器重新计算
+        if self.key == "enable_oauth_token_refresh":
+            try:
+                from src.services.system.maintenance_scheduler import get_maintenance_scheduler
+
+                get_maintenance_scheduler().trigger_oauth_refresh_check()
+            except Exception as e:
+                logger.warning("触发 OAuth Token 刷新调度失败: {}", e)
+
         # 返回时不暴露加密后的值
         display_value = "********" if self.key in self.ENCRYPTED_KEYS else config.value
 

src/core/api_format/conversion/normalizers/openai_cli.py

@@ -12,6 +12,7 @@ OpenAI CLI / Responses Normalizer (OPENAI_CLI)
 
 import json
 import time
+from collections.abc import Callable
 from typing import Any
 
 from src.core.api_format.conversion.field_mappings import (
@@ -318,12 +319,26 @@ class OpenAICliNormalizer(FormatNormalizer):
                 ss.setdefault("text_block_stopped", False)
                 events.append(MessageStartEvent(message_id=msg_id, model=model))
 
-        # response.created：响应创建事件，message_start 已在上面处理
+        handler = self._CHUNK_HANDLERS.get(etype)
-        if etype == "response.created":
+        if handler is not None:
+            events.extend(handler(self, chunk, state, ss))
             return events
 
-        # 文本增量：response.output_text.delta
+        # 未匹配的事件类型
-        if etype in ("response.output_text.delta", "response.outtext.delta"):
+        if etype:
+            return events + [UnknownStreamEvent(raw_type=etype, payload=chunk)]
+        return events + [UnknownStreamEvent(raw_type="unknown", payload=chunk)]
+
+    def _handle_response_created(
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        # message_start 已在主方法中处理
+        return []
+
+    def _handle_output_text_delta(
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         delta = chunk.get("delta")
         delta_text = ""
         if isinstance(delta, str):
@@ -334,21 +349,23 @@ class OpenAICliNormalizer(FormatNormalizer):
         if delta_text:
             if not ss.get("text_block_started"):
                 ss["text_block_started"] = True
-                    events.append(
+                events.append(ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT))
-                        ContentBlockStartEvent(block_index=0, block_type=ContentType.TEXT)
-                    )
             events.append(ContentDeltaEvent(block_index=0, text_delta=delta_text))
         return events
 
-        # 文本完成：response.output_text.done（可选）
+    def _handle_output_text_done(
-        if etype == "response.output_text.done":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         if ss.get("text_block_started") and not ss.get("text_block_stopped"):
             ss["text_block_stopped"] = True
             events.append(ContentBlockStopEvent(block_index=0))
         return events
 
-        # 完成：response.completed（包含 usage）
+    def _handle_response_completed(
-        if etype == "response.completed":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         resp_obj = chunk.get("response")
         resp_obj = resp_obj if isinstance(resp_obj, dict) else {}
         usage = self._usage_to_internal(resp_obj.get("usage") or chunk.get("usage"))
@@ -360,16 +377,19 @@ class OpenAICliNormalizer(FormatNormalizer):
         events.append(MessageStopEvent(stop_reason=StopReason.END_TURN, usage=usage))
         return events
 
-        # 失败：response.failed（最佳努力）
+    def _handle_response_failed(
-        if etype == "response.failed":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         try:
             events.append(ErrorEvent(error=self.error_to_internal(chunk)))
         except Exception:
             pass
         return events
 
-        # response.in_progress：状态更新，不产生内容事件
+    def _handle_response_in_progress(
-        if etype == "response.in_progress":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
         # 更新 state 中的元数据（如果有）
         # 注意：model 保持初始值（客户端请求的模型），不被上游覆盖
         resp_obj = chunk.get("response")
@@ -381,12 +401,13 @@ class OpenAICliNormalizer(FormatNormalizer):
                 state.model = str(resp_obj.get("model"))
         return []
 
-        # response.output_item.added：新输出项添加（如 message、function_call 等）
+    def _handle_output_item_added(
-        if etype == "response.output_item.added":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         item = chunk.get("item")
         if isinstance(item, dict):
             item_type = item.get("type")
-                # function_call 输出项
             if item_type == "function_call":
                 if not ss.get("tool_block_started"):
                     ss["tool_block_started"] = True
@@ -403,14 +424,12 @@ class OpenAICliNormalizer(FormatNormalizer):
                         )
                     )
                     ss["block_index"] = ss.get("block_index", 0) + 1
-                # message 输出项
-                elif item_type == "message":
-                    # 通常在 response.created 时已处理，这里可以忽略或更新状态
-                    pass
         return events
 
-        # response.output_item.done：输出项完成
+    def _handle_output_item_done(
-        if etype == "response.output_item.done":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         item = chunk.get("item")
         if isinstance(item, dict):
             item_type = item.get("type")
@@ -419,8 +438,10 @@ class OpenAICliNormalizer(FormatNormalizer):
                 events.append(ContentBlockStopEvent(block_index=ss.get("block_index", 1) - 1))
         return events
 
-        # response.function_call_arguments.delta：工具调用参数增量
+    def _handle_function_call_delta(
-        if etype == "response.function_call_arguments.delta":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
+    ) -> list[InternalStreamEvent]:
+        events: list[InternalStreamEvent] = []
         delta = chunk.get("delta") or ""
         if delta:
             events.append(
@@ -432,28 +453,35 @@ class OpenAICliNormalizer(FormatNormalizer):
             )
         return events
 
-        # response.function_call_arguments.done：工具调用参数完成
+    def _handle_noop(
-        if etype == "response.function_call_arguments.done":
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
-            # 参数已完整，不产生额外事件
+    ) -> list[InternalStreamEvent]:
         return []
 
-        # response.content_part.added / response.content_part.done：内容部分事件
+    def _handle_as_unknown(
-        if etype in ("response.content_part.added", "response.content_part.done"):
+        self, chunk: dict[str, Any], state: StreamState, ss: dict[str, Any]
-            # 通常伴随 output_text 事件，这里可以忽略
+    ) -> list[InternalStreamEvent]:
-            return []
+        etype = str(chunk.get("type") or "unknown")
-
-        # response.reasoning_summary_text.delta：推理摘要增量
-        if etype == "response.reasoning_summary_text.delta":
-            # 保留为 UnknownStreamEvent，让下游决定是否使用
         return [UnknownStreamEvent(raw_type=etype, payload=chunk)]
 
-        # response.reasoning_summary_text.done：推理摘要完成
+    # 事件类型 -> 处理器映射表
-        if etype == "response.reasoning_summary_text.done":
+    _CHUNK_HANDLERS: dict[str, Callable[..., list[InternalStreamEvent]]] = {
-            return [UnknownStreamEvent(raw_type=etype, payload=chunk)]
+        "response.created": _handle_response_created,
-
+        "response.output_text.delta": _handle_output_text_delta,
-        if etype:
+        "response.outtext.delta": _handle_output_text_delta,
-            return [UnknownStreamEvent(raw_type=etype, payload=chunk)]
+        "response.output_text.done": _handle_output_text_done,
-        return [UnknownStreamEvent(raw_type="unknown", payload=chunk)]
+        "response.completed": _handle_response_completed,
+        "response.failed": _handle_response_failed,
+        "response.in_progress": _handle_response_in_progress,
+        "response.output_item.added": _handle_output_item_added,
+        "response.output_item.done": _handle_output_item_done,
+        "response.function_call_arguments.delta": _handle_function_call_delta,
+        "response.function_call_arguments.done": _handle_noop,
+        "response.content_part.added": _handle_noop,
+        "response.content_part.done": _handle_noop,
+        "response.reasoning_summary_text.delta": _handle_as_unknown,
+        "response.reasoning_summary_text.done": _handle_as_unknown,
+    }
 
     def stream_event_from_internal(
         self,
@@ -461,13 +489,28 @@ class OpenAICliNormalizer(FormatNormalizer):
         state: StreamState,
     ) -> list[dict[str, Any]]:
         ss = state.substate(self.FORMAT_ID)
-        out: list[dict[str, Any]] = []
-
-        def event_block(payload: dict[str, Any]) -> dict[str, Any]:
-            # OpenAI Responses SSE 的 payload 通常自带 type 字段；这里强制保证
-            return payload
 
         if isinstance(event, MessageStartEvent):
+            return self._emit_message_start(event, state, ss)
+        if isinstance(event, ContentBlockStartEvent):
+            return self._emit_content_block_start(event, state, ss)
+        if isinstance(event, ToolCallDeltaEvent):
+            return self._emit_tool_call_delta(event, state, ss)
+        if isinstance(event, ContentBlockStopEvent):
+            return self._emit_content_block_stop(event, state, ss)
+        if isinstance(event, ContentDeltaEvent):
+            return self._emit_content_delta(event, state, ss)
+        if isinstance(event, MessageStopEvent):
+            return self._emit_message_stop(event, state, ss)
+        if isinstance(event, ErrorEvent):
+            return self._emit_error(event, state, ss)
+        # 其他事件：Responses SSE 无直接对应，跳过
+        return []
+
+    def _emit_message_start(
+        self, event: MessageStartEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         state.message_id = event.message_id or state.message_id or "resp_stream"
         # 保留初始化时设置的 model（客户端请求的模型），仅在空时用事件值
         if not state.model:
@@ -490,21 +533,17 @@ class OpenAICliNormalizer(FormatNormalizer):
             "status": "in_progress",
             "output": [],
         }
-            out.append(
+        out.append({"type": "response.created", "response": response_obj})
-                event_block(
-                    {
-                        "type": "response.created",
-                        "response": response_obj,
-                    }
-                )
-            )
         # OpenAI Responses API 常见的 in_progress 事件（可选，最佳努力）
         if not ss.get("sent_in_progress"):
             ss["sent_in_progress"] = True
-                out.append(event_block({"type": "response.in_progress", "response": response_obj}))
+            out.append({"type": "response.in_progress", "response": response_obj})
         return out
 
-        if isinstance(event, ContentBlockStartEvent):
+    def _emit_content_block_start(
+        self, event: ContentBlockStartEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         # 工具调用块：输出 function_call 添加事件
         if event.block_type == ContentType.TOOL_USE:
             tool_id = event.tool_id or ""
@@ -515,13 +554,10 @@ class OpenAICliNormalizer(FormatNormalizer):
                 tool_calls = ss.setdefault("tool_calls", {})
                 tool_calls.setdefault(tool_id, {"name": tool_name, "args": ""})
                 output_order = ss.setdefault("output_order", [])
-                    output_order.append(
+                output_order.append({"kind": "tool", "id": tool_id, "output_index": output_index})
-                        {"kind": "tool", "id": tool_id, "output_index": output_index}
-                    )
                 ss.setdefault("tool_blocks", {})[event.block_index] = tool_id
                 ss.setdefault("tool_output_index", {})[tool_id] = output_index
             out.append(
-                    event_block(
                 {
                     "type": "response.output_item.added",
                     "output_index": output_index,
@@ -535,10 +571,12 @@ class OpenAICliNormalizer(FormatNormalizer):
                     },
                 }
             )
-                )
         return out
 
-        if isinstance(event, ToolCallDeltaEvent):
+    def _emit_tool_call_delta(
+        self, event: ToolCallDeltaEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         tool_id = event.tool_id or ss.get("tool_blocks", {}).get(event.block_index, "")
         if tool_id:
             tool_calls = ss.setdefault("tool_calls", {})
@@ -546,7 +584,6 @@ class OpenAICliNormalizer(FormatNormalizer):
             entry["args"] = str(entry.get("args") or "") + (event.input_delta or "")
             output_index = ss.get("tool_output_index", {}).get(tool_id, event.block_index)
             out.append(
-                    event_block(
                 {
                     "type": "response.function_call_arguments.delta",
                     "delta": event.input_delta,
@@ -554,10 +591,12 @@ class OpenAICliNormalizer(FormatNormalizer):
                     "output_index": output_index,
                 }
             )
-                )
         return out
 
-        if isinstance(event, ContentBlockStopEvent):
+    def _emit_content_block_stop(
+        self, event: ContentBlockStopEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         tool_blocks = ss.get("tool_blocks", {})
         tool_id = (
             tool_blocks.pop(event.block_index, None) if isinstance(tool_blocks, dict) else None
@@ -567,7 +606,6 @@ class OpenAICliNormalizer(FormatNormalizer):
             entry = tool_calls.get(tool_id, {})
             output_index = ss.get("tool_output_index", {}).get(tool_id, event.block_index)
             out.append(
-                    event_block(
                 {
                     "type": "response.output_item.done",
                     "output_index": output_index,
@@ -581,10 +619,12 @@ class OpenAICliNormalizer(FormatNormalizer):
                     },
                 }
             )
-                )
         return out
 
-        if isinstance(event, ContentDeltaEvent):
+    def _emit_content_delta(
+        self, event: ContentDeltaEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         if event.text_delta:
             if not ss.get("message_output_started"):
                 output_index = int(ss.get("next_output_index") or 0)
@@ -596,7 +636,6 @@ class OpenAICliNormalizer(FormatNormalizer):
                     {"kind": "message", "id": message_id, "output_index": output_index}
                 )
                 out.append(
-                        event_block(
                     {
                         "type": "response.output_item.added",
                         "output_index": output_index,
@@ -609,20 +648,15 @@ class OpenAICliNormalizer(FormatNormalizer):
                         },
                     }
                 )
-                    )
             ss["text_started"] = True
             ss["collected_text"] = str(ss.get("collected_text") or "") + event.text_delta
-                out.append(
+            out.append({"type": "response.output_text.delta", "delta": event.text_delta})
-                    event_block(
-                        {
-                            "type": "response.output_text.delta",
-                            "delta": event.text_delta,
-                        }
-                    )
-                )
         return out
 
-        if isinstance(event, MessageStopEvent):
+    def _emit_message_stop(
+        self, event: MessageStopEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
+        out: list[dict[str, Any]] = []
         final_text = str(ss.get("collected_text") or "")
         message_id = f"msg_{state.message_id or 'stream'}"
         message_item = {
@@ -633,25 +667,16 @@ class OpenAICliNormalizer(FormatNormalizer):
             "content": ([{"type": "output_text", "text": final_text}] if final_text else []),
         }
         if ss.get("text_started"):
-                out.append(
+            out.append({"type": "response.output_text.done", "text": final_text})
-                    event_block(
-                        {
-                            "type": "response.output_text.done",
-                            "text": final_text,
-                        }
-                    )
-                )
         if ss.get("message_output_started"):
             output_index = ss.get("message_output_index") or 0
             out.append(
-                    event_block(
                 {
                     "type": "response.output_item.done",
                     "output_index": output_index,
                     "item": message_item,
                 }
             )
-                )
         response_obj = self.response_from_internal(
             InternalResponse(
                 id=state.message_id or "resp",
@@ -662,6 +687,15 @@ class OpenAICliNormalizer(FormatNormalizer):
             )
         )
         # 将工具调用添加到 output（最佳努力）
+        output_items = self._build_final_output_items(message_item, ss)
+        if output_items:
+            response_obj["output"] = output_items
+        out.append({"type": "response.completed", "response": response_obj})
+        return out
+
+    def _build_final_output_items(
+        self, message_item: dict[str, Any], ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
         tool_calls = ss.get("tool_calls", {})
         output_order = ss.get("output_order", [])
         output_items: list[dict[str, Any]] = []
@@ -678,47 +712,33 @@ class OpenAICliNormalizer(FormatNormalizer):
                     if not isinstance(tool_id, str) or not tool_id:
                         continue
                     used_tool_ids.add(tool_id)
-                        tool_entry = (
+                    tool_entry = tool_calls.get(tool_id) if isinstance(tool_calls, dict) else None
-                            tool_calls.get(tool_id) if isinstance(tool_calls, dict) else None
-                        )
                     if isinstance(tool_entry, dict):
-                            output_items.append(
+                        output_items.append(self._tool_call_item(tool_id, tool_entry))
-                                {
-                                    "type": "function_call",
-                                    "call_id": tool_id,
-                                    "id": tool_id,
-                                    "name": tool_entry.get("name") or "",
-                                    "arguments": tool_entry.get("args") or "",
-                                    "status": "completed",
-                                }
-                            )
         if isinstance(tool_calls, dict):
             for tool_id, tool_entry in tool_calls.items():
                 if tool_id in used_tool_ids or not isinstance(tool_entry, dict):
                     continue
-                    output_items.append(
+                output_items.append(self._tool_call_item(tool_id, tool_entry))
-                        {
+        return output_items
+
+    @staticmethod
+    def _tool_call_item(tool_id: str, entry: dict[str, Any]) -> dict[str, Any]:
+        return {
             "type": "function_call",
             "call_id": tool_id,
             "id": tool_id,
-                            "name": tool_entry.get("name") or "",
+            "name": entry.get("name") or "",
-                            "arguments": tool_entry.get("args") or "",
+            "arguments": entry.get("args") or "",
             "status": "completed",
         }
-                    )
-            if output_items:
-                response_obj["output"] = output_items
-            out.append(event_block({"type": "response.completed", "response": response_obj}))
-            return out
 
-        if isinstance(event, ErrorEvent):
+    def _emit_error(
+        self, event: ErrorEvent, state: StreamState, ss: dict[str, Any]
+    ) -> list[dict[str, Any]]:
         err_payload = self.error_from_internal(event.error)
         err_payload["type"] = "response.failed"
-            out.append(event_block(err_payload))
+        return [err_payload]
-            return out
-
-        # 其他事件：Responses SSE 无直接对应，跳过
-        return out
 
     # =========================
     # Error conversion
@@ -851,24 +871,40 @@ class OpenAICliNormalizer(FormatNormalizer):
         for item in input_data:
             if not isinstance(item, dict):
                 continue
+            msg = self._parse_input_item(item)
+            if msg is not None:
+                messages.append(msg)
+        return messages
 
+    def _parse_input_item(self, item: dict[str, Any]) -> InternalMessage | None:
         item_type = str(item.get("type") or "")
 
         # 标准 message（有 role 字段）
         if item_type == "message" or item.get("role"):
+            return self._parse_message_item(item)
+        if item_type == "function_call":
+            return self._parse_function_call_item(item)
+        if item_type == "function_call_output":
+            return self._parse_function_call_output_item(item)
+        if item_type == "reasoning":
+            return self._parse_reasoning_item(item)
+
+        # 其他未知类型 -> 保留为 UnknownBlock
+        return InternalMessage(
+            role=Role.UNKNOWN,
+            content=[UnknownBlock(raw_type=item_type or "unknown", payload=item)],
+        )
+
+    def _parse_message_item(self, item: dict[str, Any]) -> InternalMessage:
         role = self._role_from_value(item.get("role"))
         blocks = self._responses_content_to_blocks(item.get("content"))
-                messages.append(
+        return InternalMessage(
-                    InternalMessage(
             role=role,
             content=blocks,
             extra=self._extract_extra(item, {"type", "role", "content"}),
         )
-                )
-                continue
 
-            # function_call -> assistant 消息 + ToolUseBlock
+    def _parse_function_call_item(self, item: dict[str, Any]) -> InternalMessage:
-            if item_type == "function_call":
         tool_id = str(item.get("call_id") or item.get("id") or "")
         tool_name = str(item.get("name") or "")
         args_raw = item.get("arguments") or "{}"
@@ -890,28 +926,21 @@ class OpenAICliNormalizer(FormatNormalizer):
                 )
             },
         )
-                messages.append(InternalMessage(role=Role.ASSISTANT, content=[tool_block]))
+        return InternalMessage(role=Role.ASSISTANT, content=[tool_block])
-                continue
 
-            # function_call_output -> tool 消息 + ToolResultBlock
+    def _parse_function_call_output_item(self, item: dict[str, Any]) -> InternalMessage:
-            if item_type == "function_call_output":
         tool_use_id = str(item.get("call_id") or item.get("id") or "")
         output = item.get("output")
-                # output 可能是字符串或结构化数据
         content_text = output if isinstance(output, str) else None
         result_block = ToolResultBlock(
             tool_use_id=tool_use_id,
             output=output,
             content_text=content_text,
-                    extra={
+            extra={"openai_cli": self._extract_extra(item, {"type", "call_id", "id", "output"})},
-                        "openai_cli": self._extract_extra(item, {"type", "call_id", "id", "output"})
-                    },
         )
-                messages.append(InternalMessage(role=Role.TOOL, content=[result_block]))
+        return InternalMessage(role=Role.TOOL, content=[result_block])
-                continue
 
-            # reasoning -> assistant 消息，提取 summary 作为文本
+    def _parse_reasoning_item(self, item: dict[str, Any]) -> InternalMessage:
-            if item_type == "reasoning":
         summary_parts: list[str] = []
         summary = item.get("summary")
         if isinstance(summary, list):
@@ -925,10 +954,8 @@ class OpenAICliNormalizer(FormatNormalizer):
         elif isinstance(summary, str) and summary:
             summary_parts.append(summary)
 
-                # 如果有 summary 文本，创建一个 UnknownBlock 保留原始结构
         reasoning_blocks: list[ContentBlock] = []
         if summary_parts:
-                    # 保留 reasoning 的 summary 作为 UnknownBlock，便于输出时决策
             reasoning_blocks.append(
                 UnknownBlock(
                     raw_type="reasoning",
@@ -937,24 +964,11 @@ class OpenAICliNormalizer(FormatNormalizer):
             )
         else:
             reasoning_blocks.append(UnknownBlock(raw_type="reasoning", payload=item))
-                messages.append(
+        return InternalMessage(
-                    InternalMessage(
             role=Role.ASSISTANT,
             content=reasoning_blocks,
             extra={"openai_cli": {"type": "reasoning"}},
         )
-                )
-                continue
-
-            # 其他未知类型 -> 保留为 UnknownBlock
-            messages.append(
-                InternalMessage(
-                    role=Role.UNKNOWN,
-                    content=[UnknownBlock(raw_type=item_type or "unknown", payload=item)],
-                )
-            )
-
-        return messages
 
     def _responses_content_to_blocks(self, content: Any) -> list[ContentBlock]:
         if content is None:

src/services/system/maintenance_scheduler.py

@@ -396,6 +396,11 @@ class MaintenanceScheduler:
                 # 检查配置开关
                 if not SystemConfigService.get_config(db, "enable_oauth_token_refresh", True):
                     logger.info("OAuth Token 自动刷新已禁用，不调度任务")
+                    # 移除已调度的 job（如果存在）
+                    try:
+                        scheduler.remove_job(job_id)
+                    except Exception:
+                        pass
                     return
                 # 查找所有活跃的 OAuth 类型 Key
                 oauth_keys = (
@@ -966,9 +971,7 @@ class MaintenanceScheduler:
                     days_since_reset = (now_local.date() - last_local_date).days
 
                     if days_since_reset < 0:
-                        logger.warning(
+                        logger.warning("user_quota_last_reset_at 在未来，跳过本次用户配额自动重置")
-                            "user_quota_last_reset_at 在未来，跳过本次用户配额自动重置"
-                        )
                         should_run = False
                     elif days_since_reset < interval_days:
                         logger.info(
@@ -1006,7 +1009,9 @@ class MaintenanceScheduler:
                 "用户配额自动重置的上次执行时间（UTC，内部使用）",
             )
 
-            logger.info(f"用户配额自动重置完成: interval_days={interval_days}, 重置用户数={reset_count}")
+            logger.info(
+                f"用户配额自动重置完成: interval_days={interval_days}, 重置用户数={reset_count}"
+            )
 
         except Exception as e:
             logger.exception(f"用户配额自动重置任务执行失败: {e}")

tests/core/api_format/conversion/test_cli_conversion.py

@@ -17,6 +17,7 @@ from src.core.api_format.conversion.normalizers.gemini_cli import GeminiCliNorma
 from src.core.api_format.conversion.normalizers.openai import OpenAINormalizer
 from src.core.api_format.conversion.normalizers.openai_cli import OpenAICliNormalizer
 from src.core.api_format.conversion.registry import FormatConversionRegistry
+from src.core.api_format.conversion.stream_events import UnknownStreamEvent
 from src.core.api_format.conversion.stream_state import StreamState
 
 
@@ -309,6 +310,36 @@ def test_stream_openai_cli_in_progress_event() -> None:
     assert events2 == []
 
 
+def test_stream_openai_cli_noop_and_unknown_events() -> None:
+    """覆盖 OpenAI CLI noop/unknown 事件处理器"""
+    normalizer = OpenAICliNormalizer()
+    state = StreamState()
+
+    # 先触发 message_start 初始化
+    normalizer.stream_chunk_to_internal(
+        {"type": "response.created", "response": {"id": "resp_789", "model": "gpt-5"}}, state
+    )
+
+    # noop 事件不应产生内部事件
+    assert (
+        normalizer.stream_chunk_to_internal({"type": "response.function_call_arguments.done"}, state)
+        == []
+    )
+    assert (
+        normalizer.stream_chunk_to_internal({"type": "response.content_part.added"}, state) == []
+    )
+    assert (
+        normalizer.stream_chunk_to_internal({"type": "response.content_part.done"}, state) == []
+    )
+
+    # unknown 事件应返回 UnknownStreamEvent
+    events = normalizer.stream_chunk_to_internal(
+        {"type": "response.reasoning_summary_text.delta"}, state
+    )
+    assert len(events) == 1
+    assert isinstance(events[0], UnknownStreamEvent)
+
+
 def test_stream_openai_cli_function_call_events() -> None:
     """测试 OpenAI CLI 流式 function_call 相关事件"""
     reg = _make_registry_with_cli()