From 1d925b748812ea56a2f13943c8044c0c9064a059 Mon Sep 17 00:00:00 2001 From: hu <187184415@qq.com> Date: Thu, 3 Sep 2026 17:45:05 +0800 Subject: [PATCH] feat(claude): first-token timeout, pre-first-token SSE keepalive and latency logs for the Claude OAuth path Production Claude requests occasionally receive message_start and then nothing for 4-9 minutes (mostly effort xhigh with ~150k context). With the global first_token_timeout_seconds at 0 these attempts held concurrency slots until the client gave up (45 stuck 499s / 215 slot-minutes in 24h), and downstream gateways saw zero bytes, so they timed out and retried, multiplying the load on the account. - ClaudeConfig gains first_token_timeout_seconds (default 120, 0 = follow global, clamped to 600) and stream_keepalive_enabled (default true); Store publishes both, admin GET/PUT round-trips them, Settings UI adds a paired row under the ClaudeCode card. - The /v1/messages attempt loop uses the Claude timeout for Claude OAuth accounts; a native attempt that times out before any visible frame is classified as a first-token timeout outcome (retryable, penalized) instead of a generic stream break. - Claude OAuth streaming attempts activate the existing SSE keepalive so a ": keepalive" comment is written every 15s while waiting for the first visible frame. - Log first-token timeouts, pre-first-token client disconnects and slow (>=60s) first tokens with account/model/effort/wait. --- admin/claude_config.go | 17 ++++ admin/claude_config_test.go | 54 ++++++++++ auth/claude_fingerprint_mode.go | 87 ++++++++++++++++ auth/claude_first_token_timeout_test.go | 81 +++++++++++++++ auth/store.go | 3 + frontend/src/locales/en.json | 4 + frontend/src/locales/zh-TW.json | 4 + frontend/src/locales/zh.json | 4 + frontend/src/pages/Settings.tsx | 34 ++++++- frontend/src/types.ts | 2 + proxy/claude_first_token_timeout.go | 73 ++++++++++++++ proxy/claude_first_token_timeout_test.go | 120 +++++++++++++++++++++++ proxy/handler_anthropic.go | 11 ++- 13 files changed, 490 insertions(+), 4 deletions(-) create mode 100644 auth/claude_first_token_timeout_test.go create mode 100644 proxy/claude_first_token_timeout.go create mode 100644 proxy/claude_first_token_timeout_test.go diff --git a/admin/claude_config.go b/admin/claude_config.go index 9cb0ff05..8f61bede 100644 --- a/admin/claude_config.go +++ b/admin/claude_config.go @@ -22,6 +22,10 @@ type claudeGlobalConfigDTO struct { auth.ClaudeSecurityConfig CLIVersionSyncEnabled *bool `json:"cli_version_sync_enabled"` CLIVersionSyncIntervalHours int `json:"cli_version_sync_interval_hours"` + // FirstTokenTimeoutSeconds：Claude 路径首字超时秒数（缺失=默认 120，0=跟随全局）。 + FirstTokenTimeoutSeconds *int `json:"first_token_timeout_seconds"` + // StreamKeepaliveEnabled：Claude 流式首字前是否发 SSE 保活注释（缺失=开启）。 + StreamKeepaliveEnabled *bool `json:"stream_keepalive_enabled"` // 以下三项只读；PUT 忽略。 SyncedCLIVersion string `json:"synced_cli_version"` BuiltinCLIVersion string `json:"builtin_cli_version"` @@ -39,6 +43,8 @@ func (h *Handler) GetClaudeConfig(c *gin.Context) { ClaudeSecurityConfig: security, CLIVersionSyncEnabled: boolPtr(h.store.ClaudeCLIVersionSyncEnabled()), CLIVersionSyncIntervalHours: h.store.ClaudeCLIVersionSyncIntervalHours(), + FirstTokenTimeoutSeconds: claudeIntPtr(h.store.ClaudeFirstTokenTimeoutSeconds()), + StreamKeepaliveEnabled: boolPtr(h.store.ClaudeStreamKeepaliveEnabled()), SyncedCLIVersion: auth.ClaudeSyncedCLIVersion(), BuiltinCLIVersion: auth.BuiltinClaudeCLIVersion, EffectiveCLIVersion: auth.EffectiveClaudeCLIVersion(), @@ -80,6 +86,8 @@ func (h *Handler) UpdateClaudeConfig(c *gin.Context) { security := auth.NormalizeClaudeSecurityConfig(req.ClaudeSecurityConfig) syncEnabled := req.CLIVersionSyncEnabled == nil || *req.CLIVersionSyncEnabled syncInterval := auth.NormalizeClaudeCLIVersionSyncIntervalHours(req.CLIVersionSyncIntervalHours) + firstTokenTimeout := auth.NormalizeClaudeFirstTokenTimeoutSeconds(req.FirstTokenTimeoutSeconds) + streamKeepalive := req.StreamKeepaliveEnabled == nil || *req.StreamKeepaliveEnabled cfg := auth.ClaudeConfig{ FingerprintMode: mode, @@ -89,6 +97,8 @@ func (h *Handler) UpdateClaudeConfig(c *gin.Context) { ClaudeSecurityConfig: security, CLIVersionSyncEnabled: boolPtr(syncEnabled), CLIVersionSyncIntervalHours: syncInterval, + FirstTokenTimeoutSeconds: claudeIntPtr(firstTokenTimeout), + StreamKeepaliveEnabled: boolPtr(streamKeepalive), } raw, err := json.Marshal(cfg) if err != nil { @@ -107,6 +117,8 @@ func (h *Handler) UpdateClaudeConfig(c *gin.Context) { h.store.SetClaudeClientPolicy(clientPolicy) h.store.SetClaudeSecurityConfig(security) h.store.SetClaudeCLIVersionSync(syncEnabled, syncInterval) + h.store.SetClaudeFirstTokenTimeoutSeconds(firstTokenTimeout) + h.store.SetClaudeStreamKeepaliveEnabled(streamKeepalive) c.JSON(http.StatusOK, gin.H{ "message": "已保存 ClaudeCode 全局配置", @@ -126,12 +138,17 @@ func (h *Handler) UpdateClaudeConfig(c *gin.Context) { "max_tool_schema_bytes": security.MaxToolSchemaBytes, "cli_version_sync_enabled": syncEnabled, "cli_version_sync_interval_hours": syncInterval, + "first_token_timeout_seconds": firstTokenTimeout, + "stream_keepalive_enabled": streamKeepalive, }) } // boolPtr 返回指向给定 bool 值的指针，便于构造「显式布尔字段」的 JSON DTO。 func boolPtr(v bool) *bool { return &v } +// claudeIntPtr 返回指向给定 int 值的指针，用于「缺失与显式 0 有别」的 JSON DTO 字段。 +func claudeIntPtr(v int) *int { return &v } + // claudeCLIVersionSyncResponse 在同步结果之上附加一个可选的 warning 字段： // 抓取+持久化成功、但指纹回写部分失败时，仍以 200 响应并携带 warning， // 而不是把整次同步判为失败。 diff --git a/admin/claude_config_test.go b/admin/claude_config_test.go index 0846c14c..93219122 100644 --- a/admin/claude_config_test.go +++ b/admin/claude_config_test.go @@ -220,3 +220,57 @@ func TestClaudeConfigSyncCLIVersion_FetchFailureReturns502(t *testing.T) { t.Fatalf("status = %d, want 502: %s", recorder.Code, recorder.Body.String()) } } + +func TestClaudeConfigFirstTokenTimeoutAndKeepaliveRoundTrip(t *testing.T) { + db := newTestAdminDB(t) + store := auth.NewStore(db, nil, nil) + defer store.Stop() + h := &Handler{store: store, db: db} + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + h.GetClaudeConfig(c) + if got := gjson.GetBytes(recorder.Body.Bytes(), "first_token_timeout_seconds").Int(); got != int64(auth.DefaultClaudeFirstTokenTimeoutSeconds) { + t.Fatalf("default first_token_timeout_seconds = %d, want %d", got, auth.DefaultClaudeFirstTokenTimeoutSeconds) + } + if got := gjson.GetBytes(recorder.Body.Bytes(), "stream_keepalive_enabled"); !got.Exists() || !got.Bool() { + t.Fatalf("stream_keepalive_enabled must default to true, got %s", got.Raw) + } + + recorder = httptest.NewRecorder() + c, _ = gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest("PUT", "/api/admin/settings/claude-config", strings.NewReader(`{"first_token_timeout_seconds":90,"stream_keepalive_enabled":false}`)) + h.UpdateClaudeConfig(c) + if recorder.Code != 200 { + t.Fatalf("status = %d body=%s", recorder.Code, recorder.Body.String()) + } + if got := gjson.GetBytes(recorder.Body.Bytes(), "first_token_timeout_seconds").Int(); got != 90 { + t.Fatalf("response first_token_timeout_seconds = %d, want 90", got) + } + if store.ClaudeFirstTokenTimeoutSeconds() != 90 || store.ClaudeStreamKeepaliveEnabled() { + t.Fatalf("runtime store not updated: timeout=%d keepalive=%v", store.ClaudeFirstTokenTimeoutSeconds(), store.ClaudeStreamKeepaliveEnabled()) + } + settings, err := db.GetSystemSettings(context.Background()) + if err != nil { + t.Fatal(err) + } + raw := settings.ClaudeConfig + persisted := auth.ParseClaudeConfig(raw) + if persisted.FirstTokenTimeoutSecondsValue() != 90 || persisted.StreamKeepaliveEnabledValue() { + t.Fatalf("persisted config = %s", raw) + } + + // Explicit 0 must persist as 0 (follow global), not be re-defaulted to 120. + recorder = httptest.NewRecorder() + c, _ = gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest("PUT", "/api/admin/settings/claude-config", strings.NewReader(`{"first_token_timeout_seconds":0}`)) + h.UpdateClaudeConfig(c) + if store.ClaudeFirstTokenTimeoutSeconds() != 0 { + t.Fatalf("explicit 0 must disable the Claude timeout, got %d", store.ClaudeFirstTokenTimeoutSeconds()) + } + settings, _ = db.GetSystemSettings(context.Background()) + raw = settings.ClaudeConfig + if auth.ParseClaudeConfig(raw).FirstTokenTimeoutSecondsValue() != 0 { + t.Fatalf("persisted explicit 0 was re-defaulted: %s", raw) + } +} diff --git a/auth/claude_fingerprint_mode.go b/auth/claude_fingerprint_mode.go index 376080eb..ac57ed81 100644 --- a/auth/claude_fingerprint_mode.go +++ b/auth/claude_fingerprint_mode.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "sync/atomic" + "time" ) // Claude Code 出站请求的指纹收敛模式(账号级;空值 = 跟随全局默认): @@ -190,6 +191,83 @@ func (c ClaudeConfig) CLIVersionSyncEnabledValue() bool { return c.CLIVersionSyncEnabled == nil || *c.CLIVersionSyncEnabled } +// DefaultClaudeFirstTokenTimeoutSeconds 是 Claude OAuth 路径首字超时的默认值： +// 长推理（effort xhigh、~150k 上下文）正常也会在 1~2 分钟内吐出首个 thinking delta， +// 超过这个时间基本是上游卡死，继续等只会让并发位被僵尸请求占住。 +const DefaultClaudeFirstTokenTimeoutSeconds = 120 + +// MaxClaudeFirstTokenTimeoutSeconds 与全局 first_token_timeout_seconds 的上限保持一致。 +const MaxClaudeFirstTokenTimeoutSeconds = 600 + +// NormalizeClaudeFirstTokenTimeoutSeconds 把配置值钳到 [0,600]；nil（老配置缺失）取默认 120， +// 负数视为 0（跟随全局）。 +func NormalizeClaudeFirstTokenTimeoutSeconds(seconds *int) int { + if seconds == nil { + return DefaultClaudeFirstTokenTimeoutSeconds + } + if *seconds <= 0 { + return 0 + } + if *seconds > MaxClaudeFirstTokenTimeoutSeconds { + return MaxClaudeFirstTokenTimeoutSeconds + } + return *seconds +} + +// FirstTokenTimeoutSecondsValue 返回归一化后的 Claude 首字超时秒数（0=跟随全局）。 +func (c ClaudeConfig) FirstTokenTimeoutSecondsValue() int { + return NormalizeClaudeFirstTokenTimeoutSeconds(c.FirstTokenTimeoutSeconds) +} + +// StreamKeepaliveEnabledValue 把缺失字段解释为开启。 +func (c ClaudeConfig) StreamKeepaliveEnabledValue() bool { + return c.StreamKeepaliveEnabled == nil || *c.StreamKeepaliveEnabled +} + +// SetClaudeFirstTokenTimeoutSeconds 发布 Claude 路径首字超时（0=跟随全局）。 +func (s *Store) SetClaudeFirstTokenTimeoutSeconds(seconds int) { + if s == nil { + return + } + if seconds < 0 { + seconds = 0 + } + if seconds > MaxClaudeFirstTokenTimeoutSeconds { + seconds = MaxClaudeFirstTokenTimeoutSeconds + } + s.claudeFirstTokenTimeoutSec.Store(int64(seconds)) + s.claudeFirstTokenTimeoutSet.Store(true) +} + +// ClaudeFirstTokenTimeoutSeconds 返回 Claude 路径首字超时秒数；从未设置时取默认 120。 +func (s *Store) ClaudeFirstTokenTimeoutSeconds() int { + if s == nil { + return DefaultClaudeFirstTokenTimeoutSeconds + } + if !s.claudeFirstTokenTimeoutSet.Load() { + return DefaultClaudeFirstTokenTimeoutSeconds + } + return int(s.claudeFirstTokenTimeoutSec.Load()) +} + +// ClaudeFirstTokenTimeout 返回 Claude 路径首字超时时长；0 表示跟随全局设置。 +func (s *Store) ClaudeFirstTokenTimeout() time.Duration { + return time.Duration(s.ClaudeFirstTokenTimeoutSeconds()) * time.Second +} + +// SetClaudeStreamKeepaliveEnabled 发布 Claude 流式首字前 SSE 保活开关。 +func (s *Store) SetClaudeStreamKeepaliveEnabled(enabled bool) { + if s == nil { + return + } + s.claudeStreamKeepaliveDisabled.Store(!enabled) +} + +// ClaudeStreamKeepaliveEnabled 报告 Claude 流式首字前 SSE 保活是否开启（零值=开启）。 +func (s *Store) ClaudeStreamKeepaliveEnabled() bool { + return s != nil && !s.claudeStreamKeepaliveDisabled.Load() +} + // NormalizeClaudeCLIVersionSyncIntervalHours 钳到 [1,720]，0/负数视为默认 12。 func NormalizeClaudeCLIVersionSyncIntervalHours(hours int) int { if hours <= 0 { @@ -248,6 +326,11 @@ type ClaudeConfig struct { SessionWindowLimit int64 `json:"session_window_limit"` // 默认并发会话窗口数(0=跟随全局 maxConcurrency) CLIVersionSyncEnabled *bool `json:"cli_version_sync_enabled,omitempty"` // 缺失=true CLIVersionSyncIntervalHours int `json:"cli_version_sync_interval_hours,omitempty"` // 0=12，钳 [1,720] + // FirstTokenTimeoutSeconds 是 Claude OAuth 路径专用的首字超时（秒）。缺失=默认 + // DefaultClaudeFirstTokenTimeoutSeconds；显式 0=跟随全局 first_token_timeout_seconds。 + FirstTokenTimeoutSeconds *int `json:"first_token_timeout_seconds,omitempty"` + // StreamKeepaliveEnabled 控制 Claude 流式请求在首字前是否向下游发 SSE 保活注释。缺失=开启。 + StreamKeepaliveEnabled *bool `json:"stream_keepalive_enabled,omitempty"` ClaudeClientPolicy ClaudeSecurityConfig } @@ -354,6 +437,8 @@ func ParseClaudeConfig(raw string) ClaudeConfig { cfg.SessionWindowLimit = 0 } cfg.CLIVersionSyncIntervalHours = NormalizeClaudeCLIVersionSyncIntervalHours(cfg.CLIVersionSyncIntervalHours) + normalizedTimeout := NormalizeClaudeFirstTokenTimeoutSeconds(cfg.FirstTokenTimeoutSeconds) + cfg.FirstTokenTimeoutSeconds = &normalizedTimeout if clientPolicy, err := NormalizeClaudeClientPolicy(cfg.ClaudeClientPolicy); err == nil { cfg.ClaudeClientPolicy = clientPolicy } else { @@ -370,6 +455,8 @@ func applyClaudeConfigToStore(s *Store, raw string) { s.SetClaudeDefaultTimezone(cfg.DefaultTimezone) s.SetClaudeSessionWindowLimit(cfg.SessionWindowLimit) s.SetClaudeCLIVersionSync(cfg.CLIVersionSyncEnabledValue(), cfg.CLIVersionSyncIntervalHours) + s.SetClaudeFirstTokenTimeoutSeconds(cfg.FirstTokenTimeoutSecondsValue()) + s.SetClaudeStreamKeepaliveEnabled(cfg.StreamKeepaliveEnabledValue()) s.SetClaudeClientPolicy(cfg.ClaudeClientPolicy) s.SetClaudeSecurityConfig(cfg.SecurityConfig()) } diff --git a/auth/claude_first_token_timeout_test.go b/auth/claude_first_token_timeout_test.go new file mode 100644 index 00000000..ce852bb8 --- /dev/null +++ b/auth/claude_first_token_timeout_test.go @@ -0,0 +1,81 @@ +package auth + +import ( + "testing" + "time" +) + +func TestParseClaudeConfig_FirstTokenTimeoutDefaultsWhenMissing(t *testing.T) { + cfg := ParseClaudeConfig(`{"fingerprint_mode":"preserve"}`) + if cfg.FirstTokenTimeoutSecondsValue() != DefaultClaudeFirstTokenTimeoutSeconds { + t.Fatalf("missing field must default to %d, got %d", DefaultClaudeFirstTokenTimeoutSeconds, cfg.FirstTokenTimeoutSecondsValue()) + } + if !cfg.StreamKeepaliveEnabledValue() { + t.Fatal("missing stream_keepalive_enabled must default to true") + } +} + +func TestParseClaudeConfig_FirstTokenTimeoutExplicitZeroFollowsGlobal(t *testing.T) { + cfg := ParseClaudeConfig(`{"first_token_timeout_seconds":0,"stream_keepalive_enabled":false}`) + if cfg.FirstTokenTimeoutSecondsValue() != 0 { + t.Fatalf("explicit 0 must stay 0 (follow global), got %d", cfg.FirstTokenTimeoutSecondsValue()) + } + if cfg.StreamKeepaliveEnabledValue() { + t.Fatal("explicit false must stay false") + } +} + +func TestNormalizeClaudeFirstTokenTimeoutSeconds(t *testing.T) { + cases := map[string]struct { + in *int + want int + }{ + "nil": {nil, DefaultClaudeFirstTokenTimeoutSeconds}, + "negative": {intPtrForTest(-5), 0}, + "zero": {intPtrForTest(0), 0}, + "normal": {intPtrForTest(90), 90}, + "too big": {intPtrForTest(99999), MaxClaudeFirstTokenTimeoutSeconds}, + } + for name, tc := range cases { + if got := NormalizeClaudeFirstTokenTimeoutSeconds(tc.in); got != tc.want { + t.Fatalf("%s: got %d, want %d", name, got, tc.want) + } + } +} + +func TestStoreClaudeFirstTokenTimeoutRoundTrip(t *testing.T) { + store := NewStore(nil, nil, nil) + defer store.Stop() + if got := store.ClaudeFirstTokenTimeout(); got != time.Duration(DefaultClaudeFirstTokenTimeoutSeconds)*time.Second { + t.Fatalf("fresh store must use the default, got %s", got) + } + store.SetClaudeFirstTokenTimeoutSeconds(45) + if got := store.ClaudeFirstTokenTimeout(); got != 45*time.Second { + t.Fatalf("got %s, want 45s", got) + } + store.SetClaudeFirstTokenTimeoutSeconds(0) + if got := store.ClaudeFirstTokenTimeout(); got != 0 { + t.Fatalf("0 must disable the Claude-specific timeout, got %s", got) + } + if !store.ClaudeStreamKeepaliveEnabled() { + t.Fatal("fresh store must enable pre-first-token keepalive") + } + store.SetClaudeStreamKeepaliveEnabled(false) + if store.ClaudeStreamKeepaliveEnabled() { + t.Fatal("keepalive switch must persist false") + } +} + +func TestApplyClaudeConfigToStore_FirstTokenTimeout(t *testing.T) { + store := NewStore(nil, nil, nil) + defer store.Stop() + applyClaudeConfigToStore(store, `{"first_token_timeout_seconds":75,"stream_keepalive_enabled":false}`) + if got := store.ClaudeFirstTokenTimeout(); got != 75*time.Second { + t.Fatalf("got %s, want 75s", got) + } + if store.ClaudeStreamKeepaliveEnabled() { + t.Fatal("stream keepalive must be applied from config") + } +} + +func intPtrForTest(v int) *int { return &v } diff --git a/auth/store.go b/auth/store.go index 4c49737e..dbcdb4f9 100644 --- a/auth/store.go +++ b/auth/store.go @@ -3337,6 +3337,9 @@ type Store struct { claudeSessionWindowLimit int64 // Claude 账号默认并发会话窗口数（0=用全局 maxConcurrency） claudeCLIVersionSyncDisabled atomic.Bool // Claude CLI 版本自动同步是否关闭（零值=开启） claudeCLIVersionSyncIntervalH atomic.Int64 // Claude CLI 版本同步间隔小时（0=默认 12） + claudeFirstTokenTimeoutSec atomic.Int64 // Claude 路径首字超时秒（0=跟随全局） + claudeFirstTokenTimeoutSet atomic.Bool // 首字超时是否被显式设置过（否则取默认 120） + claudeStreamKeepaliveDisabled atomic.Bool // Claude 流式首字前 SSE 保活是否关闭（零值=开启） grokAffinityMode atomic.Value // string: "follow" / "bounded" / "off" / "strict"（"follow"=跟随全局） grokProbeEnabled atomic.Bool // 定期探测 Grok 账号状态是否开启（默认关） grokProbeIntervalMin atomic.Int64 // 定期探测间隔（分钟，默认 30，下限 grokProbeMinIntervalMinutes） diff --git a/frontend/src/locales/en.json b/frontend/src/locales/en.json index 1121669d..494d6d62 100644 --- a/frontend/src/locales/en.json +++ b/frontend/src/locales/en.json @@ -4341,6 +4341,10 @@ "claudeCliVersionAutoSyncDesc": "When on, syncs on the configured interval; when off, only the built-in version is applied at startup.", "claudeCliVersionSyncInterval": "Sync interval (hours)", "claudeCliVersionSyncIntervalDesc": "Wait time between automatic syncs (hours, range 1-720).", + "claudeFirstTokenTimeout": "First-token timeout (s)", + "claudeFirstTokenTimeoutDesc": "Claude OAuth only: if upstream produces no visible content (text/thinking delta) within this many seconds, the attempt is cancelled, its slot released, and the request retried once on another account. Default 120; 0 = follow the global first-token timeout.", + "claudeStreamKeepalive": "SSE keepalive before first token", + "claudeStreamKeepaliveDesc": "While a streaming request waits for its first content, write an SSE comment line downstream every 15s so gateways/clients can tell \"upstream is thinking\" from \"connection is dead\" and stop retrying on timeout. Note: once a keepalive is written the HTTP status is committed as 200; later failures arrive as SSE error events.", "claudeCliVersionBuiltin": "built-in", "claudeSessionWindow": "Session window (concurrency)", "claudeSessionWindowDesc": "Default max concurrency for Claude accounts; empty or 0 = follow global.", diff --git a/frontend/src/locales/zh-TW.json b/frontend/src/locales/zh-TW.json index 59fada7a..09d036b2 100644 --- a/frontend/src/locales/zh-TW.json +++ b/frontend/src/locales/zh-TW.json @@ -199,6 +199,10 @@ "claudeCliVersionAutoSyncDesc": "開啟後按間隔自動同步；關閉後僅在啟動時用內建版本回寫指紋。", "claudeCliVersionSyncInterval": "同步間隔（小時）", "claudeCliVersionSyncIntervalDesc": "兩次自動同步之間的等待時長（小時，範圍 1-720）。", + "claudeFirstTokenTimeout": "首字逾時（秒）", + "claudeFirstTokenTimeoutDesc": "Claude OAuth 路徑專用：上游在此秒數內未吐出首個可見內容（文字/思考增量）即取消該次請求、釋放並發位並換號重試一次。預設 120，0 = 跟隨全域「首字逾時」設定。", + "claudeStreamKeepalive": "首字前 SSE 保活", + "claudeStreamKeepaliveDesc": "串流請求在首個內容到達前每 15 秒向下游寫一行 SSE 註解，讓下游閘道/客戶端能區分「上游在思考」與「連線已斷」，避免其逾時重試放大並發佔用。注意：保活寫出後 HTTP 狀態已提交為 200，之後的失敗會以 SSE error 事件回傳。", "claudeCliVersionBuiltin": "內建", "claudeSessionWindow": "並發會話視窗數", "claudeSessionWindowDesc": "Claude 帳號預設最大並發;留空或 0=跟隨全域並發。", diff --git a/frontend/src/locales/zh.json b/frontend/src/locales/zh.json index b6e65f9a..a21c1f52 100644 --- a/frontend/src/locales/zh.json +++ b/frontend/src/locales/zh.json @@ -4341,6 +4341,10 @@ "claudeCliVersionAutoSyncDesc": "开启后按间隔自动同步；关闭后仅在启动时用内置版本回写指纹。", "claudeCliVersionSyncInterval": "同步间隔（小时）", "claudeCliVersionSyncIntervalDesc": "两次自动同步之间的等待时长（小时，范围 1-720）。", + "claudeFirstTokenTimeout": "首字超时（秒）", + "claudeFirstTokenTimeoutDesc": "Claude OAuth 路径专用：上游在此秒数内未吐出首个可见内容（文本/思考增量）即取消该次请求、释放并发位并换号重试一次。默认 120，0 = 跟随全局「首字超时」设置。", + "claudeStreamKeepalive": "首字前 SSE 保活", + "claudeStreamKeepaliveDesc": "流式请求在首个内容到达前每 15 秒向下游写一行 SSE 注释，让下游网关/客户端能区分「上游在思考」与「连接已断」，避免其超时重试放大并发占用。注意：保活写出后 HTTP 状态已提交为 200，之后的失败会以 SSE error 事件返回。", "claudeCliVersionBuiltin": "内置", "claudeSessionWindow": "并发会话窗口数", "claudeSessionWindowDesc": "Claude 账号默认最大并发;留空或 0=跟随全局并发。", diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 4e611bcb..1ca915e2 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -720,6 +720,8 @@ function ClaudeCodeSettingsCard() { const [maxToolSchemaBytes, setMaxToolSchemaBytes] = useState('0') const [cliVersionSyncEnabled, setCliVersionSyncEnabled] = useState(true) const [cliVersionSyncIntervalHours, setCliVersionSyncIntervalHours] = useState(12) + const [firstTokenTimeoutSeconds, setFirstTokenTimeoutSeconds] = useState(120) + const [streamKeepaliveEnabled, setStreamKeepaliveEnabled] = useState(true) const [syncedCliVersion, setSyncedCliVersion] = useState('') const [effectiveCliVersion, setEffectiveCliVersion] = useState('') const [syncingCliVersion, setSyncingCliVersion] = useState(false) @@ -749,6 +751,8 @@ function ClaudeCodeSettingsCard() { setMaxToolSchemaBytes(String(cfg.max_tool_schema_bytes ?? 0)) setCliVersionSyncEnabled(cfg.cli_version_sync_enabled ?? true) setCliVersionSyncIntervalHours(cfg.cli_version_sync_interval_hours || 12) + setFirstTokenTimeoutSeconds(cfg.first_token_timeout_seconds ?? 120) + setStreamKeepaliveEnabled(cfg.stream_keepalive_enabled ?? true) setSyncedCliVersion(cfg.synced_cli_version ?? '') setEffectiveCliVersion(cfg.effective_cli_version ?? cfg.builtin_cli_version ?? '') }) @@ -787,6 +791,8 @@ function ClaudeCodeSettingsCard() { max_tool_schema_bytes: Number.isFinite(maxToolSchemaValue) && maxToolSchemaValue >= 0 ? Math.floor(maxToolSchemaValue) : 0, cli_version_sync_enabled: cliVersionSyncEnabled, cli_version_sync_interval_hours: cliVersionSyncIntervalHours, + first_token_timeout_seconds: firstTokenTimeoutSeconds, + stream_keepalive_enabled: streamKeepaliveEnabled, }) showToast(t('settings.claudeSaved'), 'success') } catch (error) { @@ -794,7 +800,7 @@ function ClaudeCodeSettingsCard() { } finally { setSaving(false) } - }, [allowInferenceGeo, allowSafetyIdentifier, allowServiceTier, allowSpeed, allowedBetaHeaders, cliVersionSyncEnabled, cliVersionSyncIntervalHours, clientPlatform, clientVersion, fingerprintMode, maxOutputTokens, maxToolCount, maxToolSchemaBytes, sessionWindow, showToast, t, timezone, versionPolicy]) + }, [allowInferenceGeo, allowSafetyIdentifier, allowServiceTier, allowSpeed, allowedBetaHeaders, cliVersionSyncEnabled, cliVersionSyncIntervalHours, clientPlatform, clientVersion, fingerprintMode, firstTokenTimeoutSeconds, maxOutputTokens, maxToolCount, maxToolSchemaBytes, sessionWindow, showToast, streamKeepaliveEnabled, t, timezone, versionPolicy]) const handleSyncClaudeCliVersion = useCallback(async () => { setSyncingCliVersion(true) @@ -934,6 +940,32 @@ function ClaudeCodeSettingsCard() { + {/* 首字超时 + 首字前保活成对横排：两者都只作用于 Claude OAuth 路径 */} +
+
+
+ {t('settings.claudeFirstTokenTimeout')} + +
+
+ + s +
+
+
+
+ {t('settings.claudeStreamKeepalive')} + +
+ +
+
diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 3b1ff036..7c5efb38 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -3676,6 +3676,8 @@ export interface ClaudeGlobalConfig { session_window_limit: number cli_version_sync_enabled: boolean cli_version_sync_interval_hours: number + first_token_timeout_seconds: number + stream_keepalive_enabled: boolean synced_cli_version?: string builtin_cli_version?: string effective_cli_version?: string diff --git a/proxy/claude_first_token_timeout.go b/proxy/claude_first_token_timeout.go new file mode 100644 index 00000000..65f5d49f --- /dev/null +++ b/proxy/claude_first_token_timeout.go @@ -0,0 +1,73 @@ +package proxy + +import ( + "context" + "log" + "net/http" + "time" + + "github.com/codex2api/auth" +) + +// claudeSlowFirstTokenLogThreshold 是 Claude 路径"首字缓慢"日志的阈值。生产观测： +// effort xhigh + 大上下文的正常请求首字多在 60s 内，超过即值得留痕定位。 +const claudeSlowFirstTokenLogThreshold = 60 * time.Second + +// claudeFirstTokenTimeoutFor 返回本次 attempt 应使用的首字超时。Claude OAuth 账号优先用 +// ClaudeCode 全局配置里的专属超时（默认 120s），配置为 0 或非 Claude 账号时跟随全局 +// first_token_timeout_seconds。全局值在生产常年为 0（关闭），而 Claude 上游偶发 +// message_start 之后数分钟无内容，不设超时会让并发位被僵尸请求长期占住。 +func claudeFirstTokenTimeoutFor(store *auth.Store, account *auth.Account) time.Duration { + if store != nil && account.IsClaudeOAuth() { + if timeout := store.ClaudeFirstTokenTimeout(); timeout > 0 { + return timeout + } + } + return currentFirstTokenTimeout() +} + +// claudeNativeFirstTokenOutcome 把"首字看门狗触发、且首个可见帧从未到达"的原生透传结果 +// 归一成首字超时 outcome：日志与重试判定沿用 Codex 翻译路径的同一语义，而不是笼统的 +// "上游流中断"。成功流与已有可见帧的流保持原结果。 +func claudeNativeFirstTokenOutcome(guard *firstTokenTimeoutGuard, firstTokenMs int, outcome streamOutcome, timeout time.Duration) streamOutcome { + if guard == nil || !guard.TimedOut() || firstTokenMs > 0 || outcome.logStatusCode == http.StatusOK { + return outcome + } + return firstTokenTimeoutOutcome(timeout) +} + +// activateClaudeStreamKeepalive 让 Claude OAuth 流式请求在首字前就开始向下游发 SSE 保活 +// 注释（间隔沿用 continuousRetryKeepaliveInterval）。原生透传会把 message_start 等 +// 首字前帧扣住等待静默重试窗口，下游在长推理期间收不到任何字节，网关/客户端会误判 +// 连接已死而超时重试，放大并发占用；保活让"上游在思考"与"连接已死"可区分。 +func activateClaudeStreamKeepalive(ctx context.Context, store *auth.Store, account *auth.Account, isStream bool) { + if !isStream || store == nil || !account.IsClaudeOAuth() || !store.ClaudeStreamKeepaliveEnabled() { + return + } + activateContinuousRetryKeepalive(ctx) +} + +// claudeFirstTokenSlow 报告已记录的首字耗时是否超过缓慢阈值；0 表示未记录到首字。 +func claudeFirstTokenSlow(firstTokenMs int) bool { + return firstTokenMs > 0 && time.Duration(firstTokenMs)*time.Millisecond >= claudeSlowFirstTokenLogThreshold +} + +// logClaudeFirstTokenLatency 给 Claude 路径的三类首字异常留痕：看门狗超时、首字缓慢、 +// 首字前下游已断开。每条都带账号/模型/effort/等待时长，便于按 effort 分档定位卡顿。 +func logClaudeFirstTokenLatency(account *auth.Account, model, effort string, firstTokenMs int, outcome streamOutcome, start time.Time) { + if !account.IsClaudeOAuth() { + return + } + if effort == "" { + effort = "-" + } + waited := time.Since(start).Round(time.Millisecond) + switch { + case outcome.failureKind == "timeout" && firstTokenMs == 0: + log.Printf("Claude 首字超时，已取消上游并释放并发位 (account=%d, model=%s, effort=%s, waited=%s, /v1/messages)", account.ID(), model, effort, waited) + case outcome.logStatusCode == logStatusClientClosed && firstTokenMs == 0: + log.Printf("Claude 首字前下游断开 (account=%d, model=%s, effort=%s, waited=%s, /v1/messages)", account.ID(), model, effort, waited) + case claudeFirstTokenSlow(firstTokenMs): + log.Printf("Claude 首字缓慢 (account=%d, model=%s, effort=%s, ttft_ms=%d, /v1/messages)", account.ID(), model, effort, firstTokenMs) + } +} diff --git a/proxy/claude_first_token_timeout_test.go b/proxy/claude_first_token_timeout_test.go new file mode 100644 index 00000000..5311abd0 --- /dev/null +++ b/proxy/claude_first_token_timeout_test.go @@ -0,0 +1,120 @@ +package proxy + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/codex2api/auth" + "github.com/gin-gonic/gin" +) + +func TestClaudeFirstTokenTimeoutFor_PrefersClaudeSettingForClaudeAccounts(t *testing.T) { + prev := CurrentRuntimeSettings() + ApplyRuntimeSettings(NormalizeRuntimeSettings(RuntimeSettings{FirstTokenTimeoutSec: 30})) + t.Cleanup(func() { ApplyRuntimeSettings(prev) }) + + store := auth.NewStore(nil, nil, nil) + defer store.Stop() + store.SetClaudeFirstTokenTimeoutSeconds(45) + claude := &auth.Account{DBID: 251, UpstreamType: auth.UpstreamClaude} + codex := &auth.Account{DBID: 1} + + if got := claudeFirstTokenTimeoutFor(store, claude); got != 45*time.Second { + t.Fatalf("claude account must use the Claude setting: %s", got) + } + if got := claudeFirstTokenTimeoutFor(store, codex); got != 30*time.Second { + t.Fatalf("non-claude account must keep the global timeout: %s", got) + } + store.SetClaudeFirstTokenTimeoutSeconds(0) + if got := claudeFirstTokenTimeoutFor(store, claude); got != 30*time.Second { + t.Fatalf("Claude setting 0 must fall back to global: %s", got) + } + if got := claudeFirstTokenTimeoutFor(nil, claude); got != 30*time.Second { + t.Fatalf("nil store must fall back to global: %s", got) + } +} + +func TestClaudeNativeFirstTokenOutcome_MapsGuardTimeoutToFirstTokenTimeout(t *testing.T) { + _, cancel := context.WithCancel(context.Background()) + defer cancel() + guard := newFirstTokenTimeoutGuard(5*time.Millisecond, cancel) + time.Sleep(30 * time.Millisecond) + if !guard.TimedOut() { + t.Fatal("guard must have fired") + } + broken := streamOutcome{logStatusCode: logStatusUpstreamStreamBreak, failureMessage: "上游流中断"} + got := claudeNativeFirstTokenOutcome(guard, 0, broken, 5*time.Millisecond) + if got.failureKind != "timeout" || got.logStatusCode != logStatusUpstreamStreamBreak || !got.penalize { + t.Fatalf("timed-out attempt without a visible token must become a first-token timeout outcome: %+v", got) + } + // A visible token arrived before the guard fired: keep the real outcome. + if got := claudeNativeFirstTokenOutcome(guard, 1200, broken, 5*time.Millisecond); got.failureKind != "" { + t.Fatalf("visible token must keep the original outcome: %+v", got) + } + ok := streamOutcome{logStatusCode: http.StatusOK} + if got := claudeNativeFirstTokenOutcome(guard, 0, ok, 5*time.Millisecond); got.logStatusCode != http.StatusOK { + t.Fatalf("successful stream must stay successful: %+v", got) + } + if got := claudeNativeFirstTokenOutcome(nil, 0, broken, 0); got.failureKind != "" { + t.Fatalf("nil guard must keep the original outcome: %+v", got) + } +} + +func TestActivateClaudeStreamKeepalive(t *testing.T) { + store := auth.NewStore(nil, nil, nil) + defer store.Stop() + claude := &auth.Account{DBID: 251, UpstreamType: auth.UpstreamClaude} + codex := &auth.Account{DBID: 1} + + newCtx := func() (*gin.Context, func()) { + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodPost, "/v1/messages", nil) + stop := installContinuousRetrySSEKeepalive(c, true, "text/event-stream; charset=utf-8") + return c, stop + } + + c, stop := newCtx() + activateClaudeStreamKeepalive(c.Request.Context(), store, claude, true) + if !continuousRetryKeepaliveActive(c.Request.Context()) { + t.Fatal("claude stream must activate the pre-first-token keepalive") + } + stop() + + c, stop = newCtx() + activateClaudeStreamKeepalive(c.Request.Context(), store, codex, true) + if continuousRetryKeepaliveActive(c.Request.Context()) { + t.Fatal("non-claude account must not activate the keepalive") + } + stop() + + c, stop = newCtx() + activateClaudeStreamKeepalive(c.Request.Context(), store, claude, false) + if continuousRetryKeepaliveActive(c.Request.Context()) { + t.Fatal("non-stream request must not activate the keepalive") + } + stop() + + store.SetClaudeStreamKeepaliveEnabled(false) + c, stop = newCtx() + activateClaudeStreamKeepalive(c.Request.Context(), store, claude, true) + if continuousRetryKeepaliveActive(c.Request.Context()) { + t.Fatal("disabled switch must not activate the keepalive") + } + stop() +} + +func TestClaudeFirstTokenSlow(t *testing.T) { + if claudeFirstTokenSlow(59_999) { + t.Fatal("below threshold must not be slow") + } + if !claudeFirstTokenSlow(60_000) { + t.Fatal("threshold must count as slow") + } + if claudeFirstTokenSlow(0) { + t.Fatal("no first token recorded must not be reported as slow") + } +} diff --git a/proxy/handler_anthropic.go b/proxy/handler_anthropic.go index 1c88d152..74697ac0 100644 --- a/proxy/handler_anthropic.go +++ b/proxy/handler_anthropic.go @@ -659,10 +659,13 @@ func (h *Handler) Messages(c *gin.Context) { attemptIdentity := ruleIdentity.WithSelectedAccount(account, h.store) upstreamCtx = WithPayloadRuleIdentity(upstreamCtx, attemptIdentity) lastUpstreamCancel = upstreamCancel - ttftGuard := newFirstTokenTimeoutGuard(currentFirstTokenTimeout(), upstreamCancel) + attemptFirstTokenTimeout := claudeFirstTokenTimeoutFor(h.store, account) + ttftGuard := newFirstTokenTimeoutGuard(attemptFirstTokenTimeout, upstreamCancel) var resp *http.Response var reqErr error if account.IsClaudeOAuth() { + // 首字前保活：长推理期间让下游能区分"上游在思考"与"连接已死"。 + activateClaudeStreamKeepalive(c.Request.Context(), h.store, account, isStream) // Claude Code OAuth 账号本身说 Anthropic Messages API：不翻译成 Codex， // 直接把原始入站 body 透传到 api.anthropic.com/v1/messages；返回的响应 // 已是原生 Anthropic SSE，打上原生路由标记复用既有透传链路。 @@ -741,7 +744,7 @@ func (h *Handler) Messages(c *gin.Context) { timedOut := ttftGuard.TimedOut() ttftGuard.Stop() if timedOut { - reqErr = firstTokenTimeoutError(currentFirstTokenTimeout()) + reqErr = firstTokenTimeoutError(attemptFirstTokenTimeout) } kind := classifyTransportFailure(reqErr) if wsHTTPFallback.ForceHTTP() && !useWebsocket { @@ -1016,6 +1019,8 @@ func (h *Handler) Messages(c *gin.Context) { applyAnthropicUsageSemantics(usage) } outcome = normalizeNativeFailureMessageForAccount(account, outcome) + outcome = claudeNativeFirstTokenOutcome(ttftGuard, firstTokenMs, outcome, attemptFirstTokenTimeout) + logClaudeFirstTokenLatency(account, attemptEffectiveModel, reasoningEffort, firstTokenMs, outcome, start) // The native forwarder consumes the body before returning. Synchronize // Anthropic's unified quota headers now, once per attempt, so Claude // usage remains fresh without adding a write before first token. @@ -1379,7 +1384,7 @@ func (h *Handler) Messages(c *gin.Context) { outcome = overlayContinuousRetryLocalFailure(outcome, readErr, writeErr) terminalFailurePayload, _ = resolvePreContentRetryErrorCandidate(terminalFailurePayload, preContentErrorCandidate, contentStarted, wroteAnyBody, gotTerminal, readErr, c.Request.Context().Err(), writeErr) if ttftGuard.TimedOut() && !ttftRecorded && !gotTerminal { - outcome = firstTokenTimeoutOutcome(currentFirstTokenTimeout()) + outcome = firstTokenTimeoutOutcome(attemptFirstTokenTimeout) } ttftGuard.Stop() if len(terminalFailurePayload) > 0 && !outcome.terminalLocal {