diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 504aaa7..b81337c 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -13,11 +13,6 @@ import { getAllTasks } from './db.js'; import { GroupQueue } from './group-queue.js'; import { logger } from './logger.js'; import { buildRoomMemoryBriefing } from './memento-client.js'; -import { - isClaudeAuthError, - isClaudeAuthExpiredMessage, - isClaudeUsageExhaustedMessage, -} from './agent-error-detection.js'; import { detectFallbackTrigger, getActiveProvider, @@ -30,6 +25,10 @@ import { } from './provider-fallback.js'; import { runClaudeRotationLoop } from './provider-retry.js'; import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; +import { + evaluateStreamedOutput, + type StreamedOutputState, +} from './streamed-output-evaluator.js'; import { detectCodexRotationTrigger, rotateCodexToken, @@ -130,14 +129,10 @@ export async function runAgentForGroup( }; }> => { const persistSessionIds = provider === 'claude'; - let sawOutput = false; - let sawSuccessNullResultWithoutOutput = false; - let streamedTriggerReason: - | { - reason: string; - retryAfterMs?: number; - } - | undefined; + let streamedState: StreamedOutputState = { + sawOutput: false, + sawSuccessNullResultWithoutOutput: false, + }; const wrappedOnOutput = onOutput ? async (output: AgentOutput) => { @@ -151,92 +146,69 @@ export async function runAgentForGroup( ) { resetSessionRequested = true; } + const evaluation = evaluateStreamedOutput(output, streamedState, { + agentType: isClaudeCodeAgent ? 'claude' : 'codex', + provider, + suppressClaudeAuthErrorOutput: provider === 'claude', + trackSuccessNullResult: true, + shortCircuitTriggeredErrors: + provider === 'claude' + ? canFallback || canRotateToken + : getCodexAccountCount() > 1, + }); + streamedState = evaluation.state; + if ( - isClaudeCodeAgent && - provider === 'claude' && - output.status === 'success' && - !sawOutput && + evaluation.newTrigger && typeof output.result === 'string' && - (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 fallback trigger in successful output', - ); - } - streamedTriggerReason = { - reason: isClaudeUsageExhaustedMessage(output.result) - ? 'usage-exhausted' - : 'auth-expired', - }; - return; - } - // 401 auth errors — suppress from chat, log only - if ( - provider === 'claude' && - output.status === 'success' && - !sawOutput && - typeof output.result === 'string' && - isClaudeAuthError(output.result) + output.status === 'success' ) { logger.warn( { chatJid, group: group.name, runId, + reason: evaluation.newTrigger.reason, resultPreview: output.result.slice(0, 120), }, + 'Detected Claude fallback trigger in successful output', + ); + } else if ( + evaluation.newTrigger && + typeof output.error === 'string' + ) { + logger.warn( + { + chatJid, + group: group.name, + runId, + reason: evaluation.newTrigger.reason, + errorPreview: output.error.slice(0, 120), + }, + provider === 'claude' + ? 'Detected Claude fallback trigger in streamed error output' + : 'Detected Codex rotation trigger in streamed error output', + ); + } + + if (evaluation.suppressedAuthError) { + logger.warn( + { + chatJid, + group: group.name, + runId, + resultPreview: + typeof output.result === 'string' + ? output.result.slice(0, 120) + : undefined, + }, 'Suppressed Claude 401 auth error from chat output', ); return; } - if (output.result !== null && output.result !== undefined) { - sawOutput = true; - } else if ( - provider === 'claude' && - output.status === 'success' && - !sawOutput - ) { - sawSuccessNullResultWithoutOutput = true; - } - if ( - output.status === 'error' && - !sawOutput && - !streamedTriggerReason - ) { - if (provider === 'claude') { - const trigger = detectFallbackTrigger(output.error); - if (trigger.shouldFallback) { - streamedTriggerReason = { - reason: trigger.reason, - retryAfterMs: trigger.retryAfterMs, - }; - if (canFallback || canRotateToken) { - return; - } - } - } else { - const trigger = detectCodexRotationTrigger(output.error); - if (trigger.shouldRotate) { - streamedTriggerReason = { - reason: trigger.reason, - }; - if (getCodexAccountCount() > 1) { - return; - } - } - } + + if (!evaluation.shouldForwardOutput) { + return; } await onOutput(output); } @@ -280,7 +252,9 @@ export async function runAgentForGroup( (proc, processName, ipcDir) => deps.queue.registerProcess(chatJid, proc, processName, ipcDir), wrappedOnOutput, - provider === 'claude' ? undefined : getFallbackEnvOverrides(), + isClaudeCodeAgent && provider !== 'claude' + ? getFallbackEnvOverrides() + : undefined, ); if (persistSessionIds && output.newSessionId) { @@ -295,23 +269,25 @@ export async function runAgentForGroup( runId, provider, status: output.status, - sawOutput, + sawOutput: streamedState.sawOutput, }, `Provider response completed (provider: ${provider})`, ); return { output, - sawOutput, - sawSuccessNullResultWithoutOutput, - streamedTriggerReason, + sawOutput: streamedState.sawOutput, + sawSuccessNullResultWithoutOutput: + streamedState.sawSuccessNullResultWithoutOutput, + streamedTriggerReason: streamedState.streamedTriggerReason, }; } catch (error) { return { error, - sawOutput, - sawSuccessNullResultWithoutOutput, - streamedTriggerReason, + sawOutput: streamedState.sawOutput, + sawSuccessNullResultWithoutOutput: + streamedState.sawSuccessNullResultWithoutOutput, + streamedTriggerReason: streamedState.streamedTriggerReason, }; } }; diff --git a/src/streamed-output-evaluator.ts b/src/streamed-output-evaluator.ts new file mode 100644 index 0000000..31a8b57 --- /dev/null +++ b/src/streamed-output-evaluator.ts @@ -0,0 +1,131 @@ +import { + isClaudeAuthError, + isClaudeAuthExpiredMessage, + isClaudeUsageExhaustedMessage, +} from './agent-error-detection.js'; +import type { AgentOutput } from './agent-runner.js'; +import { detectCodexRotationTrigger } from './codex-token-rotation.js'; +import { detectFallbackTrigger } from './provider-fallback.js'; + +export interface StreamedTriggerReason { + reason: string; + retryAfterMs?: number; +} + +export interface StreamedOutputState { + sawOutput: boolean; + sawSuccessNullResultWithoutOutput: boolean; + streamedTriggerReason?: StreamedTriggerReason; +} + +export interface EvaluateStreamedOutputOptions { + agentType: 'claude' | 'codex'; + provider: string; + suppressClaudeAuthErrorOutput?: boolean; + trackSuccessNullResult?: boolean; + shortCircuitTriggeredErrors?: boolean; +} + +export interface EvaluateStreamedOutputResult { + state: StreamedOutputState; + shouldForwardOutput: boolean; + newTrigger?: StreamedTriggerReason; + suppressedAuthError?: boolean; +} + +export function evaluateStreamedOutput( + output: AgentOutput, + state: StreamedOutputState, + options: EvaluateStreamedOutputOptions, +): EvaluateStreamedOutputResult { + const nextState: StreamedOutputState = { ...state }; + const isPrimaryClaude = + options.agentType === 'claude' && options.provider === 'claude'; + const isPrimaryCodex = + options.agentType === 'codex' && options.provider === 'codex'; + + if ( + isPrimaryClaude && + output.status === 'success' && + !state.sawOutput && + typeof output.result === 'string' + ) { + const triggerReason = isClaudeUsageExhaustedMessage(output.result) + ? 'usage-exhausted' + : isClaudeAuthExpiredMessage(output.result) + ? 'auth-expired' + : undefined; + + if (triggerReason) { + const newTrigger = nextState.streamedTriggerReason + ? undefined + : { reason: triggerReason }; + nextState.streamedTriggerReason = + nextState.streamedTriggerReason ?? newTrigger; + return { + state: nextState, + shouldForwardOutput: false, + newTrigger, + }; + } + + if ( + options.suppressClaudeAuthErrorOutput && + isClaudeAuthError(output.result) + ) { + return { + state: nextState, + shouldForwardOutput: false, + suppressedAuthError: true, + }; + } + } + + if (output.result !== null && output.result !== undefined) { + nextState.sawOutput = true; + } else if ( + options.trackSuccessNullResult && + isPrimaryClaude && + output.status === 'success' && + !state.sawOutput + ) { + nextState.sawSuccessNullResultWithoutOutput = true; + } + + if ( + output.status === 'error' && + !nextState.sawOutput && + !nextState.streamedTriggerReason + ) { + let newTrigger: StreamedTriggerReason | undefined; + + if (isPrimaryClaude) { + const trigger = detectFallbackTrigger(output.error); + if (trigger.shouldFallback) { + newTrigger = { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; + } + } else if (isPrimaryCodex) { + const trigger = detectCodexRotationTrigger(output.error); + if (trigger.shouldRotate) { + newTrigger = { reason: trigger.reason }; + } + } + + if (newTrigger) { + nextState.streamedTriggerReason = newTrigger; + return { + state: nextState, + shouldForwardOutput: !options.shortCircuitTriggeredErrors, + newTrigger, + }; + } + } + + return { + state: nextState, + shouldForwardOutput: true, + }; +} diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index a0462a7..d8f772d 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -72,6 +72,7 @@ vi.mock('./agent-runner.js', async () => { import { _initTestDatabase, createTask, getTaskById } from './db.js'; import * as providerFallback from './provider-fallback.js'; +import * as codexTokenRotation from './codex-token-rotation.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'; @@ -99,6 +100,10 @@ describe('task scheduler', () => { vi.mocked(tokenRotation.markTokenHealthy).mockClear(); vi.mocked(tokenRotation.rotateToken).mockClear(); vi.mocked(tokenRotation.rotateToken).mockReturnValue(false); + vi.mocked(codexTokenRotation.rotateCodexToken).mockClear(); + vi.mocked(codexTokenRotation.rotateCodexToken).mockReturnValue(false); + vi.mocked(codexTokenRotation.getCodexAccountCount).mockReturnValue(1); + vi.mocked(codexTokenRotation.markCodexTokenHealthy).mockClear(); vi.useFakeTimers(); }); @@ -488,6 +493,101 @@ Check the run. ); }); + it('retries Codex scheduled tasks with a rotated account on streamed auth expiry', async () => { + const dueAt = new Date(Date.now() - 60_000).toISOString(); + createTask({ + id: 'task-codex-auth-expired', + group_folder: 'shared-group', + chat_jid: 'shared@g.us', + agent_type: 'codex', + prompt: 'codex 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(codexTokenRotation.getCodexAccountCount).mockReturnValue(2); + vi.mocked(codexTokenRotation.rotateCodexToken).mockReturnValueOnce(true); + + (runAgentProcessMock as any) + .mockImplementationOnce( + async ( + _group: unknown, + _input: unknown, + _onProcess: unknown, + onOutput?: (output: Record) => Promise, + ) => { + await onOutput?.({ + status: 'error', + 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."}}', + result: null, + }); + return { + status: 'error', + 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."}}', + result: null, + }; + }, + ) + .mockImplementationOnce( + async ( + _group: unknown, + _input: unknown, + _onProcess: unknown, + onOutput?: (output: Record) => Promise, + ) => { + await onOutput?.({ + status: 'success', + result: 'rotated codex 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: 'codex', + registeredGroups: () => ({ + 'shared@g.us': { + name: 'Shared', + folder: 'shared-group', + trigger: '@Codex', + added_at: '2026-02-22T00:00:00.000Z', + agentType: 'codex', + }, + }), + getSessions: () => ({}), + queue: { enqueueTask } as any, + onProcess: () => {}, + sendMessage, + }); + + await vi.advanceTimersByTimeAsync(10); + + expect(runAgentProcessMock).toHaveBeenCalledTimes(2); + expect(codexTokenRotation.rotateCodexToken).toHaveBeenCalledTimes(1); + expect(codexTokenRotation.markCodexTokenHealthy).toHaveBeenCalledTimes(1); + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(sendMessage).toHaveBeenCalledWith( + 'shared@g.us', + 'rotated codex scheduled task 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 63d6efe..51aff7a 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -32,10 +32,6 @@ import { } from './group-folder.js'; import { logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; -import { - isClaudeAuthExpiredMessage, - isClaudeUsageExhaustedMessage, -} from './agent-error-detection.js'; import { detectFallbackTrigger, getActiveProvider, @@ -63,6 +59,10 @@ import { formatSuspensionNotice, suspendTask, } from './task-suspension.js'; +import { + evaluateStreamedOutput, + type StreamedOutputState, +} from './streamed-output-evaluator.js'; import { getTaskQueueJid, getTaskRuntimeTaskId, @@ -296,15 +296,12 @@ async function runTask( attemptResult: string | null; attemptError: string | null; }> => { - let sawOutput = false; + let streamedState: StreamedOutputState = { + sawOutput: false, + sawSuccessNullResultWithoutOutput: false, + }; let attemptResult: string | null = null; let attemptError: string | null = null; - let streamedTriggerReason: - | { - reason: string; - retryAfterMs?: number; - } - | undefined; const output = await runAgentProcess( context.group, @@ -330,62 +327,67 @@ async function runTask( if (streamedOutput.phase === 'progress') { return; } + const evaluation = evaluateStreamedOutput(streamedOutput, streamedState, { + agentType: isClaudeAgent ? 'claude' : 'codex', + provider, + shortCircuitTriggeredErrors: true, + }); + streamedState = evaluation.state; if ( - isClaudeAgent && - provider === 'claude' && - !sawOutput && - streamedOutput.status === 'success' && + evaluation.newTrigger && typeof streamedOutput.result === 'string' && - (isClaudeUsageExhaustedMessage(streamedOutput.result) || - isClaudeAuthExpiredMessage(streamedOutput.result)) + streamedOutput.status === 'success' ) { - 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 fallback trigger during scheduled task output', - ); + logger.warn( + { + taskId: task.id, + taskChatJid: task.chat_jid, + group: context.group.name, + groupFolder: task.group_folder, + reason: evaluation.newTrigger.reason, + resultPreview: streamedOutput.result.slice(0, 120), + }, + 'Detected Claude fallback trigger during scheduled task output', + ); + } else if ( + evaluation.newTrigger && + typeof streamedOutput.error === 'string' + ) { + logger.warn( + { + taskId: task.id, + taskChatJid: task.chat_jid, + group: context.group.name, + groupFolder: task.group_folder, + reason: evaluation.newTrigger.reason, + errorPreview: streamedOutput.error.slice(0, 120), + }, + provider === 'claude' + ? 'Detected Claude fallback trigger during scheduled task error output' + : 'Detected Codex rotation trigger during scheduled task error output', + ); + } + + if (!evaluation.shouldForwardOutput) { + if (streamedOutput.status === 'error') { + attemptError = streamedOutput.error || 'Unknown error'; } - streamedTriggerReason = { - reason: isClaudeUsageExhaustedMessage(streamedOutput.result) - ? 'usage-exhausted' - : 'auth-expired', - }; 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(), + isClaudeAgent && provider !== 'claude' + ? getFallbackEnvOverrides() + : undefined, ); if (output.status === 'error' && !attemptError) { @@ -396,8 +398,8 @@ async function runTask( return { output, - sawOutput, - streamedTriggerReason, + sawOutput: streamedState.sawOutput, + streamedTriggerReason: streamedState.streamedTriggerReason, attemptResult, attemptError, }; @@ -482,10 +484,78 @@ async function runTask( } }; - const provider = canFallback ? await getActiveProvider() : 'claude'; + const retryCodexTaskWithRotation = async ( + initialTrigger: { reason: string }, + rotationMessage?: string, + ): Promise => { + let trigger = initialTrigger; + let lastRotationMessage = rotationMessage; + + while ( + getCodexAccountCount() > 1 && + rotateCodexToken(lastRotationMessage) + ) { + logger.info( + { + taskId: task.id, + group: context.group.name, + groupFolder: task.group_folder, + reason: trigger.reason, + }, + 'Codex account unhealthy, retrying scheduled task with rotated account', + ); + + const retryAttempt = await runTaskAttempt('codex'); + result = retryAttempt.attemptResult; + error = retryAttempt.attemptError; + + if ( + !retryAttempt.sawOutput && + retryAttempt.streamedTriggerReason && + retryAttempt.output.status !== 'error' + ) { + trigger = { reason: retryAttempt.streamedTriggerReason.reason }; + lastRotationMessage = + typeof retryAttempt.output.result === 'string' + ? retryAttempt.output.result + : undefined; + continue; + } + + if (retryAttempt.output.status === 'error') { + const retryTrigger = retryAttempt.streamedTriggerReason + ? { + shouldRotate: true, + reason: retryAttempt.streamedTriggerReason.reason, + } + : detectCodexRotationTrigger( + retryAttempt.attemptError || retryAttempt.output.error, + ); + + if (retryTrigger.shouldRotate) { + trigger = { reason: retryTrigger.reason }; + lastRotationMessage = + retryAttempt.attemptError || retryAttempt.output.error || undefined; + continue; + } + return; + } + + markCodexTokenHealthy(); + error = null; + return; + } + }; + + const provider = + context.taskAgentType === 'codex' + ? 'codex' + : canFallback + ? await getActiveProvider() + : 'claude'; // Already in usage-exhausted cooldown — skip task instead of running on Kimi - if (provider !== 'claude' && isUsageExhausted()) { + if (isClaudeAgent && provider !== 'claude' && isUsageExhausted()) { logger.info( { taskId: task.id, group: context.group.name, provider }, 'Claude usage exhausted (cooldown active), skipping scheduled task', @@ -503,6 +573,17 @@ async function runTask( !attempt.sawOutput ) { await retryClaudeTaskWithRotation(attempt.streamedTriggerReason); + } else if ( + provider === 'codex' && + attempt.streamedTriggerReason && + !attempt.sawOutput + ) { + await retryCodexTaskWithRotation( + { reason: attempt.streamedTriggerReason.reason }, + typeof attempt.output.error === 'string' + ? attempt.output.error + : undefined, + ); } else if (attempt.output.status === 'error' && provider === 'claude') { const trigger = attempt.streamedTriggerReason ? { @@ -517,6 +598,19 @@ async function runTask( retryAfterMs: trigger.retryAfterMs, }); } + } else if (attempt.output.status === 'error' && provider === 'codex') { + const trigger = attempt.streamedTriggerReason + ? { + shouldRotate: true, + reason: attempt.streamedTriggerReason.reason, + } + : detectCodexRotationTrigger(error); + if (trigger.shouldRotate) { + await retryCodexTaskWithRotation( + { reason: trigger.reason }, + error || undefined, + ); + } } else if (attempt.output.status === 'error') { error = attempt.attemptError || 'Unknown error'; }