diff --git a/src/agent-runner-environment.test.ts b/src/agent-runner-environment.test.ts new file mode 100644 index 0000000..d358961 --- /dev/null +++ b/src/agent-runner-environment.test.ts @@ -0,0 +1,172 @@ +import fs from 'fs'; +import path from 'path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const { mockReadEnvFile, mockGetActiveCodexAuthPath } = vi.hoisted(() => ({ + mockReadEnvFile: vi.fn<() => Record>(), + mockGetActiveCodexAuthPath: vi.fn<() => string | null>(), +})); + +vi.mock('./config.js', () => ({ + GROUPS_DIR: '/tmp/ejclaw-test-groups', + TIMEZONE: 'Asia/Seoul', +})); + +vi.mock('./db.js', () => ({ + isPairedRoomJid: vi.fn(() => false), +})); + +vi.mock('./env.js', () => ({ + readEnvFile: mockReadEnvFile, +})); + +vi.mock('./codex-token-rotation.js', () => ({ + getActiveCodexAuthPath: mockGetActiveCodexAuthPath, +})); + +vi.mock('./token-rotation.js', () => ({ + getCurrentToken: vi.fn(() => undefined), +})); + +vi.mock('./platform-prompts.js', () => ({ + readPlatformPrompt: vi.fn(() => 'platform prompt'), + readPairedRoomPrompt: vi.fn(() => 'paired room prompt'), +})); + +vi.mock('./group-folder.js', () => ({ + resolveGroupFolderPath: (folder: string) => + `${process.env.EJ_TEST_ROOT}/groups/${folder}`, + resolveGroupIpcPath: (folder: string) => + `${process.env.EJ_TEST_ROOT}/ipc/${folder}`, + resolveGroupSessionsPath: (folder: string) => + `${process.env.EJ_TEST_ROOT}/sessions/${folder}`, + resolveTaskRuntimeIpcPath: (folder: string, taskId: string) => + `${process.env.EJ_TEST_ROOT}/task-ipc/${folder}/${taskId}`, + resolveTaskSessionsPath: (folder: string, taskId: string) => + `${process.env.EJ_TEST_ROOT}/task-sessions/${folder}/${taskId}`, +})); + +vi.mock('os', async () => { + const actual = await vi.importActual('os'); + return { + ...actual, + default: { + ...actual, + homedir: () => process.env.EJ_TEST_HOME || '/tmp', + }, + homedir: () => process.env.EJ_TEST_HOME || '/tmp', + }; +}); + +import { prepareGroupEnvironment } from './agent-runner-environment.js'; +import type { RegisteredGroup } from './types.js'; + +const group: RegisteredGroup = { + name: 'Codex Test Group', + folder: 'codex-test-group', + trigger: '@Codex', + added_at: new Date().toISOString(), + agentType: 'codex', +}; + +describe('prepareGroupEnvironment codex auth handling', () => { + let tempRoot: string; + let previousCwd: string; + let previousOpenAiKey: string | undefined; + let previousCodexOpenAiKey: string | undefined; + + beforeEach(() => { + tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-agent-env-')); + previousCwd = process.cwd(); + process.chdir(tempRoot); + + process.env.EJ_TEST_ROOT = tempRoot; + process.env.EJ_TEST_HOME = path.join(tempRoot, 'home'); + previousOpenAiKey = process.env.OPENAI_API_KEY; + previousCodexOpenAiKey = process.env.CODEX_OPENAI_API_KEY; + delete process.env.OPENAI_API_KEY; + delete process.env.CODEX_OPENAI_API_KEY; + + fs.mkdirSync(process.env.EJ_TEST_HOME, { recursive: true }); + fs.mkdirSync(path.join(process.env.EJ_TEST_HOME, '.codex'), { + recursive: true, + }); + + mockReadEnvFile.mockReset(); + mockGetActiveCodexAuthPath.mockReset(); + }); + + afterEach(() => { + process.chdir(previousCwd); + delete process.env.EJ_TEST_ROOT; + delete process.env.EJ_TEST_HOME; + if (previousOpenAiKey) process.env.OPENAI_API_KEY = previousOpenAiKey; + else delete process.env.OPENAI_API_KEY; + if (previousCodexOpenAiKey) { + process.env.CODEX_OPENAI_API_KEY = previousCodexOpenAiKey; + } else { + delete process.env.CODEX_OPENAI_API_KEY; + } + fs.rmSync(tempRoot, { recursive: true, force: true }); + }); + + it('writes API-key auth when OPENAI_API_KEY is available', () => { + const rotatedAuthPath = path.join(tempRoot, 'rotated-auth.json'); + fs.writeFileSync( + rotatedAuthPath, + JSON.stringify({ auth_mode: 'chatgpt', tokens: { access_token: 'x' } }), + ); + mockGetActiveCodexAuthPath.mockReturnValue(rotatedAuthPath); + mockReadEnvFile.mockReturnValue({ + OPENAI_API_KEY: 'sk-test-api-key', + CODEX_MODEL: 'gpt-5.4', + }); + + prepareGroupEnvironment(group, false, 'dc:test'); + + const authPath = path.join( + tempRoot, + 'sessions', + group.folder, + '.codex', + 'auth.json', + ); + const auth = JSON.parse(fs.readFileSync(authPath, 'utf-8')) as { + auth_mode: string; + OPENAI_API_KEY?: string; + tokens?: unknown; + }; + + expect(auth.auth_mode).toBe('apikey'); + expect(auth.OPENAI_API_KEY).toBe('sk-test-api-key'); + expect(auth.tokens).toBeUndefined(); + }); + + it('falls back to rotated OAuth auth when no API key is configured', () => { + const rotatedAuthPath = path.join(tempRoot, 'rotated-auth.json'); + const rotatedAuth = { + auth_mode: 'chatgpt', + tokens: { + access_token: 'oauth-access', + refresh_token: 'oauth-refresh', + }, + }; + fs.writeFileSync(rotatedAuthPath, JSON.stringify(rotatedAuth)); + mockGetActiveCodexAuthPath.mockReturnValue(rotatedAuthPath); + mockReadEnvFile.mockReturnValue({}); + + prepareGroupEnvironment(group, false, 'dc:test'); + + const authPath = path.join( + tempRoot, + 'sessions', + group.folder, + '.codex', + 'auth.json', + ); + const auth = JSON.parse(fs.readFileSync(authPath, 'utf-8')); + + expect(auth).toEqual(rotatedAuth); + }); +}); diff --git a/src/agent-runner-environment.ts b/src/agent-runner-environment.ts index 34ea8af..aed5f00 100644 --- a/src/agent-runner-environment.ts +++ b/src/agent-runner-environment.ts @@ -20,6 +20,23 @@ import { } from './platform-prompts.js'; import type { AgentType, RegisteredGroup } from './types.js'; +function writeCodexApiKeyAuth( + authPath: string, + openaiKey: string, +): void { + fs.writeFileSync( + authPath, + JSON.stringify( + { + auth_mode: 'apikey', + OPENAI_API_KEY: openaiKey, + }, + null, + 2, + ) + '\n', + ); +} + function syncDirectoryEntries(sources: string[], destination: string): void { for (const source of sources) { if (!fs.existsSync(source)) continue; @@ -185,13 +202,21 @@ function prepareCodexSessionEnvironment(args: { const sessionCodexDir = path.join(args.sessionRootDir, '.codex'); fs.mkdirSync(sessionCodexDir, { recursive: true }); - const rotatedAuthSrc = getActiveCodexAuthPath(); - const authSrc = - rotatedAuthSrc && fs.existsSync(rotatedAuthSrc) - ? rotatedAuthSrc - : path.join(hostCodexDir, 'auth.json'); const authDst = path.join(sessionCodexDir, 'auth.json'); - if (fs.existsSync(authSrc)) fs.copyFileSync(authSrc, authDst); + if (openaiKey) { + writeCodexApiKeyAuth(authDst, openaiKey); + } else { + const rotatedAuthSrc = getActiveCodexAuthPath(); + const authSrc = + rotatedAuthSrc && fs.existsSync(rotatedAuthSrc) + ? rotatedAuthSrc + : path.join(hostCodexDir, 'auth.json'); + if (fs.existsSync(authSrc)) { + fs.copyFileSync(authSrc, authDst); + } else if (fs.existsSync(authDst)) { + fs.unlinkSync(authDst); + } + } for (const file of ['config.toml', 'config.json']) { const src = path.join(hostCodexDir, file); const dst = path.join(sessionCodexDir, file); diff --git a/src/codex-token-rotation.ts b/src/codex-token-rotation.ts index f02932a..8d58501 100644 --- a/src/codex-token-rotation.ts +++ b/src/codex-token-rotation.ts @@ -32,6 +32,11 @@ interface CodexAccount { resetD7At?: string; } +export interface CodexRotationTriggerResult { + shouldRotate: boolean; + reason: string; +} + function parseJwtAuth(idToken: string): { planType: string; expiresAt: string | null; @@ -226,6 +231,53 @@ export function getActiveCodexAuthPath(): string | null { return accounts[currentIndex]?.authPath ?? null; } +export function detectCodexRotationTrigger( + error?: string | null, +): CodexRotationTriggerResult { + if (!error) return { shouldRotate: false, reason: '' }; + + const lower = error.toLowerCase(); + + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('usage limit') || + lower.includes('hit your limit') || + lower.includes('too many requests') || + lower.includes('rate_limit') + ) { + return { shouldRotate: true, reason: '429' }; + } + + if (lower.includes('503') || lower.includes('overloaded')) { + return { shouldRotate: true, reason: 'overloaded' }; + } + + if ( + lower.includes('econnrefused') || + lower.includes('econnreset') || + lower.includes('etimedout') || + lower.includes('enotfound') || + lower.includes('fetch failed') || + lower.includes('network error') + ) { + return { shouldRotate: true, reason: 'network-error' }; + } + + if ( + lower.includes('401') || + lower.includes('authentication_error') || + lower.includes('failed to authenticate') || + lower.includes('oauth token has expired') || + lower.includes('refresh your existing token') || + lower.includes('unauthorized') + ) { + return { shouldRotate: true, reason: 'auth-expired' }; + } + + return { shouldRotate: false, reason: '' }; +} + /** * Try to rotate to the next available Codex account. * Returns true if a fresh account was found. diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 07e1674..6747b48 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -1,4 +1,5 @@ import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; +import { EventEmitter } from 'events'; import fs from 'fs'; import { GroupQueue } from './group-queue.js'; @@ -304,6 +305,54 @@ describe('GroupQueue', () => { expect(processMessages).not.toHaveBeenCalled(); }); + it('signals and terminates active agent processes during shutdown', async () => { + class FakeProcess extends EventEmitter { + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + kill = vi.fn((signal?: NodeJS.Signals) => { + this.signalCode = signal ?? 'SIGTERM'; + return true; + }); + } + + const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder'; + let releaseRun!: (value: boolean) => void; + const blocker = new Promise((resolve) => { + releaseRun = resolve; + }); + + const processMessages = vi.fn(async () => await blocker); + queue.setProcessMessagesFn(processMessages); + + queue.enqueueMessageCheck('group1@g.us', ipcDir); + await vi.advanceTimersByTimeAsync(10); + + const proc = new FakeProcess(); + queue.registerProcess( + 'group1@g.us', + proc as unknown as import('child_process').ChildProcess, + 'proc-1', + ipcDir, + ); + + vi.mocked(fs.writeFileSync).mockClear(); + + const shutdownPromise = queue.shutdown(1_000); + expect(fs.writeFileSync).toHaveBeenCalledWith( + expect.stringContaining('/input/_close'), + '', + ); + + await vi.advanceTimersByTimeAsync(1_000); + await vi.advanceTimersByTimeAsync(2_000); + await shutdownPromise; + + expect(proc.kill).toHaveBeenCalledWith('SIGTERM'); + + releaseRun(true); + await vi.advanceTimersByTimeAsync(10); + }); + // --- Max retries exceeded --- it('stops retrying after MAX_RETRIES and resets', async () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index fcd2d28..1cd9597 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -553,22 +553,132 @@ export class GroupQueue { }); } - async shutdown(_gracePeriodMs: number): Promise { + private isProcessAlive(proc: ChildProcess): boolean { + return proc.exitCode === null && proc.signalCode === null; + } + + private waitForProcessExit(proc: ChildProcess): Promise { + if (!this.isProcessAlive(proc)) { + return Promise.resolve(); + } + + return new Promise((resolve) => { + const handleExit = () => { + proc.off('close', handleExit); + proc.off('exit', handleExit); + resolve(); + }; + + proc.once('close', handleExit); + proc.once('exit', handleExit); + }); + } + + async shutdown(gracePeriodMs: number): Promise { this.shuttingDown = true; - // Count active agent processes but don't kill them — they'll finish on their own - // via idle timeout or agent timeout. - // This prevents reconnection restarts from killing working agents. - const activeProcesses: string[] = []; - for (const [, state] of this.groups) { - if (state.process && !state.process.killed && state.processName) { - activeProcesses.push(state.processName); + const activeProcesses: Array<{ + groupJid: string; + process: ChildProcess; + processName: string; + }> = []; + + for (const [groupJid, state] of this.groups) { + if (state.retryTimer) { + clearTimeout(state.retryTimer); + state.retryTimer = null; + } + state.retryScheduledAt = null; + + if (state.process && state.processName) { + activeProcesses.push({ + groupJid, + process: state.process, + processName: state.processName, + }); + + if (state.active && state.ipcDir && !state.closingStdin) { + this.closeStdin(groupJid, { reason: 'shutdown' }); + } } } + if (activeProcesses.length === 0) { + logger.info('GroupQueue shutdown with no active agent processes'); + return; + } + logger.info( - { activeCount: this.activeCount, detachedProcesses: activeProcesses }, - 'GroupQueue shutting down (agent processes detached, not killed)', + { + activeCount: this.activeCount, + processNames: activeProcesses.map(({ processName }) => processName), + gracePeriodMs, + }, + 'GroupQueue shutting down, waiting for active agent processes to exit', ); + + const graceWaitMs = Math.max(gracePeriodMs, 0); + if (graceWaitMs > 0) { + await Promise.race([ + Promise.all( + activeProcesses.map(({ process }) => this.waitForProcessExit(process)), + ), + new Promise((resolve) => setTimeout(resolve, graceWaitMs)), + ]); + } + + const stillRunning = activeProcesses.filter(({ process }) => + this.isProcessAlive(process), + ); + + if (stillRunning.length === 0) { + logger.info('All active agent processes exited during shutdown'); + return; + } + + logger.warn( + { + processNames: stillRunning.map(({ processName }) => processName), + }, + 'Terminating lingering agent processes during shutdown', + ); + + for (const { process } of stillRunning) { + try { + process.kill('SIGTERM'); + } catch (err) { + logger.warn({ err }, 'Failed to SIGTERM lingering agent process'); + } + } + + await Promise.race([ + Promise.all( + stillRunning.map(({ process }) => this.waitForProcessExit(process)), + ), + new Promise((resolve) => setTimeout(resolve, 2_000)), + ]); + + const stubborn = stillRunning.filter(({ process }) => + this.isProcessAlive(process), + ); + + if (stubborn.length === 0) { + return; + } + + logger.error( + { + processNames: stubborn.map(({ processName }) => processName), + }, + 'Force-killing stubborn agent processes during shutdown', + ); + + for (const { process } of stubborn) { + try { + process.kill('SIGKILL'); + } catch (err) { + logger.warn({ err }, 'Failed to SIGKILL stubborn agent process'); + } + } } } diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 0f39602..8991fc7 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -63,6 +63,20 @@ vi.mock('./token-rotation.js', () => ({ })); vi.mock('./codex-token-rotation.js', () => ({ + detectCodexRotationTrigger: vi.fn((error?: string | null) => { + const lower = (error || '').toLowerCase(); + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('oauth token has expired') || + lower.includes('authentication_error') || + lower.includes('failed to authenticate') || + lower.includes('401') + ) { + return { shouldRotate: true, reason: 'auth-expired' }; + } + return { shouldRotate: false, reason: '' }; + }), rotateCodexToken: vi.fn(() => false), getCodexAccountCount: vi.fn(() => 1), markCodexTokenHealthy: vi.fn(), @@ -73,6 +87,7 @@ vi.mock('./memento-client.js', () => ({ })); import * as agentRunner from './agent-runner.js'; +import * as codexTokenRotation from './codex-token-rotation.js'; import { buildRoomMemoryBriefing } from './memento-client.js'; import { runAgentForGroup } from './message-agent-executor.js'; import * as providerFallback from './provider-fallback.js'; @@ -237,6 +252,61 @@ describe('runAgentForGroup Claude rotation', () => { expect(outputs).toEqual(['회전된 Claude 응답입니다.']); }); + it('rotates to another Claude account when Claude streams an OAuth expiry banner', 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: 'intermediate', + result: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}', + }); + await onOutput?.({ + status: 'success', + phase: 'final', + result: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}', + }); + 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-auth-expired-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('stops after all Claude accounts are usage-exhausted without falling back to Kimi', async () => { const outputs: string[] = []; @@ -338,3 +408,69 @@ describe('runAgentForGroup Claude rotation', () => { ]); }); }); + +describe('runAgentForGroup Codex rotation', () => { + beforeEach(() => { + vi.resetAllMocks(); + vi.mocked(buildRoomMemoryBriefing).mockResolvedValue(undefined); + vi.mocked(providerFallback.getActiveProvider).mockResolvedValue('claude'); + vi.mocked(providerFallback.isFallbackEnabled).mockReturnValue(false); + vi.mocked(providerFallback.hasGroupProviderOverride).mockReturnValue(false); + vi.mocked(codexTokenRotation.getCodexAccountCount).mockReturnValue(2); + vi.mocked(codexTokenRotation.rotateCodexToken).mockReturnValueOnce(true); + }); + + it('retries Codex with a rotated account when OAuth auth expires', async () => { + const codexGroup: RegisteredGroup = { + ...makeGroup(), + folder: 'test-codex', + agentType: 'codex', + }; + const outputs: string[] = []; + + vi.mocked(agentRunner.runAgentProcess) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'error', + result: null, + error: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"OAuth token has expired. Please obtain a new token or refresh your existing token."}}', + }); + return { + status: 'error', + result: null, + error: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"OAuth token has expired. Please obtain a new token or refresh your existing token."}}', + }; + }) + .mockImplementationOnce(async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '새 계정으로 재시도 성공', + newSessionId: 'codex-thread-2', + }); + return { + status: 'success', + result: null, + newSessionId: 'codex-thread-2', + }; + }); + + const result = await runAgentForGroup(makeDeps(), { + group: codexGroup, + prompt: 'hello codex', + chatJid: 'group@test', + runId: 'run-codex-auth-expired', + onOutput: async (output) => { + if (typeof output.result === 'string') outputs.push(output.result); + }, + }); + + expect(result).toBe('success'); + expect(codexTokenRotation.rotateCodexToken).toHaveBeenCalledTimes(1); + expect(codexTokenRotation.markCodexTokenHealthy).toHaveBeenCalledTimes(1); + expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(2); + expect(outputs).toEqual(['새 계정으로 재시도 성공']); + }); +}); diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 93e831a..56c4b8c 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -24,6 +24,7 @@ import { } from './provider-fallback.js'; import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { + detectCodexRotationTrigger, rotateCodexToken, getCodexAccountCount, markCodexTokenHealthy, @@ -71,6 +72,28 @@ function isClaudeUsageExhaustedMessage(text: string): boolean { return looksLikeBanner && hasResetHint && normalized.length <= 160; } +function isClaudeAuthExpiredMessage(text: string): boolean { + const normalized = text.trim().toLowerCase().replace(/\s+/g, ' '); + const looksLikeAuthFailure = normalized.startsWith( + 'failed to authenticate', + ); + const hasExpiredTokenMarker = + normalized.includes('oauth token has expired') || + normalized.includes('authentication_error') || + normalized.includes('obtain a new token') || + normalized.includes('refresh your existing token') || + normalized.includes('invalid authentication credentials'); + const hasUnauthorizedMarker = + normalized.includes('401') || normalized.includes('authentication error'); + const hasTerminatedMarker = normalized.includes('terminated'); + + return ( + looksLikeAuthFailure && + hasUnauthorizedMarker && + (hasExpiredTokenMarker || hasTerminatedMarker) + ); +} + export async function runAgentForGroup( deps: MessageAgentExecutorDeps, args: { @@ -179,21 +202,28 @@ export async function runAgentForGroup( output.status === 'success' && !sawOutput && typeof output.result === 'string' && - isClaudeUsageExhaustedMessage(output.result) + (isClaudeUsageExhaustedMessage(output.result) || + isClaudeAuthExpiredMessage(output.result)) ) { if (!streamedTriggerReason) { + const reason = isClaudeUsageExhaustedMessage(output.result) + ? 'usage-exhausted' + : 'auth-expired'; logger.warn( { chatJid, group: group.name, runId, + reason, resultPreview: output.result.slice(0, 120), }, - 'Detected Claude usage exhaustion banner in successful output', + 'Detected Claude fallback trigger in successful output', ); } streamedTriggerReason = { - reason: 'usage-exhausted', + reason: isClaudeUsageExhaustedMessage(output.result) + ? 'usage-exhausted' + : 'auth-expired', }; return; } @@ -230,12 +260,27 @@ export async function runAgentForGroup( !sawOutput && !streamedTriggerReason ) { - const trigger = detectFallbackTrigger(output.error); - if (trigger.shouldFallback) { - streamedTriggerReason = { - reason: trigger.reason, - retryAfterMs: trigger.retryAfterMs, - }; + if (provider === 'claude') { + const trigger = detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + streamedTriggerReason = { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; + if (canFallback) { + return; + } + } + } else { + const trigger = detectCodexRotationTrigger(output.error); + if (trigger.shouldRotate) { + streamedTriggerReason = { + reason: trigger.reason, + }; + if (getCodexAccountCount() > 1) { + return; + } + } } } await onOutput(output); @@ -371,7 +416,115 @@ export async function runAgentForGroup( }; const shouldRotateClaudeToken = (reason: string): boolean => - reason === '429' || reason === 'usage-exhausted'; + reason === '429' || + reason === 'usage-exhausted' || + reason === 'auth-expired'; + + const retryCodexWithRotation = async ( + initialTrigger: { reason: string }, + rotationMessage?: string, + ): Promise<'success' | 'error'> => { + let trigger = initialTrigger; + let lastRotationMessage = rotationMessage; + + while ( + getCodexAccountCount() > 1 && + rotateCodexToken(lastRotationMessage) + ) { + logger.info( + { chatJid, group: group.name, runId, reason: trigger.reason }, + 'Codex account unhealthy, retrying with rotated account', + ); + + const retryAttempt = await runAttempt('codex'); + + if (retryAttempt.error) { + const errMsg = + retryAttempt.error instanceof Error + ? retryAttempt.error.message + : String(retryAttempt.error); + const retryTrigger = detectCodexRotationTrigger(errMsg); + if (retryTrigger.shouldRotate) { + trigger = { reason: retryTrigger.reason }; + lastRotationMessage = errMsg; + continue; + } + + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: 'codex', + err: retryAttempt.error, + }, + 'Rotated Codex account also threw', + ); + return 'error'; + } + + const retryOutput = retryAttempt.output; + if (!retryOutput) { + logger.error( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + provider: 'codex', + }, + 'Rotated Codex account produced no output object', + ); + return 'error'; + } + + if ( + !retryAttempt.sawOutput && + retryAttempt.streamedTriggerReason && + retryOutput.status !== 'error' + ) { + trigger = { reason: retryAttempt.streamedTriggerReason.reason }; + lastRotationMessage = + typeof retryOutput.result === 'string' + ? retryOutput.result + : undefined; + continue; + } + + if (retryOutput.status === 'error') { + const retryTrigger = retryAttempt.streamedTriggerReason + ? { + shouldRotate: true, + reason: retryAttempt.streamedTriggerReason.reason, + } + : detectCodexRotationTrigger(retryOutput.error); + + if (retryTrigger.shouldRotate) { + trigger = { reason: retryTrigger.reason }; + lastRotationMessage = retryOutput.error ?? undefined; + continue; + } + + logger.error( + { + group: group.name, + chatJid, + runId, + provider: 'codex', + error: retryOutput.error, + }, + 'Rotated Codex account failed', + ); + return 'error'; + } + + markCodexTokenHealthy(); + return 'success'; + } + + return 'error'; + }; const retryClaudeWithRotation = async ( initialTrigger: { @@ -556,6 +709,17 @@ export async function runAgentForGroup( } } + if (!isClaudeCodeAgent) { + const errMsg = + primaryAttempt.error instanceof Error + ? primaryAttempt.error.message + : String(primaryAttempt.error); + const trigger = detectCodexRotationTrigger(errMsg); + if (trigger.shouldRotate && getCodexAccountCount() > 1) { + return retryCodexWithRotation({ reason: trigger.reason }, errMsg); + } + } + logger.error( { chatJid, @@ -638,22 +802,13 @@ export async function runAgentForGroup( } } - // Codex rate-limit rotation (non-Claude agents) if (!isClaudeCodeAgent && getCodexAccountCount() > 1) { - const trigger = detectFallbackTrigger(output.error); - if ( - trigger.shouldFallback && - rotateCodexToken(output.error ?? undefined) - ) { - logger.info( - { chatJid, group: group.name, runId, reason: trigger.reason }, - 'Codex rate-limited, retrying with rotated account', + const trigger = detectCodexRotationTrigger(output.error); + if (trigger.shouldRotate) { + return retryCodexWithRotation( + { reason: trigger.reason }, + output.error ?? undefined, ); - const retryAttempt = await runAttempt('codex'); - if (!retryAttempt.error) { - markCodexTokenHealthy(); - return 'success'; - } } } @@ -670,28 +825,15 @@ export async function runAgentForGroup( return 'error'; } - // Codex may report success but have streamed a rate-limit error. - // Rotate token and retry immediately with the new account. if ( !isClaudeCodeAgent && primaryAttempt.streamedTriggerReason && - getCodexAccountCount() > 1 && - rotateCodexToken(output.error ?? undefined) + getCodexAccountCount() > 1 ) { - logger.info( - { - chatJid, - group: group.name, - runId, - reason: primaryAttempt.streamedTriggerReason.reason, - }, - 'Codex rate-limited (streamed), retrying with rotated account', + return retryCodexWithRotation( + { reason: primaryAttempt.streamedTriggerReason.reason }, + output.error ?? output.result ?? undefined, ); - const retryAttempt = await runAttempt('codex'); - if (!retryAttempt.error) { - markCodexTokenHealthy(); - return 'success'; - } } return 'success'; diff --git a/src/provider-fallback.test.ts b/src/provider-fallback.test.ts index 54a9cc9..5e03f5d 100644 --- a/src/provider-fallback.test.ts +++ b/src/provider-fallback.test.ts @@ -27,6 +27,7 @@ vi.mock('./logger.js', () => ({ import { fetchClaudeUsage } from './claude-usage.js'; import { clearPrimaryCooldown, + detectFallbackTrigger, getActiveProvider, getCooldownInfo, markPrimaryCooldown, @@ -100,4 +101,24 @@ describe('provider fallback usage recovery', () => { await expect(getActiveProvider()).resolves.toBe('claude'); expect(getCooldownInfo()).toEqual({ active: false }); }); + + it('treats terminated 401 auth failures as an auth-expired fallback trigger', () => { + expect( + detectFallbackTrigger('Failed to authenticate. API Error: 401 terminated'), + ).toEqual({ + shouldFallback: true, + reason: 'auth-expired', + }); + }); + + it('treats invalid authentication credentials as an auth-expired fallback trigger', () => { + expect( + detectFallbackTrigger( + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}', + ), + ).toEqual({ + shouldFallback: true, + reason: 'auth-expired', + }); + }); }); diff --git a/src/provider-fallback.ts b/src/provider-fallback.ts index 20343b7..71867b5 100644 --- a/src/provider-fallback.ts +++ b/src/provider-fallback.ts @@ -378,6 +378,19 @@ export function detectFallbackTrigger( return { shouldFallback: true, reason: '429', retryAfterMs }; } + if ( + (lower.includes('failed to authenticate') || + lower.includes('authentication_error')) && + (lower.includes('401') || lower.includes('unauthorized')) && + (lower.includes('oauth token has expired') || + lower.includes('obtain a new token') || + lower.includes('refresh your existing token') || + lower.includes('invalid authentication credentials') || + lower.includes('terminated')) + ) { + return { shouldFallback: true, reason: 'auth-expired' }; + } + // 503 Overloaded if (lower.includes('503') || lower.includes('overloaded')) { return { shouldFallback: true, reason: 'overloaded' }; diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 40f79cb..a0462a7 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -39,6 +39,20 @@ vi.mock('./token-rotation.js', () => ({ })); vi.mock('./codex-token-rotation.js', () => ({ + detectCodexRotationTrigger: vi.fn((error?: string | null) => { + const lower = (error || '').toLowerCase(); + if ( + lower.includes('429') || + lower.includes('rate limit') || + lower.includes('oauth token has expired') || + lower.includes('authentication_error') || + lower.includes('failed to authenticate') || + lower.includes('401') + ) { + return { shouldRotate: true, reason: 'auth-expired' }; + } + return { shouldRotate: false, reason: '' }; + }), rotateCodexToken: vi.fn(() => false), getCodexAccountCount: vi.fn(() => 1), markCodexTokenHealthy: vi.fn(), @@ -77,10 +91,13 @@ describe('task scheduler', () => { _resetSchedulerLoopForTests(); runAgentProcessMock.mockClear(); writeTasksSnapshotMock.mockClear(); + vi.mocked(providerFallback.markPrimaryCooldown).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.markTokenHealthy).mockClear(); + vi.mocked(tokenRotation.rotateToken).mockClear(); vi.mocked(tokenRotation.rotateToken).mockReturnValue(false); vi.useFakeTimers(); }); @@ -372,6 +389,105 @@ Check the run. ); }); + it('suppresses Claude OAuth expiry banners for scheduled tasks and retries with a rotated account', async () => { + const dueAt = new Date(Date.now() - 60_000).toISOString(); + createTask({ + id: 'task-auth-expired-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: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}', + }); + await onOutput?.({ + status: 'success', + result: + 'Failed to authenticate. API Error: 401 {"type":"error","error":{"type":"authentication_error","message":"Invalid authentication credentials"}}', + }); + 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 auth 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 auth response', + ); + }); + it('picks up newly due tasks immediately when nudged', async () => { const enqueueTask = vi.fn(); diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 4db678b..2cb6a80 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -42,6 +42,7 @@ import { markPrimaryCooldown, } from './provider-fallback.js'; import { + detectCodexRotationTrigger, rotateCodexToken, getCodexAccountCount, markCodexTokenHealthy, @@ -91,6 +92,28 @@ function isClaudeUsageExhaustedMessage(text: string): boolean { return looksLikeBanner && hasResetHint && normalized.length <= 160; } +function isClaudeAuthExpiredMessage(text: string): boolean { + const normalized = text.trim().toLowerCase().replace(/\s+/g, ' '); + const looksLikeAuthFailure = normalized.startsWith( + 'failed to authenticate', + ); + const hasExpiredTokenMarker = + normalized.includes('oauth token has expired') || + normalized.includes('authentication_error') || + normalized.includes('obtain a new token') || + normalized.includes('refresh your existing token') || + normalized.includes('invalid authentication credentials'); + const hasUnauthorizedMarker = + normalized.includes('401') || normalized.includes('authentication error'); + const hasTerminatedMarker = normalized.includes('terminated'); + + return ( + looksLikeAuthFailure && + hasUnauthorizedMarker && + (hasExpiredTokenMarker || hasTerminatedMarker) + ); +} + /** * Compute the next run time for a recurring task, anchored to the * task's scheduled time rather than Date.now() to prevent cumulative @@ -347,21 +370,32 @@ async function runTask( !sawOutput && streamedOutput.status === 'success' && typeof streamedOutput.result === 'string' && - isClaudeUsageExhaustedMessage(streamedOutput.result) + (isClaudeUsageExhaustedMessage(streamedOutput.result) || + isClaudeAuthExpiredMessage(streamedOutput.result)) ) { if (!streamedTriggerReason) { + const reason = isClaudeUsageExhaustedMessage( + streamedOutput.result, + ) + ? 'usage-exhausted' + : 'auth-expired'; logger.warn( { taskId: task.id, taskChatJid: task.chat_jid, group: context.group.name, groupFolder: task.group_folder, + reason, resultPreview: streamedOutput.result.slice(0, 120), }, - 'Detected Claude usage exhaustion banner during scheduled task output', + 'Detected Claude fallback trigger during scheduled task output', ); } - streamedTriggerReason = { reason: 'usage-exhausted' }; + streamedTriggerReason = { + reason: isClaudeUsageExhaustedMessage(streamedOutput.result) + ? 'usage-exhausted' + : 'auth-expired', + }; return; } @@ -403,7 +437,9 @@ async function runTask( }; const shouldRotateClaudeToken = (reason: string): boolean => - reason === '429' || reason === 'usage-exhausted'; + reason === '429' || + reason === 'usage-exhausted' || + reason === 'auth-expired'; const runFallbackTaskAttempt = async ( reason: string, @@ -596,25 +632,42 @@ async function runTask( // Try token rotation before suspending if (error) { - const trigger = detectFallbackTrigger(error); - if (trigger.shouldFallback) { - const isCodex = SERVICE_AGENT_TYPE === 'codex'; - const rotated = isCodex - ? getCodexAccountCount() > 1 && rotateCodexToken(error) - : getTokenCount() > 1 && rotateToken(error); - if (rotated) { - logger.info( - { - taskId: task.id, - agent: SERVICE_AGENT_TYPE, - reason: trigger.reason, - }, - 'Task rate-limited, rotated token — will retry on next schedule', - ); - if (isCodex) markCodexTokenHealthy(); - else markTokenHealthy(); - // Clear the error so suspension doesn't trigger - error = null; + const isCodex = SERVICE_AGENT_TYPE === 'codex'; + if (isCodex) { + const trigger = detectCodexRotationTrigger(error); + if (trigger.shouldRotate) { + const rotated = getCodexAccountCount() > 1 && rotateCodexToken(error); + if (rotated) { + logger.info( + { + taskId: task.id, + agent: SERVICE_AGENT_TYPE, + reason: trigger.reason, + }, + 'Task rate-limited, rotated token — will retry on next schedule', + ); + markCodexTokenHealthy(); + // Clear the error so suspension doesn't trigger + error = null; + } + } + } else { + const trigger = detectFallbackTrigger(error); + if (trigger.shouldFallback) { + const rotated = getTokenCount() > 1 && rotateToken(error); + if (rotated) { + logger.info( + { + taskId: task.id, + agent: SERVICE_AGENT_TYPE, + reason: trigger.reason, + }, + 'Task rate-limited, rotated token — will retry on next schedule', + ); + markTokenHealthy(); + // Clear the error so suspension doesn't trigger + error = null; + } } } }