From 120f16e9e1836c013932428971409adbc33c1e4d Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Tue, 24 Mar 2026 03:40:51 +0900 Subject: [PATCH] fix: Claude token rotation, usage-exhausted fallback, and usage API caching - Fix token rotation not taking effect: getCurrentToken() was shadowed by static .env CLAUDE_CODE_OAUTH_TOKEN value in agent environment - Don't fall back to Kimi on usage-exhausted (only on transient 429/network) - Add disk cache for Claude usage API data (survives restarts, 429s) - Rate-limit usage API calls to 1 per token per 5 minutes - Add ignoreRateLimits option to rotateToken() for exhausted recovery - Rotate to next token in getActiveProvider() when current is exhausted - Add isUsageExhausted() helper to provider-fallback --- src/agent-runner-environment.ts | 3 +- src/claude-usage.ts | 89 ++++++-- src/message-agent-executor.test.ts | 269 ++++++++++++++++++++++ src/message-agent-executor.ts | 255 ++++++++++++++++++--- src/message-runtime.test.ts | 169 +++++++++++++- src/provider-fallback.test.ts | 103 +++++++++ src/provider-fallback.ts | 123 ++++++++++- src/task-scheduler.test.ts | 142 ++++++++++++ src/task-scheduler.ts | 344 +++++++++++++++++++++++++---- src/token-rotation.ts | 14 +- src/unified-dashboard.test.ts | 140 ++++++++++++ src/unified-dashboard.ts | 240 ++++++-------------- 12 files changed, 1633 insertions(+), 258 deletions(-) create mode 100644 src/message-agent-executor.test.ts create mode 100644 src/provider-fallback.test.ts create mode 100644 src/unified-dashboard.test.ts diff --git a/src/agent-runner-environment.ts b/src/agent-runner-environment.ts index 6608b7f..59687f5 100644 --- a/src/agent-runner-environment.ts +++ b/src/agent-runner-environment.ts @@ -123,9 +123,10 @@ function prepareClaudeEnvironment(args: { args.envVars.ANTHROPIC_BASE_URL || process.env.ANTHROPIC_BASE_URL || ''; } { + // Token rotation takes priority over static .env value const oauthToken = - args.envVars.CLAUDE_CODE_OAUTH_TOKEN || getCurrentToken() || + args.envVars.CLAUDE_CODE_OAUTH_TOKEN || process.env.CLAUDE_CODE_OAUTH_TOKEN; if (oauthToken) { args.env.CLAUDE_CODE_OAUTH_TOKEN = oauthToken; diff --git a/src/claude-usage.ts b/src/claude-usage.ts index 0f265b4..0c5def7 100644 --- a/src/claude-usage.ts +++ b/src/claude-usage.ts @@ -5,9 +5,15 @@ * Supports multiple tokens for rotation-aware usage checking. */ +import fs from 'fs'; +import path from 'path'; + +import { DATA_DIR } from './config.js'; import { logger } from './logger.js'; import { getCurrentToken, getAllTokens } from './token-rotation.js'; +const USAGE_CACHE_FILE = path.join(DATA_DIR, 'claude-usage-cache.json'); + const PROFILE_ENDPOINT = 'https://api.anthropic.com/api/oauth/profile'; export interface ClaudeUsageData { @@ -35,9 +41,50 @@ function mapWindow(w?: { return { utilization: w.utilization, resets_at: w.resets_at || '' }; } +// ── Disk cache for usage data (survives restarts, 429s) ── + +interface UsageCacheEntry { + usage: ClaudeUsageData; + fetchedAt: number; +} + +let usageDiskCache: Record = {}; +let diskCacheLoaded = false; + +function loadUsageDiskCache(): void { + if (diskCacheLoaded) return; + diskCacheLoaded = true; + try { + if (fs.existsSync(USAGE_CACHE_FILE)) { + usageDiskCache = JSON.parse(fs.readFileSync(USAGE_CACHE_FILE, 'utf-8')); + } + } catch { /* start fresh */ } +} + +function saveUsageDiskCache(): void { + try { + fs.writeFileSync(USAGE_CACHE_FILE, JSON.stringify(usageDiskCache)); + } catch { /* best effort */ } +} + +function cacheKey(token: string): string { + return token.slice(0, 12); +} + +// Rate limit: at most one API call per token per 5 minutes +const MIN_FETCH_INTERVAL_MS = 300_000; + async function fetchUsageForToken( token: string, ): Promise { + loadUsageDiskCache(); + + // Return cached data if fetched recently (avoid API rate-limit) + const key = cacheKey(token); + const cached = usageDiskCache[key]; + if (cached && Date.now() - cached.fetchedAt < MIN_FETCH_INTERVAL_MS) { + return cached.usage; + } const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS); @@ -57,32 +104,38 @@ async function fetchUsageForToken( return null; } if (res.status === 429) { - logger.warn('Claude usage API: rate limited (429)'); - return null; + logger.warn('Claude usage API: rate limited (429), returning cached data'); + return usageDiskCache[cacheKey(token)]?.usage ?? null; } if (!res.ok) { logger.warn( { status: res.status }, `Claude usage API: unexpected status ${res.status}`, ); - return null; + return usageDiskCache[cacheKey(token)]?.usage ?? null; } const data = (await res.json()) as UsageApiResponse; - return { + const result: ClaudeUsageData = { five_hour: mapWindow(data.five_hour), seven_day: mapWindow(data.seven_day), seven_day_sonnet: mapWindow(data.seven_day_sonnet), seven_day_opus: mapWindow(data.seven_day_opus), }; + + // Persist to disk cache + usageDiskCache[cacheKey(token)] = { usage: result, fetchedAt: Date.now() }; + saveUsageDiskCache(); + + return result; } catch (err) { if ((err as Error).name === 'AbortError') { logger.warn('Claude usage API: request timed out'); } else { logger.warn({ err }, 'Claude usage API: fetch failed'); } - return null; + return usageDiskCache[cacheKey(token)]?.usage ?? null; } finally { clearTimeout(timer); } @@ -103,12 +156,14 @@ export async function fetchClaudeUsage(): Promise { export interface ClaudeAccountProfile { email: string; - planType: string; // "max", "pro", "free" + planType: string; // "max", "pro", "free" } const profileCache = new Map(); -async function fetchProfileForToken(token: string): Promise { +async function fetchProfileForToken( + token: string, +): Promise { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS); try { @@ -121,14 +176,20 @@ async function fetchProfileForToken(token: string): Promise { } } -export function getClaudeProfile(index: number): ClaudeAccountProfile | undefined { +export function getClaudeProfile( + index: number, +): ClaudeAccountProfile | undefined { return profileCache.get(index); } diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts new file mode 100644 index 0000000..52e4bf4 --- /dev/null +++ b/src/message-agent-executor.test.ts @@ -0,0 +1,269 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./agent-runner.js', () => ({ + runAgentProcess: vi.fn(), + writeGroupsSnapshot: vi.fn(), + writeTasksSnapshot: vi.fn(), +})); + +vi.mock('./available-groups.js', () => ({ + listAvailableGroups: vi.fn(() => []), +})); + +vi.mock('./config.js', () => ({ + DATA_DIR: '/tmp/ejclaw-test-data', +})); + +vi.mock('./db.js', () => ({ + getAllTasks: vi.fn(() => []), +})); + +vi.mock('./logger.js', () => ({ + logger: { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +vi.mock('./provider-fallback.js', () => ({ + detectFallbackTrigger: vi.fn((error?: string | null) => { + const lower = (error || '').toLowerCase(); + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('hit your limit') + ) { + return { shouldFallback: true, reason: '429' }; + } + return { shouldFallback: false, reason: '' }; + }), + getActiveProvider: vi.fn(async () => 'claude'), + getFallbackEnvOverrides: vi.fn(() => ({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_AUTH_TOKEN: 'test-kimi-key', + ANTHROPIC_MODEL: 'kimi-k2.5', + })), + getFallbackProviderName: vi.fn(() => 'kimi'), + hasGroupProviderOverride: vi.fn(() => false), + isFallbackEnabled: vi.fn(() => true), + markPrimaryCooldown: vi.fn(), +})); + +vi.mock('./session-recovery.js', () => ({ + shouldResetSessionOnAgentFailure: vi.fn(() => false), +})); + +vi.mock('./token-rotation.js', () => ({ + rotateToken: vi.fn(() => false), + getTokenCount: vi.fn(() => 1), + markTokenHealthy: vi.fn(), +})); + +vi.mock('./codex-token-rotation.js', () => ({ + rotateCodexToken: vi.fn(() => false), + getCodexAccountCount: vi.fn(() => 1), + markCodexTokenHealthy: vi.fn(), +})); + +import * as agentRunner from './agent-runner.js'; +import { runAgentForGroup } from './message-agent-executor.js'; +import * as providerFallback from './provider-fallback.js'; +import * as tokenRotation from './token-rotation.js'; +import type { RegisteredGroup } from './types.js'; + +function makeGroup(): RegisteredGroup { + return { + name: 'Test Group', + folder: 'test-claude', + trigger: '@Andy', + added_at: new Date().toISOString(), + requiresTrigger: false, + agentType: 'claude-code', + }; +} + +function makeDeps() { + return { + assistantName: 'Andy', + queue: { + registerProcess: vi.fn(), + }, + getRegisteredGroups: () => ({}), + getSessions: () => ({}), + persistSession: vi.fn(), + clearSession: vi.fn(), + }; +} + +describe('runAgentForGroup Claude rotation', () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(providerFallback.getActiveProvider).mockResolvedValue('claude'); + vi.mocked(providerFallback.isFallbackEnabled).mockReturnValue(true); + vi.mocked(providerFallback.hasGroupProviderOverride).mockReturnValue(false); + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(1); + vi.mocked(tokenRotation.rotateToken).mockReturnValue(false); + }); + + it('rotates to another Claude account before falling back to Kimi', async () => { + const outputs: string[] = []; + + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(2); + vi.mocked(tokenRotation.rotateToken).mockReturnValueOnce(true); + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'You’re out of extra usage · resets 4am (Asia/Seoul)', + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '회전된 Claude 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const result = await runAgentForGroup(makeDeps(), { + group: makeGroup(), + prompt: 'hello', + chatJid: 'group@test', + runId: 'run-rotate-claude', + onOutput: async (output) => { + if (typeof output.result === 'string') outputs.push(output.result); + }, + }); + + expect(result).toBe('success'); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(tokenRotation.rotateToken).toHaveBeenCalledTimes(1); + expect(tokenRotation.markTokenHealthy).toHaveBeenCalledTimes(1); + expect(providerFallback.markPrimaryCooldown).not.toHaveBeenCalled(); + expect(outputs).toEqual(['회전된 Claude 응답입니다.']); + }); + + it('falls back to Kimi only after all Claude accounts are exhausted', async () => { + const outputs: string[] = []; + + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(2); + vi.mocked(tokenRotation.rotateToken) + .mockReturnValueOnce(true) + .mockReturnValueOnce(false); + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'You’re out of extra usage · resets 4am (Asia/Seoul)', + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'Kimi 폴백 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const result = await runAgentForGroup(makeDeps(), { + group: makeGroup(), + prompt: 'hello', + chatJid: 'group@test', + runId: 'run-fallback-after-rotation', + onOutput: async (output) => { + if (typeof output.result === 'string') outputs.push(output.result); + }, + }); + + expect(result).toBe('success'); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(3); + expect(tokenRotation.rotateToken).toHaveBeenCalledTimes(2); + expect(providerFallback.markPrimaryCooldown).toHaveBeenCalledWith( + 'usage-exhausted', + undefined, + ); + expect(agentRunner.runAgentProcess).toHaveBeenNthCalledWith( + 3, + expect.anything(), + expect.anything(), + expect.any(Function), + expect.any(Function), + expect.objectContaining({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_MODEL: 'kimi-k2.5', + }), + ); + expect(outputs).toEqual(['Kimi 폴백 응답입니다.']); + }); + + it('does not mistake a normal response quoting the banner text for a usage error', async () => { + const outputs: string[] = []; + + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(2); + + vi.mocked(agentRunner.runAgentProcess).mockImplementationOnce( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: + "상태 문구 예시: You're out of extra usage · resets 4am (Asia/Seoul) 라는 배너가 뜰 수 있습니다.", + }); + return { + status: 'success', + result: null, + }; + }, + ); + + const result = await runAgentForGroup(makeDeps(), { + group: makeGroup(), + prompt: 'hello', + chatJid: 'group@test', + runId: 'run-normal-quoted-banner', + onOutput: async (output) => { + if (typeof output.result === 'string') outputs.push(output.result); + }, + }); + + expect(result).toBe('success'); + expect(tokenRotation.rotateToken).not.toHaveBeenCalled(); + expect(providerFallback.markPrimaryCooldown).not.toHaveBeenCalled(); + expect(outputs).toEqual([ + "상태 문구 예시: You're out of extra usage · resets 4am (Asia/Seoul) 라는 배너가 뜰 수 있습니다.", + ]); + }); +}); diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 9b21d3c..82b9568 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -18,6 +18,7 @@ import { getFallbackProviderName, hasGroupProviderOverride, isFallbackEnabled, + isUsageExhausted, markPrimaryCooldown, } from './provider-fallback.js'; import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; @@ -42,6 +43,25 @@ export interface MessageAgentExecutorDeps { clearSession: (groupFolder: string) => void; } +function isClaudeUsageExhaustedMessage(text: string): boolean { + const normalized = text + .trim() + .toLowerCase() + .replace(/[’‘`]/g, "'") + .replace(/\s+/g, ' ') + .replace(/^error:\s*/i, ''); + const looksLikeBanner = + normalized.startsWith("you're out of extra usage") || + normalized.startsWith('you are out of extra usage') || + normalized.startsWith("you've hit your limit") || + normalized.startsWith('you have hit your limit'); + const hasResetHint = + normalized.includes('resets ') || + normalized.includes('reset at ') || + normalized.includes('try again'); + return looksLikeBanner && hasResetHint && normalized.length <= 160; +} + export async function runAgentForGroup( deps: MessageAgentExecutorDeps, args: { @@ -137,6 +157,30 @@ export async function runAgentForGroup( ) { resetSessionRequested = true; } + if ( + canFallback && + provider === 'claude' && + output.status === 'success' && + !sawOutput && + typeof output.result === 'string' && + isClaudeUsageExhaustedMessage(output.result) + ) { + if (!streamedTriggerReason) { + logger.warn( + { + chatJid, + group: group.name, + runId, + resultPreview: output.result.slice(0, 120), + }, + 'Detected Claude usage exhaustion banner in successful output', + ); + } + streamedTriggerReason = { + reason: 'usage-exhausted', + }; + return; + } if (output.result !== null && output.result !== undefined) { sawOutput = true; } else if ( @@ -291,7 +335,164 @@ export async function runAgentForGroup( return 'success'; }; - const provider = canFallback ? getActiveProvider() : 'claude'; + const shouldRotateClaudeToken = (reason: string): boolean => + reason === '429' || reason === 'usage-exhausted'; + + const retryClaudeWithRotation = async ( + initialTrigger: { + reason: string; + retryAfterMs?: number; + }, + rotationMessage?: string, + ): Promise<'success' | 'error'> => { + let trigger = initialTrigger; + let lastRotationMessage = rotationMessage; + + while ( + shouldRotateClaudeToken(trigger.reason) && + getTokenCount() > 1 && + rotateToken(lastRotationMessage) + ) { + logger.info( + { chatJid, group: group.name, runId, reason: trigger.reason }, + 'Claude rate-limited, retrying with rotated account', + ); + + const retryAttempt = await runAttempt('claude'); + + if (retryAttempt.error) { + if (!retryAttempt.sawOutput) { + const errMsg = + retryAttempt.error instanceof Error + ? retryAttempt.error.message + : String(retryAttempt.error); + const retryTrigger = retryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: retryAttempt.streamedTriggerReason.reason, + retryAfterMs: retryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(errMsg); + if (retryTrigger.shouldFallback) { + trigger = { + reason: retryTrigger.reason, + retryAfterMs: retryTrigger.retryAfterMs, + }; + lastRotationMessage = errMsg; + continue; + } + } + + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: 'claude', + err: retryAttempt.error, + }, + 'Rotated Claude account also threw', + ); + return 'error'; + } + + const retryOutput = retryAttempt.output; + if (!retryOutput) { + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: 'claude', + }, + 'Rotated Claude account produced no output object', + ); + return 'error'; + } + + if ( + !retryAttempt.sawOutput && + retryAttempt.streamedTriggerReason && + retryOutput.status !== 'error' + ) { + trigger = { + reason: retryAttempt.streamedTriggerReason.reason, + retryAfterMs: retryAttempt.streamedTriggerReason.retryAfterMs, + }; + lastRotationMessage = + typeof retryOutput.result === 'string' ? retryOutput.result : undefined; + continue; + } + + if ( + !retryAttempt.sawOutput && + retryAttempt.sawSuccessNullResultWithoutOutput + ) { + return runFallbackAttempt('success-null-result'); + } + + if (retryOutput.status === 'error') { + if (!retryAttempt.sawOutput) { + const retryTrigger = retryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: retryAttempt.streamedTriggerReason.reason, + retryAfterMs: retryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(retryOutput.error); + if (retryTrigger.shouldFallback) { + trigger = { + reason: retryTrigger.reason, + retryAfterMs: retryTrigger.retryAfterMs, + }; + lastRotationMessage = retryOutput.error ?? undefined; + continue; + } + } + + logger.error( + { + group: group.name, + chatJid, + runId, + provider: 'claude', + error: retryOutput.error, + }, + 'Rotated Claude account failed', + ); + return 'error'; + } + + markTokenHealthy(); + return 'success'; + } + + // Usage exhausted: don't fall back to Kimi — log only, no response + if (trigger.reason === 'usage-exhausted') { + markPrimaryCooldown(trigger.reason, trigger.retryAfterMs); + logger.info( + { chatJid, group: group.name, runId }, + 'All Claude tokens usage-exhausted, silently skipping (no Kimi fallback)', + ); + return 'error'; + } + + return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + }; + + const provider = canFallback ? await getActiveProvider() : 'claude'; + + // Already in usage-exhausted cooldown — log only, no response + if (provider !== 'claude' && isUsageExhausted()) { + logger.info( + { chatJid, group: group.name, runId, provider }, + 'Claude usage exhausted (cooldown active), silently skipping', + ); + return 'error'; + } + const primaryAttempt = await runAttempt(provider); if (primaryAttempt.error) { @@ -308,19 +509,13 @@ export async function runAgentForGroup( } : detectFallbackTrigger(errMsg); if (trigger.shouldFallback) { - // Try rotating token before falling back to another provider - if (getTokenCount() > 1 && rotateToken(errMsg)) { - logger.info( - { chatJid, group: group.name, runId, reason: trigger.reason }, - 'Rate-limited, retrying with rotated token', - ); - const retryAttempt = await runAttempt('claude'); - if (!retryAttempt.error) { - markTokenHealthy(); - return 'success'; - } - } - return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + return retryClaudeWithRotation( + { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }, + errMsg, + ); } } @@ -353,6 +548,19 @@ export async function runAgentForGroup( return 'error'; } + if ( + canFallback && + provider === 'claude' && + !primaryAttempt.sawOutput && + primaryAttempt.streamedTriggerReason && + output.status !== 'error' + ) { + return retryClaudeWithRotation({ + reason: primaryAttempt.streamedTriggerReason.reason, + retryAfterMs: primaryAttempt.streamedTriggerReason.retryAfterMs, + }); + } + if ( canFallback && provider === 'claude' && @@ -383,18 +591,13 @@ export async function runAgentForGroup( } : detectFallbackTrigger(output.error); if (trigger.shouldFallback) { - if (getTokenCount() > 1 && rotateToken(output.error ?? undefined)) { - logger.info( - { chatJid, group: group.name, runId, reason: trigger.reason }, - 'Rate-limited (output error), retrying with rotated token', - ); - const retryAttempt = await runAttempt('claude'); - if (!retryAttempt.error) { - markTokenHealthy(); - return 'success'; - } - } - return runFallbackAttempt(trigger.reason, trigger.retryAfterMs); + return retryClaudeWithRotation( + { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }, + output.error ?? undefined, + ); } } diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 1af4b31..cadd29f 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -122,7 +122,7 @@ vi.mock('./logger.js', () => ({ vi.mock('./provider-fallback.js', () => ({ detectFallbackTrigger: vi.fn(() => ({ shouldFallback: false, reason: '' })), - getActiveProvider: vi.fn(() => 'claude'), + getActiveProvider: vi.fn(async () => 'claude'), getFallbackEnvOverrides: vi.fn(() => ({ ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', ANTHROPIC_AUTH_TOKEN: 'test-kimi-key', @@ -181,7 +181,7 @@ function makeChannel(chatJid: string): Channel { describe('createMessageRuntime', () => { beforeEach(() => { vi.clearAllMocks(); - vi.mocked(providerFallback.getActiveProvider).mockReturnValue('claude'); + vi.mocked(providerFallback.getActiveProvider).mockResolvedValue('claude'); vi.mocked(providerFallback.getFallbackProviderName).mockReturnValue('kimi'); vi.mocked(providerFallback.getFallbackEnvOverrides).mockReturnValue({ ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', @@ -1957,6 +1957,171 @@ describe('createMessageRuntime', () => { ); }); + it('retries with the fallback provider when Claude returns only a usage exhaustion banner', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const channel = makeChannel(chatJid); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-24T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'usage fallback 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-fallback-usage-exhausted', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(providerFallback.markPrimaryCooldown).toHaveBeenCalledWith( + 'usage-exhausted', + undefined, + ); + expect(channel.sendMessage).toHaveBeenCalledTimes(1); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'usage fallback 응답입니다.', + ); + }); + + it('suppresses duplicate streamed usage banners before retrying the fallback provider', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const channel = makeChannel(chatJid); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-24T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'intermediate', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + await onOutput?.({ + status: 'success', + phase: 'final', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + return { + status: 'success', + result: null, + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'duplicate banner fallback 응답입니다.', + }); + return { + status: 'success', + result: null, + }; + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-fallback-usage-exhausted-duplicate', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(providerFallback.markPrimaryCooldown).toHaveBeenCalledWith( + 'usage-exhausted', + undefined, + ); + expect(channel.sendMessage).toHaveBeenCalledTimes(1); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'duplicate banner fallback 응답입니다.', + ); + }); + it('retries with the fallback provider when Claude ends with success-null-result before any output', async () => { const chatJid = 'group@test'; const group = makeGroup('claude-code'); diff --git a/src/provider-fallback.test.ts b/src/provider-fallback.test.ts new file mode 100644 index 0000000..54a9cc9 --- /dev/null +++ b/src/provider-fallback.test.ts @@ -0,0 +1,103 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./claude-usage.js', () => ({ + fetchClaudeUsage: vi.fn(), +})); + +vi.mock('./env.js', () => ({ + readEnvFile: vi.fn(() => ({ + FALLBACK_PROVIDER_NAME: 'kimi', + FALLBACK_BASE_URL: 'https://api.kimi.com/coding/', + FALLBACK_AUTH_TOKEN: 'test-kimi-key', + FALLBACK_MODEL: 'kimi-k2.5', + FALLBACK_SMALL_MODEL: 'kimi-k2.5', + FALLBACK_COOLDOWN_MS: '600000', + })), +})); + +vi.mock('./logger.js', () => ({ + logger: { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +import { fetchClaudeUsage } from './claude-usage.js'; +import { + clearPrimaryCooldown, + getActiveProvider, + getCooldownInfo, + markPrimaryCooldown, + resetFallbackConfig, +} from './provider-fallback.js'; + +describe('provider fallback usage recovery', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-03-24T00:00:00.000Z')); + vi.clearAllMocks(); + clearPrimaryCooldown(); + resetFallbackConfig(); + delete process.env.FALLBACK_PROVIDER_NAME; + delete process.env.FALLBACK_BASE_URL; + delete process.env.FALLBACK_AUTH_TOKEN; + delete process.env.FALLBACK_MODEL; + delete process.env.FALLBACK_SMALL_MODEL; + delete process.env.FALLBACK_COOLDOWN_MS; + }); + + afterEach(() => { + clearPrimaryCooldown(); + resetFallbackConfig(); + vi.useRealTimers(); + }); + + it('keeps the fallback provider active while Claude usage is still exhausted', async () => { + vi.mocked(fetchClaudeUsage).mockResolvedValue({ + five_hour: { + utilization: 100, + resets_at: '2026-03-24T04:00:00.000+09:00', + }, + }); + + markPrimaryCooldown('usage-exhausted', 1_000); + vi.advanceTimersByTime(5_000); + + await expect(getActiveProvider()).resolves.toBe('kimi'); + expect(getCooldownInfo()).toMatchObject({ + active: true, + reason: 'usage-exhausted', + remainingMs: 0, + }); + }); + + it('returns to Claude immediately when usage is no longer exhausted', async () => { + vi.mocked(fetchClaudeUsage).mockResolvedValue({ + five_hour: { + utilization: 72, + resets_at: '2026-03-24T04:00:00.000+09:00', + }, + seven_day: { + utilization: 55, + resets_at: '2026-03-31T04:00:00.000+09:00', + }, + }); + + markPrimaryCooldown('usage-exhausted', 600_000); + + await expect(getActiveProvider()).resolves.toBe('claude'); + expect(getCooldownInfo()).toEqual({ active: false }); + }); + + it('falls back to time-based retry when usage status cannot be fetched', async () => { + vi.mocked(fetchClaudeUsage).mockResolvedValue(null); + + markPrimaryCooldown('usage-exhausted', 1_000); + vi.advanceTimersByTime(5_000); + + await expect(getActiveProvider()).resolves.toBe('claude'); + expect(getCooldownInfo()).toEqual({ active: false }); + }); +}); diff --git a/src/provider-fallback.ts b/src/provider-fallback.ts index f79d6c0..5140bdb 100644 --- a/src/provider-fallback.ts +++ b/src/provider-fallback.ts @@ -14,8 +14,10 @@ import fs from 'fs'; +import { fetchClaudeUsage, type ClaudeUsageData } from './claude-usage.js'; import { readEnvFile } from './env.js'; import { logger } from './logger.js'; +import { rotateToken, getTokenCount } from './token-rotation.js'; // ── Types ──────────────────────────────────────────────────────── @@ -46,6 +48,17 @@ interface FallbackConfig { // ── State ──────────────────────────────────────────────────────── let cooldown: CooldownState | null = null; +let lastUsageAvailabilityCheck: + | { + checkedAt: number; + result: 'available' | 'exhausted' | 'unknown'; + } + | null = null; +let usageAvailabilityCheckPromise: + | Promise<'available' | 'exhausted' | 'unknown'> + | null = null; + +const USAGE_RECOVERY_RECHECK_MS = 30_000; // ── Config ─────────────────────────────────────────────────────── @@ -116,16 +129,109 @@ export function getFallbackProviderName(): string { return loadConfig().providerName; } +function normalizeUtilization(utilization: number): number { + return utilization > 1 ? utilization : utilization * 100; +} + +type ClaudeUsageWindow = NonNullable; + +function hasExhaustedClaudeUsageWindow( + usage: ClaudeUsageData | null, +): boolean | null { + if (!usage) return null; + const windows: ClaudeUsageWindow[] = []; + if (usage.five_hour) windows.push(usage.five_hour); + if (usage.seven_day) windows.push(usage.seven_day); + if (usage.seven_day_sonnet) windows.push(usage.seven_day_sonnet); + if (usage.seven_day_opus) windows.push(usage.seven_day_opus); + if (windows.length === 0) return null; + return windows.some( + (window) => normalizeUtilization(window.utilization) >= 100, + ); +} + +function clearUsageAvailabilityCache(): void { + lastUsageAvailabilityCheck = null; + usageAvailabilityCheckPromise = null; +} + +async function getClaudeUsageAvailability(): Promise< + 'available' | 'exhausted' | 'unknown' +> { + const now = Date.now(); + if ( + lastUsageAvailabilityCheck && + now - lastUsageAvailabilityCheck.checkedAt < USAGE_RECOVERY_RECHECK_MS + ) { + return lastUsageAvailabilityCheck.result; + } + + if (!usageAvailabilityCheckPromise) { + usageAvailabilityCheckPromise = (async () => { + const usage = await fetchClaudeUsage(); + const exhausted = hasExhaustedClaudeUsageWindow(usage); + const result = + exhausted === null ? 'unknown' : exhausted ? 'exhausted' : 'available'; + lastUsageAvailabilityCheck = { + checkedAt: Date.now(), + result, + }; + return result; + })(); + + void usageAvailabilityCheckPromise.finally(() => { + usageAvailabilityCheckPromise = null; + }); + } + + return usageAvailabilityCheckPromise; +} + /** * Determine which provider should be used for the next request. * Returns 'claude' when Claude is healthy or cooldown has expired, * or the fallback provider name during an active cooldown. */ -export function getActiveProvider(): string { +export async function getActiveProvider(): Promise { const config = loadConfig(); if (!config.enabled) return 'claude'; if (cooldown) { + if (cooldown.reason === 'usage-exhausted') { + const usageAvailability = await getClaudeUsageAvailability(); + if (usageAvailability === 'available') { + logger.info( + { + provider: 'claude', + reason: cooldown.reason, + }, + 'Claude usage recovered, retrying primary provider', + ); + cooldown = null; + clearUsageAvailabilityCache(); + return 'claude'; + } + if (usageAvailability === 'exhausted') { + // Current token exhausted — try rotating to another token (ignore cooldowns) + if (getTokenCount() > 1 && rotateToken(undefined, { ignoreRateLimits: true })) { + logger.info( + 'Claude current token exhausted, rotated to next token — retrying', + ); + cooldown = null; + clearUsageAvailabilityCache(); + return 'claude'; + } + logger.debug( + { + provider: config.providerName, + reason: cooldown.reason, + }, + 'All Claude tokens exhausted, keeping cooldown active', + ); + return config.providerName; + } + } + if (Date.now() < cooldown.expiresAt) { return config.providerName; } @@ -161,6 +267,7 @@ export function markPrimaryCooldown( expiresAt: now + durationMs, reason, }; + clearUsageAvailabilityCache(); logger.info( { @@ -175,6 +282,7 @@ export function markPrimaryCooldown( /** Manually clear cooldown (e.g. after a successful Claude response). */ export function clearPrimaryCooldown(): void { + clearUsageAvailabilityCache(); if (cooldown) { logger.info( { reason: cooldown.reason }, @@ -184,6 +292,11 @@ export function clearPrimaryCooldown(): void { } } +/** Check if Claude is currently in usage-exhausted cooldown. */ +export function isUsageExhausted(): boolean { + return cooldown?.reason === 'usage-exhausted'; +} + /** Get current cooldown info (for diagnostics / status dashboard). */ export function getCooldownInfo(): { active: boolean; @@ -191,14 +304,18 @@ export function getCooldownInfo(): { expiresAt?: string; remainingMs?: number; } { - if (!cooldown || Date.now() >= cooldown.expiresAt) { + if (!cooldown) { + return { active: false }; + } + const remainingMs = Math.max(cooldown.expiresAt - Date.now(), 0); + if (cooldown.reason !== 'usage-exhausted' && remainingMs === 0) { return { active: false }; } return { active: true, reason: cooldown.reason, expiresAt: new Date(cooldown.expiresAt).toISOString(), - remainingMs: cooldown.expiresAt - Date.now(), + remainingMs, }; } diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 97fa40d..6bf9a37 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -8,6 +8,42 @@ const { runAgentProcessMock, writeTasksSnapshotMock } = vi.hoisted(() => ({ writeTasksSnapshotMock: vi.fn(), })); +vi.mock('./provider-fallback.js', () => ({ + detectFallbackTrigger: vi.fn((error?: string | null) => { + const lower = (error || '').toLowerCase(); + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('hit your limit') + ) { + return { shouldFallback: true, reason: '429' }; + } + return { shouldFallback: false, reason: '' }; + }), + getActiveProvider: vi.fn(async () => 'claude'), + getFallbackEnvOverrides: vi.fn(() => ({ + ANTHROPIC_BASE_URL: 'https://api.kimi.com/coding/', + ANTHROPIC_AUTH_TOKEN: 'test-kimi-key', + ANTHROPIC_MODEL: 'kimi-k2.5', + })), + getFallbackProviderName: vi.fn(() => 'kimi'), + hasGroupProviderOverride: vi.fn(() => false), + isFallbackEnabled: vi.fn(() => true), + markPrimaryCooldown: vi.fn(), +})); + +vi.mock('./token-rotation.js', () => ({ + rotateToken: vi.fn(() => false), + getTokenCount: vi.fn(() => 1), + markTokenHealthy: vi.fn(), +})); + +vi.mock('./codex-token-rotation.js', () => ({ + rotateCodexToken: vi.fn(() => false), + getCodexAccountCount: vi.fn(() => 1), + markCodexTokenHealthy: vi.fn(), +})); + vi.mock('./agent-runner.js', async () => { const actual = await vi.importActual( @@ -21,8 +57,10 @@ vi.mock('./agent-runner.js', async () => { }); import { _initTestDatabase, createTask, getTaskById } from './db.js'; +import * as providerFallback from './provider-fallback.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js'; +import * as tokenRotation from './token-rotation.js'; import { _resetSchedulerLoopForTests, computeNextRun, @@ -39,6 +77,11 @@ describe('task scheduler', () => { _resetSchedulerLoopForTests(); runAgentProcessMock.mockClear(); writeTasksSnapshotMock.mockClear(); + vi.mocked(providerFallback.getActiveProvider).mockResolvedValue('claude'); + vi.mocked(providerFallback.isFallbackEnabled).mockReturnValue(true); + vi.mocked(providerFallback.hasGroupProviderOverride).mockReturnValue(false); + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(1); + vi.mocked(tokenRotation.rotateToken).mockReturnValue(false); vi.useFakeTimers(); }); @@ -232,6 +275,103 @@ Check the run. expect(enqueueTask.mock.calls[0][1]).toBe('task-watch-group'); }); + it('suppresses Claude usage banners for scheduled tasks and retries with a rotated account', async () => { + const dueAt = new Date(Date.now() - 60_000).toISOString(); + createTask({ + id: 'task-usage-banner', + group_folder: 'shared-group', + chat_jid: 'shared@g.us', + agent_type: 'claude-code', + prompt: 'claude task', + schedule_type: 'once', + schedule_value: dueAt, + context_mode: 'isolated', + next_run: dueAt, + status: 'active', + created_at: '2026-02-22T00:00:00.000Z', + }); + + vi.mocked(tokenRotation.getTokenCount).mockReturnValue(2); + vi.mocked(tokenRotation.rotateToken).mockReturnValueOnce(true); + + (runAgentProcessMock as any) + .mockImplementationOnce( + async ( + _group: unknown, + _input: unknown, + _onProcess: unknown, + onOutput?: (output: Record) => Promise, + ) => { + await onOutput?.({ + status: 'success', + phase: 'intermediate', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + await onOutput?.({ + status: 'success', + result: "You're out of extra usage · resets 4am (Asia/Seoul)", + }); + return { + status: 'success', + result: null, + }; + }, + ) + .mockImplementationOnce( + async ( + _group: unknown, + _input: unknown, + _onProcess: unknown, + onOutput?: (output: Record) => Promise, + ) => { + await onOutput?.({ + status: 'success', + result: 'rotated scheduled task response', + }); + return { + status: 'success', + result: null, + }; + }, + ); + + const enqueueTask = vi.fn( + (_groupJid: string, _taskId: string, fn: () => Promise) => { + void fn(); + }, + ); + const sendMessage = vi.fn(async () => {}); + + startSchedulerLoop({ + serviceAgentType: 'claude-code', + registeredGroups: () => ({ + 'shared@g.us': { + name: 'Shared', + folder: 'shared-group', + trigger: '@Claude', + added_at: '2026-02-22T00:00:00.000Z', + agentType: 'claude-code', + }, + }), + getSessions: () => ({}), + queue: { enqueueTask } as any, + onProcess: () => {}, + sendMessage, + }); + + await vi.advanceTimersByTimeAsync(10); + + expect(runAgentProcessMock).toHaveBeenCalledTimes(2); + expect(tokenRotation.rotateToken).toHaveBeenCalledTimes(1); + expect(tokenRotation.markTokenHealthy).toHaveBeenCalledTimes(1); + expect(providerFallback.markPrimaryCooldown).not.toHaveBeenCalled(); + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(sendMessage).toHaveBeenCalledWith( + 'shared@g.us', + 'rotated scheduled task response', + ); + }); + it('picks up newly due tasks immediately when nudged', async () => { const enqueueTask = vi.fn(); @@ -352,6 +492,7 @@ Check the run. }), expect.any(Function), expect.any(Function), + undefined, ); expect(writeTasksSnapshotMock).toHaveBeenCalledWith( 'shared-group', @@ -411,6 +552,7 @@ Check the run. }), expect.any(Function), expect.any(Function), + undefined, ); expect(writeTasksSnapshotMock).toHaveBeenCalledWith( 'shared-group', diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 3542c0a..903554b 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -1,9 +1,11 @@ import { ChildProcess } from 'child_process'; import { CronExpressionParser } from 'cron-parser'; import fs from 'fs'; +import path from 'path'; import { ASSISTANT_NAME, + DATA_DIR, SCHEDULER_POLL_INTERVAL, SERVICE_AGENT_TYPE, TIMEZONE, @@ -29,7 +31,16 @@ import { } from './group-folder.js'; import { logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; -import { detectFallbackTrigger } from './provider-fallback.js'; +import { + detectFallbackTrigger, + getActiveProvider, + getFallbackEnvOverrides, + getFallbackProviderName, + hasGroupProviderOverride, + isFallbackEnabled, + isUsageExhausted, + markPrimaryCooldown, +} from './provider-fallback.js'; import { rotateCodexToken, getCodexAccountCount, @@ -61,6 +72,25 @@ export { shouldUseTaskScopedSession, } from './task-watch-status.js'; +function isClaudeUsageExhaustedMessage(text: string): boolean { + const normalized = text + .trim() + .toLowerCase() + .replace(/[’‘`]/g, "'") + .replace(/\s+/g, ' ') + .replace(/^error:\s*/i, ''); + const looksLikeBanner = + normalized.startsWith("you're out of extra usage") || + normalized.startsWith('you are out of extra usage') || + normalized.startsWith("you've hit your limit") || + normalized.startsWith('you have hit your limit'); + const hasResetHint = + normalized.includes('resets ') || + normalized.includes('reset at ') || + normalized.includes('try again'); + return looksLikeBanner && hasResetHint && normalized.length <= 160; +} + /** * Compute the next run time for a recurring task, anchored to the * task's scheduled time rather than Date.now() to prevent cumulative @@ -249,52 +279,292 @@ async function runTask( sendTrackedMessage: deps.sendTrackedMessage, editTrackedMessage: deps.editTrackedMessage, }); + const settingsPath = path.join( + DATA_DIR, + 'sessions', + task.group_folder, + '.claude', + 'settings.json', + ); + const canFallback = + context.taskAgentType === 'claude-code' && + isFallbackEnabled() && + !hasGroupProviderOverride(settingsPath); try { await statusTracker.update('checking'); - const output = await runAgentProcess( - context.group, - { - prompt: task.prompt, - sessionId: context.sessionId, - groupFolder: task.group_folder, - chatJid: task.chat_jid, - isMain: context.isMain, - isScheduledTask: true, - runtimeTaskId: context.runtimeTaskId, - useTaskScopedSession: context.useTaskScopedSession, - assistantName: ASSISTANT_NAME, + const runTaskAttempt = async ( + provider: string, + ): Promise<{ + output: AgentOutput; + sawOutput: boolean; + streamedTriggerReason?: { + reason: string; + retryAfterMs?: number; + }; + attemptResult: string | null; + attemptError: string | null; + }> => { + let sawOutput = false; + let attemptResult: string | null = null; + let attemptError: string | null = null; + let streamedTriggerReason: + | { + reason: string; + retryAfterMs?: number; + } + | undefined; + + const output = await runAgentProcess( + context.group, + { + prompt: task.prompt, + sessionId: context.sessionId, + groupFolder: task.group_folder, + chatJid: task.chat_jid, + isMain: context.isMain, + isScheduledTask: true, + runtimeTaskId: context.runtimeTaskId, + useTaskScopedSession: context.useTaskScopedSession, + assistantName: ASSISTANT_NAME, + }, + (proc, processName) => + deps.onProcess( + context.queueJid, + proc, + processName, + context.runtimeIpcDir, + ), + async (streamedOutput: AgentOutput) => { + if (streamedOutput.phase === 'progress') { + return; + } + + if ( + canFallback && + provider === 'claude' && + !sawOutput && + streamedOutput.status === 'success' && + typeof streamedOutput.result === 'string' && + isClaudeUsageExhaustedMessage(streamedOutput.result) + ) { + if (!streamedTriggerReason) { + logger.warn( + { + taskId: task.id, + taskChatJid: task.chat_jid, + group: context.group.name, + groupFolder: task.group_folder, + resultPreview: streamedOutput.result.slice(0, 120), + }, + 'Detected Claude usage exhaustion banner during scheduled task output', + ); + } + streamedTriggerReason = { reason: 'usage-exhausted' }; + return; + } + + if (streamedOutput.result) { + sawOutput = true; + attemptResult = streamedOutput.result; + await deps.sendMessage(task.chat_jid, streamedOutput.result); + } + + if (streamedOutput.status === 'error') { + attemptError = streamedOutput.error || 'Unknown error'; + if (!sawOutput && !streamedTriggerReason) { + const trigger = detectFallbackTrigger(streamedOutput.error); + if (trigger.shouldFallback) { + streamedTriggerReason = { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; + } + } + } + }, + provider === 'claude' ? undefined : getFallbackEnvOverrides(), + ); + + if (output.status === 'error' && !attemptError) { + attemptError = output.error || 'Unknown error'; + } else if (output.result && !attemptResult) { + attemptResult = output.result; + } + + return { + output, + sawOutput, + streamedTriggerReason, + attemptResult, + attemptError, + }; + }; + + const shouldRotateClaudeToken = (reason: string): boolean => + reason === '429' || reason === 'usage-exhausted'; + + const runFallbackTaskAttempt = async ( + reason: string, + retryAfterMs?: number, + ): Promise => { + if (!canFallback) { + error = reason; + return; + } + + const fallbackName = getFallbackProviderName(); + markPrimaryCooldown(reason, retryAfterMs); + + logger.info( + { + taskId: task.id, + group: context.group.name, + groupFolder: task.group_folder, + reason, + retryAfterMs, + fallbackProvider: fallbackName, + }, + `Falling back to provider: ${fallbackName} for scheduled task (reason: ${reason})`, + ); + + const fallbackAttempt = await runTaskAttempt(fallbackName); + result = fallbackAttempt.attemptResult; + error = + fallbackAttempt.output.status === 'error' + ? fallbackAttempt.attemptError || 'Unknown error' + : null; + }; + + const retryClaudeTaskWithRotation = async ( + initialTrigger: { + reason: string; + retryAfterMs?: number; }, - (proc, processName) => - deps.onProcess( - context.queueJid, - proc, - processName, - context.runtimeIpcDir, - ), - async (streamedOutput: AgentOutput) => { - if (streamedOutput.phase === 'progress') { + rotationMessage?: string, + ): Promise => { + let trigger = initialTrigger; + let lastRotationMessage = rotationMessage; + + while ( + shouldRotateClaudeToken(trigger.reason) && + getTokenCount() > 1 && + rotateToken(lastRotationMessage) + ) { + logger.info( + { + taskId: task.id, + group: context.group.name, + groupFolder: task.group_folder, + reason: trigger.reason, + }, + 'Scheduled task Claude rate-limited, retrying with rotated account', + ); + + const retryAttempt = await runTaskAttempt('claude'); + result = retryAttempt.attemptResult; + error = retryAttempt.attemptError; + + if ( + retryAttempt.streamedTriggerReason && + !retryAttempt.sawOutput && + retryAttempt.output.status !== 'error' + ) { + trigger = { + reason: retryAttempt.streamedTriggerReason.reason, + retryAfterMs: retryAttempt.streamedTriggerReason.retryAfterMs, + }; + lastRotationMessage = + typeof retryAttempt.output.result === 'string' + ? retryAttempt.output.result + : undefined; + continue; + } + + if (retryAttempt.output.status === 'error' && !retryAttempt.sawOutput) { + const retryTrigger = retryAttempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: retryAttempt.streamedTriggerReason.reason, + retryAfterMs: retryAttempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(retryAttempt.attemptError); + if (retryTrigger.shouldFallback) { + trigger = { + reason: retryTrigger.reason, + retryAfterMs: retryTrigger.retryAfterMs, + }; + lastRotationMessage = retryAttempt.attemptError || undefined; + continue; + } + } + + if (retryAttempt.output.status === 'success') { + markTokenHealthy(); + error = null; return; } - if (streamedOutput.result) { - result = streamedOutput.result; - // Forward result to user (sendMessage handles formatting) - await deps.sendMessage(task.chat_jid, streamedOutput.result); - } - if (streamedOutput.status === 'error') { - error = streamedOutput.error || 'Unknown error'; - } - }, - ); - if (output.status === 'error') { - error = output.error || 'Unknown error'; - } else if (output.result) { - // Result was already forwarded to the user via the streaming callback above - result = output.result; + return; + } + + // Usage exhausted: don't fall back to Kimi — just mark cooldown and skip + if (trigger.reason === 'usage-exhausted') { + markPrimaryCooldown(trigger.reason, trigger.retryAfterMs); + logger.info( + { taskId: task.id, group: context.group.name }, + 'All Claude tokens usage-exhausted, skipping Kimi fallback for scheduled task', + ); + error = 'Claude usage exhausted'; + return; + } + + await runFallbackTaskAttempt(trigger.reason, trigger.retryAfterMs); + }; + + const provider = canFallback ? await getActiveProvider() : 'claude'; + + // Already in usage-exhausted cooldown — skip task instead of running on Kimi + if (provider !== 'claude' && isUsageExhausted()) { + logger.info( + { taskId: task.id, group: context.group.name, provider }, + 'Claude usage exhausted (cooldown active), skipping scheduled task', + ); + error = 'Claude usage exhausted'; + // Fall through to task completion handling below + } else { + + const attempt = await runTaskAttempt(provider); + result = attempt.attemptResult; + error = attempt.attemptError; + + if ( + provider === 'claude' && + attempt.streamedTriggerReason && + !attempt.sawOutput + ) { + await retryClaudeTaskWithRotation(attempt.streamedTriggerReason); + } else if (attempt.output.status === 'error' && provider === 'claude') { + const trigger = attempt.streamedTriggerReason + ? { + shouldFallback: true, + reason: attempt.streamedTriggerReason.reason, + retryAfterMs: attempt.streamedTriggerReason.retryAfterMs, + } + : detectFallbackTrigger(error); + if (trigger.shouldFallback) { + await retryClaudeTaskWithRotation({ + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }); + } + } else if (attempt.output.status === 'error') { + error = attempt.attemptError || 'Unknown error'; } + } // end else (non-exhausted path) + logger.info( { taskId: task.id, diff --git a/src/token-rotation.ts b/src/token-rotation.ts index 2e73dbc..62e3473 100644 --- a/src/token-rotation.ts +++ b/src/token-rotation.ts @@ -142,21 +142,29 @@ export function getCurrentToken(): string | undefined { * Try to rotate to the next available (non-rate-limited) token. * Returns true if a fresh token was found, false if all are exhausted. */ -export function rotateToken(errorMessage?: string): boolean { +export function rotateToken( + errorMessage?: string, + opts?: { ignoreRateLimits?: boolean }, +): boolean { if (tokens.length <= 1) return false; const now = Date.now(); tokens[currentIndex].rateLimitedUntil = computeCooldownUntil(errorMessage); + const ignoreRL = opts?.ignoreRateLimits ?? false; // Find next available token for (let i = 1; i < tokens.length; i++) { const idx = (currentIndex + i) % tokens.length; const state = tokens[idx]; - if (!state.rateLimitedUntil || state.rateLimitedUntil <= now) { + if ( + ignoreRL || + !state.rateLimitedUntil || + state.rateLimitedUntil <= now + ) { state.rateLimitedUntil = null; currentIndex = idx; logger.info( - { tokenIndex: currentIndex, totalTokens: tokens.length }, + { tokenIndex: currentIndex, totalTokens: tokens.length, ignoreRL }, `Rotated to token #${currentIndex + 1}/${tokens.length}`, ); saveState(); diff --git a/src/unified-dashboard.test.ts b/src/unified-dashboard.test.ts new file mode 100644 index 0000000..5600537 --- /dev/null +++ b/src/unified-dashboard.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { ClaudeAccountUsage } from './claude-usage.js'; + +vi.mock('./claude-usage.js', () => ({ + fetchAllClaudeUsage: vi.fn(), + fetchAllClaudeProfiles: vi.fn(), + getClaudeProfile: (index: number) => + index === 0 + ? { email: 'a@example.com', planType: 'max' } + : { email: 'b@example.com', planType: 'pro' }, +})); + +vi.mock('./codex-token-rotation.js', () => ({ + getAllCodexAccounts: () => [], + updateCodexAccountUsage: vi.fn(), +})); + +import { + buildClaudeUsageRows, + mergeClaudeDashboardAccounts, +} from './unified-dashboard.js'; + +describe('unified dashboard Claude usage rows', () => { + it('keeps both Claude accounts visible when one account usage is unavailable', () => { + const rows = buildClaudeUsageRows([ + { + index: 0, + masked: 'tok-a', + isActive: true, + isRateLimited: false, + usage: { + five_hour: { + utilization: 0.4, + resets_at: '2026-03-24T04:00:00+09:00', + }, + seven_day: { + utilization: 0.7, + resets_at: '2026-03-29T04:00:00+09:00', + }, + }, + }, + { + index: 1, + masked: 'tok-b', + isActive: false, + isRateLimited: true, + usage: null, + }, + ]); + + expect(rows).toHaveLength(2); + expect(rows[0]).toMatchObject({ + name: 'Claude1* max', + h5pct: 40, + d7pct: 70, + }); + expect(rows[1]).toMatchObject({ + name: 'Claude2! pro', + h5pct: -1, + d7pct: -1, + }); + }); + + it('preserves the last successful usage per account instead of collapsing to one cache entry', () => { + const cachedAccounts: ClaudeAccountUsage[] = [ + { + index: 0, + masked: 'tok-a', + isActive: false, + isRateLimited: false, + usage: { + five_hour: { + utilization: 0.25, + resets_at: '2026-03-24T04:00:00+09:00', + }, + seven_day: { + utilization: 0.5, + resets_at: '2026-03-29T04:00:00+09:00', + }, + }, + }, + { + index: 1, + masked: 'tok-b', + isActive: true, + isRateLimited: false, + usage: { + five_hour: { + utilization: 0.6, + resets_at: '2026-03-24T06:00:00+09:00', + }, + seven_day: { + utilization: 0.8, + resets_at: '2026-03-30T04:00:00+09:00', + }, + }, + }, + ]; + const liveAccounts: ClaudeAccountUsage[] = [ + { + index: 0, + masked: 'tok-a', + isActive: true, + isRateLimited: false, + usage: null, + }, + { + index: 1, + masked: 'tok-b', + isActive: false, + isRateLimited: true, + usage: { + five_hour: { + utilization: 0.9, + resets_at: '2026-03-24T08:00:00+09:00', + }, + seven_day: { + utilization: 0.95, + resets_at: '2026-03-31T04:00:00+09:00', + }, + }, + }, + ]; + + const merged = mergeClaudeDashboardAccounts(liveAccounts, cachedAccounts); + + expect(merged).toHaveLength(2); + expect(merged[0]).toMatchObject({ + index: 0, + isActive: true, + usage: cachedAccounts[0].usage, + }); + expect(merged[1]).toMatchObject({ + index: 1, + isRateLimited: true, + usage: liveAccounts[1].usage, + }); + }); +}); diff --git a/src/unified-dashboard.ts b/src/unified-dashboard.ts index f26fdac..e388cec 100644 --- a/src/unified-dashboard.ts +++ b/src/unified-dashboard.ts @@ -5,11 +5,9 @@ import path from 'path'; import { STATUS_SHOW_ROOMS, USAGE_DASHBOARD_ENABLED } from './config.js'; import { - fetchClaudeUsage as fetchClaudeUsageApi, fetchAllClaudeUsage, fetchAllClaudeProfiles, getClaudeProfile, - type ClaudeUsageData, type ClaudeAccountUsage, } from './claude-usage.js'; import { @@ -24,7 +22,6 @@ import { renderCategorizedRoomSections, } from './dashboard-render.js'; import { getAllChats, updateRegisteredGroupName } from './db.js'; -import { readEnvFile } from './env.js'; import type { GroupQueue } from './group-queue.js'; import { logger } from './logger.js'; import { @@ -69,11 +66,8 @@ const STATUS_SNAPSHOT_MAX_AGE_MS = 60000; let statusMessageId: string | null = null; let cachedUsageContent = ''; -let cachedClaudeUsageData: ClaudeUsageData | null = null; +let cachedClaudeAccounts: ClaudeAccountUsage[] = []; let usageUpdateInProgress = false; -let usageApiBackoffUntil = 0; -let usageApi429Streak = 0; -let usageApiPollingDisabled = false; let channelMetaCache = new Map(); let channelMetaLastRefresh = 0; @@ -135,22 +129,6 @@ export async function purgeDashboardChannel( } } -function parseRetryAfterMs(retryAfter: string | null): number | null { - if (!retryAfter) return null; - - const seconds = Number(retryAfter); - if (Number.isFinite(seconds) && seconds > 0) { - return seconds * 1000; - } - - const absolute = Date.parse(retryAfter); - if (!Number.isNaN(absolute)) { - return Math.max(0, absolute - Date.now()); - } - - return null; -} - async function refreshChannelMeta( opts: UnifiedDashboardOptions, ): Promise { @@ -366,85 +344,6 @@ function buildStatusContent(): string { return `${header}\n\n${sections}`; } -async function fetchClaudeUsage(): Promise { - if (usageApiPollingDisabled) { - logger.debug('Skipping usage API call (polling disabled for this process)'); - return null; - } - if (Date.now() < usageApiBackoffUntil) { - logger.debug('Skipping usage API call (backoff active)'); - return null; - } - - const apiUsage = await fetchClaudeUsageApi(); - if (apiUsage) { - usageApi429Streak = 0; - return apiUsage; - } - - try { - const envToken = readEnvFile(['CLAUDE_CODE_OAUTH_TOKEN']); - let token = - process.env.CLAUDE_CODE_OAUTH_TOKEN || - envToken.CLAUDE_CODE_OAUTH_TOKEN || - ''; - if (!token) { - const configDir = - process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude'); - const credsPath = path.join(configDir, '.credentials.json'); - if (!fs.existsSync(credsPath)) return null; - const creds = JSON.parse(fs.readFileSync(credsPath, 'utf-8')); - token = creds?.claudeAiOauth?.accessToken || ''; - } - if (!token) return null; - - const res = await fetch('https://api.anthropic.com/api/oauth/usage', { - headers: { - Authorization: `Bearer ${token}`, - 'anthropic-beta': 'oauth-2025-04-20', - }, - }); - if (!res.ok) { - const body = await res.text().catch(() => ''); - if (res.status === 429) { - const retryAfter = res.headers.get('retry-after'); - const retryAfterMs = parseRetryAfterMs(retryAfter); - const backoffMs = Math.max(600_000, retryAfterMs ?? 0); - usageApi429Streak += 1; - usageApiBackoffUntil = Date.now() + backoffMs; - if (usageApi429Streak >= 3) { - usageApiPollingDisabled = true; - } - logger.warn( - { - status: 429, - retryAfter, - retryAfterMs, - backoffMs, - consecutive429s: usageApi429Streak, - pollingDisabled: usageApiPollingDisabled, - body: body.slice(0, 200), - }, - usageApiPollingDisabled - ? 'Usage API rate limited repeatedly (429), disabling usage polling for this process' - : 'Usage API rate limited (429), backing off', - ); - } else { - logger.warn( - { status: res.status, body: body.slice(0, 200) }, - 'Usage API returned non-OK status', - ); - } - return null; - } - usageApi429Streak = 0; - return (await res.json()) as ClaudeUsageData; - } catch (err) { - logger.debug({ err }, 'Usage API fetch failed'); - return null; - } -} - async function fetchCodexUsage( codexHomeOverride?: string, ): Promise { @@ -543,74 +442,46 @@ async function fetchCodexUsage( }); } -async function buildUsageContent(): Promise { - const activeSnapshots = readStatusSnapshots(STATUS_SNAPSHOT_MAX_AGE_MS); - const hasActiveClaudeWork = activeSnapshots.some( - (snapshot) => - snapshot.agentType === 'claude-code' && - snapshot.entries.some( - (entry) => entry.status === 'processing' || entry.status === 'waiting', - ), +type UsageRow = { + name: string; + h5pct: number; + h5reset: string; + d7pct: number; + d7reset: string; +}; + +export function mergeClaudeDashboardAccounts( + liveAccounts: ClaudeAccountUsage[] | null | undefined, + cachedAccounts: ClaudeAccountUsage[], +): ClaudeAccountUsage[] { + if (!liveAccounts) return cachedAccounts; + + const cachedByIndex = new Map( + cachedAccounts.map((account) => [account.index, account]), ); - const shouldFetchClaudeUsage = USAGE_DASHBOARD_ENABLED; - const [liveClaudeAccounts, codexUsage] = await Promise.all([ - shouldFetchClaudeUsage && !hasActiveClaudeWork - ? fetchAllClaudeUsage() - : Promise.resolve(null), - fetchCodexUsage(), - ]); + return liveAccounts.map((account) => ({ + ...account, + usage: account.usage || cachedByIndex.get(account.index)?.usage || null, + })); +} - // Update cache with first account's data for backward compat - const claudeUsageIsCached = - shouldFetchClaudeUsage && !liveClaudeAccounts && !!cachedClaudeUsageData; - if (shouldFetchClaudeUsage && liveClaudeAccounts?.length) { - cachedClaudeUsageData = liveClaudeAccounts[0].usage; - } +export function buildClaudeUsageRows( + claudeAccounts: ClaudeAccountUsage[], +): UsageRow[] { + const isMultiAccount = claudeAccounts.length > 1; - const lines: string[] = ['📊 *사용량*']; - const bar = (pct: number) => { - const filled = Math.round(pct / 10); - return '█'.repeat(filled) + '░'.repeat(10 - filled); - }; - - type UsageRow = { - name: string; - h5pct: number; - h5reset: string; - d7pct: number; - d7reset: string; - }; - const rows: UsageRow[] = []; - - const claudeAccounts = - liveClaudeAccounts || - (cachedClaudeUsageData - ? [ - { - index: 0, - masked: '', - isActive: true, - isRateLimited: false, - usage: cachedClaudeUsageData, - }, - ] - : []); - - for (const account of claudeAccounts) { - const u = account.usage; - if (!u) continue; - const h5 = u.five_hour; - const d7 = u.seven_day; + return claudeAccounts.map((account) => { + const usage = account.usage; + const h5 = usage?.five_hour; + const d7 = usage?.seven_day; const profile = getClaudeProfile(account.index); const planSuffix = profile ? ` ${profile.planType}` : ''; - const label = - claudeAccounts.length > 1 - ? `Claude${account.index + 1}${account.isActive ? '*' : ''}${account.isRateLimited ? '!' : ''}${planSuffix}` - : claudeUsageIsCached - ? `Claude*${planSuffix}` - : `Claude${planSuffix}`; - rows.push({ + const label = isMultiAccount + ? `Claude${account.index + 1}${account.isActive ? '*' : ''}${account.isRateLimited ? '!' : ''}${planSuffix}` + : `Claude${account.isActive ? '*' : ''}${account.isRateLimited ? '!' : ''}${planSuffix}`; + + return { name: label, h5pct: h5 ? h5.utilization > 1 @@ -624,7 +495,38 @@ async function buildUsageContent(): Promise { : Math.round(d7.utilization * 100) : -1, d7reset: d7 ? formatResetRemaining(d7.resets_at) : '', - }); + }; + }); +} + +async function buildUsageContent(): Promise { + const shouldFetchClaudeUsage = USAGE_DASHBOARD_ENABLED; + let liveClaudeAccounts: ClaudeAccountUsage[] | null = null; + + const codexUsagePromise = fetchCodexUsage(); + if (shouldFetchClaudeUsage) { + try { + liveClaudeAccounts = await fetchAllClaudeUsage(); + } catch (err) { + logger.warn({ err }, 'Failed to fetch Claude usage for dashboard'); + } + } + const codexUsage = await codexUsagePromise; + + const lines: string[] = ['📊 *사용량*']; + const bar = (pct: number) => { + const filled = Math.round(pct / 10); + return '█'.repeat(filled) + '░'.repeat(10 - filled); + }; + + const rows: UsageRow[] = []; + + if (shouldFetchClaudeUsage) { + cachedClaudeAccounts = mergeClaudeDashboardAccounts( + liveClaudeAccounts, + cachedClaudeAccounts, + ); + rows.push(...buildClaudeUsageRows(cachedClaudeAccounts)); } const codexAccounts = getAllCodexAccounts(); @@ -712,14 +614,6 @@ async function buildUsageContent(): Promise { lines.push('_조회 불가_'); } - if (shouldFetchClaudeUsage && usageApiPollingDisabled) { - lines.push( - '_* Claude 사용량 조회는 반복된 429로 이번 프로세스에서 일시 중지_', - ); - } - if (claudeUsageIsCached) { - lines.push('_* Claude 사용량은 작업 중일 때는 캐시값 유지_'); - } lines.push(''); lines.push('🖥️ *서버*');