diff --git a/src/agent-attempt-retry.ts b/src/agent-attempt-retry.ts new file mode 100644 index 0000000..49b5491 --- /dev/null +++ b/src/agent-attempt-retry.ts @@ -0,0 +1,164 @@ +import { + classifyRotationTrigger, + type AgentTriggerReason, + type CodexRotationReason, + isCodexRotationReason, +} from './agent-error-detection.js'; +import { detectCodexRotationTrigger } from './codex-token-rotation.js'; +import { getErrorMessage } from './utils.js'; + +export interface AttemptStreamedTrigger { + reason: AgentTriggerReason; + retryAfterMs?: number; +} + +export interface AttemptRetryState { + sawOutput: boolean; + retryableSessionFailureDetected?: boolean; + streamedTriggerReason?: AttemptStreamedTrigger; + error?: unknown; + outputError?: string | null; +} + +export function isRetryableClaudeSessionFailureAttempt(args: { + attempt: AttemptRetryState; + isClaudeCodeAgent: boolean; + provider: 'claude' | 'codex'; + shouldRetryFreshSessionOnAgentFailure: (args: { + result: null; + error: string; + }) => boolean; +}): boolean { + if ( + !args.isClaudeCodeAgent || + args.provider !== 'claude' || + args.attempt.sawOutput + ) { + return false; + } + + if (args.attempt.retryableSessionFailureDetected === true) { + return true; + } + + if (args.attempt.error == null) { + return false; + } + + return args.shouldRetryFreshSessionOnAgentFailure({ + result: null, + error: getErrorMessage(args.attempt.error), + }); +} + +export function resolveClaudeRetryTrigger(args: { + canRetryClaudeCredentials: boolean; + provider: 'claude' | 'codex'; + attempt: Pick; + fallbackMessage?: string | null; +}): { reason: AgentTriggerReason; retryAfterMs?: number } | null { + if ( + !( + args.canRetryClaudeCredentials && + args.provider === 'claude' && + !args.attempt.sawOutput + ) + ) { + return null; + } + + if (args.attempt.streamedTriggerReason) { + return { + reason: args.attempt.streamedTriggerReason.reason, + retryAfterMs: args.attempt.streamedTriggerReason.retryAfterMs, + }; + } + + const trigger = classifyRotationTrigger(args.fallbackMessage); + if (!trigger.shouldRetry) { + return null; + } + + return { + reason: trigger.reason, + retryAfterMs: trigger.retryAfterMs, + }; +} + +export function resolveCodexRetryTrigger(args: { + canRetryCodex: boolean; + attempt: Pick; + rotationMessage?: string | null; +}): { reason: CodexRotationReason } | null { + if (!args.canRetryCodex) { + return null; + } + + if (args.attempt.streamedTriggerReason) { + if (!isCodexRotationReason(args.attempt.streamedTriggerReason.reason)) { + return null; + } + return { + reason: args.attempt.streamedTriggerReason.reason, + }; + } + + const trigger = detectCodexRotationTrigger(args.rotationMessage); + if (!trigger.shouldRotate) { + return null; + } + + return { reason: trigger.reason }; +} + +export type AttemptRetryAction = + | { + kind: 'claude'; + trigger: { reason: AgentTriggerReason; retryAfterMs?: number }; + rotationMessage?: string; + } + | { + kind: 'codex'; + trigger: { reason: CodexRotationReason }; + rotationMessage?: string; + } + | { kind: 'none' }; + +export function resolveAttemptRetryAction(args: { + provider: 'claude' | 'codex'; + canRetryClaudeCredentials: boolean; + canRetryCodex: boolean; + attempt: Pick; + rotationMessage?: string | null; +}): AttemptRetryAction { + const normalizedRotationMessage = args.rotationMessage ?? undefined; + + const claudeTrigger = resolveClaudeRetryTrigger({ + canRetryClaudeCredentials: args.canRetryClaudeCredentials, + provider: args.provider, + attempt: args.attempt, + fallbackMessage: normalizedRotationMessage, + }); + if (claudeTrigger) { + return { + kind: 'claude', + trigger: claudeTrigger, + rotationMessage: normalizedRotationMessage, + }; + } + + const codexTrigger = resolveCodexRetryTrigger({ + canRetryCodex: args.canRetryCodex, + attempt: args.attempt, + rotationMessage: normalizedRotationMessage, + }); + if (codexTrigger) { + return { + kind: 'codex', + trigger: codexTrigger, + rotationMessage: normalizedRotationMessage, + }; + } + + return { kind: 'none' }; +} diff --git a/src/message-agent-executor-rules.test.ts b/src/message-agent-executor-rules.test.ts index 315461e..bb14e43 100644 --- a/src/message-agent-executor-rules.test.ts +++ b/src/message-agent-executor-rules.test.ts @@ -23,6 +23,7 @@ import { classifyRotationTrigger } from './agent-error-detection.js'; import { detectCodexRotationTrigger } from './codex-token-rotation.js'; import { isRetryableClaudeSessionFailureAttempt, + resolveAttemptRetryAction, resolveClaudeRetryTrigger, resolveCodexRetryTrigger, resolvePairedFollowUpQueueAction, @@ -88,6 +89,47 @@ describe('message-agent-executor-rules', () => { expect(detectCodexRotationTrigger).not.toHaveBeenCalled(); }); + it('resolves a shared Claude retry action from fallback output text', () => { + vi.mocked(classifyRotationTrigger).mockReturnValue({ + shouldRetry: true, + reason: '429', + retryAfterMs: 30000, + }); + + expect( + resolveAttemptRetryAction({ + provider: 'claude', + canRetryClaudeCredentials: true, + canRetryCodex: false, + attempt: { sawOutput: false }, + rotationMessage: '429 rate limit', + }), + ).toEqual({ + kind: 'claude', + trigger: { reason: '429', retryAfterMs: 30000 }, + rotationMessage: '429 rate limit', + }); + }); + + it('resolves a shared Codex retry action from streamed state', () => { + expect( + resolveAttemptRetryAction({ + provider: 'codex', + canRetryClaudeCredentials: false, + canRetryCodex: true, + attempt: { + sawOutput: false, + streamedTriggerReason: { reason: 'auth-expired' }, + }, + rotationMessage: 'oauth token expired', + }), + ).toEqual({ + kind: 'codex', + trigger: { reason: 'auth-expired' }, + rotationMessage: 'oauth token expired', + }); + }); + it('detects retryable Claude session failures from either flag or classifier', () => { expect( isRetryableClaudeSessionFailureAttempt({ diff --git a/src/message-agent-executor-rules.ts b/src/message-agent-executor-rules.ts index de2ca9b..daef4f8 100644 --- a/src/message-agent-executor-rules.ts +++ b/src/message-agent-executor-rules.ts @@ -1,116 +1,14 @@ -import { - classifyRotationTrigger, - type AgentTriggerReason, - type CodexRotationReason, - isCodexRotationReason, -} from './agent-error-detection.js'; -import { detectCodexRotationTrigger } from './codex-token-rotation.js'; import { resolveNextTurnAction } from './message-runtime-rules.js'; import type { PairedRoomRole, PairedTaskStatus } from './types.js'; -import { getErrorMessage } from './utils.js'; - -export interface ExecutorStreamedTrigger { - reason: AgentTriggerReason; - retryAfterMs?: number; -} - -export interface ExecutorAttemptState { - sawOutput: boolean; - retryableSessionFailureDetected?: boolean; - streamedTriggerReason?: ExecutorStreamedTrigger; - error?: unknown; -} - -export function isRetryableClaudeSessionFailureAttempt(args: { - attempt: ExecutorAttemptState; - isClaudeCodeAgent: boolean; - provider: 'claude' | 'codex'; - shouldRetryFreshSessionOnAgentFailure: (args: { - result: null; - error: string; - }) => boolean; -}): boolean { - if ( - !args.isClaudeCodeAgent || - args.provider !== 'claude' || - args.attempt.sawOutput - ) { - return false; - } - - if (args.attempt.retryableSessionFailureDetected === true) { - return true; - } - - if (args.attempt.error == null) { - return false; - } - - return args.shouldRetryFreshSessionOnAgentFailure({ - result: null, - error: getErrorMessage(args.attempt.error), - }); -} - -export function resolveClaudeRetryTrigger(args: { - canRetryClaudeCredentials: boolean; - provider: 'claude' | 'codex'; - attempt: Pick; - fallbackMessage?: string | null; -}): { reason: AgentTriggerReason; retryAfterMs?: number } | null { - if ( - !( - args.canRetryClaudeCredentials && - args.provider === 'claude' && - !args.attempt.sawOutput - ) - ) { - return null; - } - - if (args.attempt.streamedTriggerReason) { - return { - reason: args.attempt.streamedTriggerReason.reason, - retryAfterMs: args.attempt.streamedTriggerReason.retryAfterMs, - }; - } - - const trigger = classifyRotationTrigger(args.fallbackMessage); - if (!trigger.shouldRetry) { - return null; - } - - return { - reason: trigger.reason, - retryAfterMs: trigger.retryAfterMs, - }; -} - -export function resolveCodexRetryTrigger(args: { - canRetryCodex: boolean; - attempt: Pick; - rotationMessage?: string | null; -}): { reason: CodexRotationReason } | null { - if (!args.canRetryCodex) { - return null; - } - - if (args.attempt.streamedTriggerReason) { - if (!isCodexRotationReason(args.attempt.streamedTriggerReason.reason)) { - return null; - } - return { - reason: args.attempt.streamedTriggerReason.reason, - }; - } - - const trigger = detectCodexRotationTrigger(args.rotationMessage); - if (!trigger.shouldRotate) { - return null; - } - - return { reason: trigger.reason }; -} +export { + isRetryableClaudeSessionFailureAttempt, + resolveAttemptRetryAction, + resolveClaudeRetryTrigger, + resolveCodexRetryTrigger, + type AttemptRetryAction, + type AttemptRetryState as ExecutorAttemptState, + type AttemptStreamedTrigger as ExecutorStreamedTrigger, +} from './agent-attempt-retry.js'; export type PairedFollowUpQueueAction = | 'generic' diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 003a9e2..37db074 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -5,6 +5,10 @@ import { getErrorMessage } from './utils.js'; import { getAgentOutputText } from './agent-output.js'; import { createEvaluatedOutputHandler } from './agent-attempt.js'; +import { + isRetryableClaudeSessionFailureAttempt, + resolveAttemptRetryAction, +} from './agent-attempt-retry.js'; import { AgentOutput, runAgentProcess, @@ -31,9 +35,6 @@ import { import { resolveCodexFallbackHandoff } from './paired-turn-fallback.js'; import { resolveExecutionTarget } from './message-runtime-rules.js'; import { - isRetryableClaudeSessionFailureAttempt, - resolveClaudeRetryTrigger, - resolveCodexRetryTrigger, resolvePairedFollowUpQueueAction, } from './message-agent-executor-rules.js'; import { buildRoomRoleContext } from './room-role-context.js'; @@ -683,22 +684,23 @@ export async function runAgentForGroup( attempt: AgentAttempt, rotationMessage?: string | null, ): Promise<'success' | 'error' | null> => { - const trigger = resolveClaudeRetryTrigger({ - canRetryClaudeCredentials, + const retryAction = resolveAttemptRetryAction({ provider, + canRetryClaudeCredentials, + canRetryCodex: false, attempt, - fallbackMessage: rotationMessage, + rotationMessage, }); - if (!trigger) { + if (retryAction.kind !== 'claude') { return null; } const result = await retryClaudeWithRotation( - trigger, - rotationMessage ?? undefined, + retryAction.trigger, + retryAction.rotationMessage, ); if (result === 'error') { - return maybeHandoffAfterError(trigger.reason, attempt); + return maybeHandoffAfterError(retryAction.trigger.reason, attempt); } pairedExecutionStatus = 'succeeded'; @@ -709,18 +711,20 @@ export async function runAgentForGroup( attempt: AgentAttempt, rotationMessage?: string | null, ): Promise<'success' | 'error' | null> => { - const trigger = resolveCodexRetryTrigger({ + const retryAction = resolveAttemptRetryAction({ + provider, + canRetryClaudeCredentials: false, canRetryCodex: !isClaudeCodeAgent && getCodexAccountCount() > 1, attempt, rotationMessage, }); - if (!trigger) { + if (retryAction.kind !== 'codex') { return null; } const result = await retryCodexWithRotation( - { reason: trigger.reason }, - rotationMessage ?? undefined, + retryAction.trigger, + retryAction.rotationMessage, ); if (result === 'success') { pairedExecutionStatus = 'succeeded'; diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index b6c29fe..7f7311a 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -8,6 +8,7 @@ import { buildArbiterPromptForTask, buildFinalizePendingPrompt, buildOwnerPendingPrompt, + buildPairedTurnPrompt, buildReviewerPendingPrompt, } from './message-runtime-prompts.js'; import { @@ -47,6 +48,26 @@ export type BotOnlyPairedFollowUpAction = nextRole: 'owner' | 'reviewer' | 'arbiter'; }; +export type QueuedTurnDispatch = { + formatted: string; + botOnlyFollowUpAction: BotOnlyPairedFollowUpAction; + isBotOnlyPairedFollowUp: boolean; + loopCursorKey: string; + endSeq: number | null; +}; + +export function isBotOnlyPairedRoomTurn( + chatJid: string, + messages: NewMessage[], +): boolean { + return ( + hasReviewerLease(chatJid) && + messages.every( + (message) => message.is_from_me === true || !!message.is_bot_message, + ) + ); +} + export function resolveLastDeliveredBotRole( messages: NewMessage[], ): PairedRoomRole | null { @@ -384,6 +405,54 @@ export async function executeBotOnlyPairedFollowUpAction(args: { return true; } +export function buildQueuedTurnDispatch(args: { + chatJid: string; + timezone: string; + loopPendingTask: PairedTask | null | undefined; + rawPendingMessages: NewMessage[]; + messagesToSend: NewMessage[]; + labeledMessagesToSend: NewMessage[]; + formatMessages: (messages: NewMessage[], timezone: string) => string; +}): QueuedTurnDispatch { + const loopCursorKey = resolveCursorKey( + args.chatJid, + args.loopPendingTask?.status, + ); + const formatted = args.loopPendingTask + ? buildPairedTurnPrompt({ + taskId: args.loopPendingTask.id, + chatJid: args.chatJid, + timezone: args.timezone, + missedMessages: args.messagesToSend, + labeledFallbackMessages: args.labeledMessagesToSend, + turnOutputs: getPairedTurnOutputs(args.loopPendingTask.id), + }) + : args.formatMessages(args.labeledMessagesToSend, args.timezone); + const isBotOnlyPairedFollowUp = isBotOnlyPairedRoomTurn( + args.chatJid, + args.messagesToSend, + ); + const pendingCursorSource = + args.rawPendingMessages.length > 0 + ? args.rawPendingMessages[args.rawPendingMessages.length - 1] + : args.messagesToSend[args.messagesToSend.length - 1]; + const botOnlyFollowUpAction = resolveBotOnlyPairedFollowUpAction({ + chatJid: args.chatJid, + task: args.loopPendingTask, + isBotOnlyPairedFollowUp, + lastDeliveredMessages: args.labeledMessagesToSend, + pendingCursorSource, + }); + + return { + formatted, + botOnlyFollowUpAction, + isBotOnlyPairedFollowUp, + loopCursorKey, + endSeq: args.messagesToSend[args.messagesToSend.length - 1]?.seq ?? null, + }; +} + export function shouldSkipGenericFollowUpAfterDeliveryRetry(args: { chatJid: string; deliveryRole: PairedRoomRole; diff --git a/src/message-runtime.ts b/src/message-runtime.ts index fde7752..32a8880 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -43,11 +43,11 @@ import { } from './message-runtime-rules.js'; import { runAgentForGroup } from './message-agent-executor.js'; import { + buildQueuedTurnDispatch, buildPendingPairedTurn, executeBotOnlyPairedFollowUpAction, executePendingPairedTurn, - resolveBotOnlyPairedFollowUpAction, - resolveLastDeliveredBotRole, + isBotOnlyPairedRoomTurn, shouldSkipGenericFollowUpAfterDeliveryRetry, } from './message-runtime-flow.js'; import { MessageTurnController } from './message-turn-controller.js'; @@ -58,6 +58,7 @@ import { buildPairedTurnPrompt, buildReviewerPendingPrompt, } from './message-runtime-prompts.js'; +import { transitionPairedTaskStatus } from './paired-execution-context-shared.js'; import { extractSessionCommand, handleSessionCommand, @@ -207,15 +208,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }); }; - const isBotOnlyPairedRoomTurn = ( - chatJid: string, - messages: NewMessage[], - ): boolean => - hasReviewerLease(chatJid) && - messages.every( - (message) => message.is_from_me === true || !!message.is_bot_message, - ); - const enqueueScopedGroupMessageCheck = ( chatJid: string, groupFolder: string, @@ -937,10 +929,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (hasReviewerLease(chatJid)) { const task = getLatestOpenPairedTaskForChat(chatJid); if (task) { - updatePairedTask(task.id, { - status: 'completed', - completion_reason: 'stopped', - updated_at: new Date().toISOString(), + const now = new Date().toISOString(); + transitionPairedTaskStatus({ + taskId: task.id, + currentStatus: task.status, + nextStatus: 'completed', + updatedAt: now, + patch: { + completion_reason: 'stopped', + }, }); } } @@ -1223,30 +1220,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { chatJid, messagesToSend, ); - const formatted = loopPendingTask - ? buildPairedTurnPrompt({ - taskId: loopPendingTask.id, - chatJid, - timezone: deps.timezone, - missedMessages: messagesToSend, - labeledFallbackMessages: labeledMessagesToSend, - turnOutputs: getPairedTurnOutputs(loopPendingTask.id), - }) - : formatMessages(labeledMessagesToSend, deps.timezone); - const isBotOnlyPairedFollowUp = isBotOnlyPairedRoomTurn( - chatJid, - messagesToSend, - ); - const pendingCursorSource = - rawPendingMessages.length > 0 - ? rawPendingMessages[rawPendingMessages.length - 1] - : messagesToSend[messagesToSend.length - 1]; - const botOnlyFollowUpAction = resolveBotOnlyPairedFollowUpAction({ - chatJid, - task: loopPendingTask, + const { + formatted, + botOnlyFollowUpAction, isBotOnlyPairedFollowUp, - lastDeliveredMessages: labeledMessagesToSend, - pendingCursorSource, + loopCursorKey: dispatchCursorKey, + endSeq, + } = buildQueuedTurnDispatch({ + chatJid, + timezone: deps.timezone, + loopPendingTask, + rawPendingMessages, + messagesToSend, + labeledMessagesToSend, + formatMessages, }); if ( await executeBotOnlyPairedFollowUpAction({ @@ -1271,14 +1258,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } if (deps.queue.sendMessage(chatJid, formatted)) { - const endSeq = messagesToSend[messagesToSend.length - 1]?.seq; if (endSeq != null) { advanceLastAgentCursor( deps.getLastAgentTimestamps(), deps.saveState, chatJid, endSeq, - loopCursorKey, + dispatchCursorKey, ); } logger.debug( diff --git a/src/paired-execution-context-arbiter.ts b/src/paired-execution-context-arbiter.ts index de46e5b..04c33f1 100644 --- a/src/paired-execution-context-arbiter.ts +++ b/src/paired-execution-context-arbiter.ts @@ -1,7 +1,9 @@ import { ARBITER_DEADLOCK_THRESHOLD } from './config.js'; -import { updatePairedTask } from './db.js'; import { logger } from './logger.js'; -import { classifyArbiterVerdict } from './paired-execution-context-shared.js'; +import { + classifyArbiterVerdict, + transitionPairedTaskStatus, +} from './paired-execution-context-shared.js'; import type { PairedTask } from './types.js'; export function handleFailedArbiterExecution(args: { @@ -15,7 +17,12 @@ export function handleFailedArbiterExecution(args: { ? 'arbiter_requested' : task.status; if (fallbackStatus !== task.status) { - updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: fallbackStatus, + updatedAt: now, + }); logger.warn( { taskId, @@ -45,11 +52,15 @@ export function handleArbiterCompletion(args: { case 'proceed': case 'revise': case 'reset': - updatePairedTask(taskId, { - status: 'active', - round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), - arbiter_verdict: arbiterVerdict, - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: 'in_arbitration', + nextStatus: 'active', + updatedAt: now, + patch: { + round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), + arbiter_verdict: arbiterVerdict, + }, }); logger.info( { taskId, arbiterVerdict }, @@ -57,20 +68,28 @@ export function handleArbiterCompletion(args: { ); return; case 'escalate': - updatePairedTask(taskId, { - status: 'completed', - arbiter_verdict: 'escalate', - completion_reason: 'arbiter_escalated', - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: 'in_arbitration', + nextStatus: 'completed', + updatedAt: now, + patch: { + arbiter_verdict: 'escalate', + completion_reason: 'arbiter_escalated', + }, }); logger.info({ taskId }, 'Arbiter escalated to user — task completed'); return; default: - updatePairedTask(taskId, { - status: 'active', - round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), - arbiter_verdict: 'unknown', - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: 'in_arbitration', + nextStatus: 'active', + updatedAt: now, + patch: { + round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), + arbiter_verdict: 'unknown', + }, }); logger.warn( { taskId, summary: summary?.slice(0, 200) }, diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index 186b3f4..a7dca12 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -14,6 +14,7 @@ import { hasCodeChangesSinceRef, requestArbiterOrEscalate, resolveCanonicalSourceRef, + transitionPairedTaskStatus, } from './paired-execution-context-shared.js'; import type { PairedTask } from './types.js'; @@ -24,7 +25,12 @@ export function handleFailedOwnerExecution(args: { const { task, taskId } = args; if (task.status !== 'active') { const now = new Date().toISOString(); - updatePairedTask(taskId, { status: 'active', updated_at: now }); + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'active', + updatedAt: now, + }); logger.info( { taskId, role: 'owner', previousStatus: task.status }, 'Reset task to active after failed execution', @@ -44,6 +50,7 @@ function handleOwnerFinalizeCompletion(args: { if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Owner blocked during finalize — requesting arbiter', escalateLogMessage: 'Owner blocked during finalize — escalating to user', @@ -60,6 +67,7 @@ function handleOwnerFinalizeCompletion(args: { if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Owner finalize loop detected — requesting arbiter', escalateLogMessage: 'Owner finalize loop detected — escalating to user', @@ -71,7 +79,12 @@ function handleOwnerFinalizeCompletion(args: { }); return true; } - updatePairedTask(taskId, { status: 'active', updated_at: now }); + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'active', + updatedAt: now, + }); logger.info( { taskId, @@ -92,6 +105,7 @@ function handleOwnerFinalizeCompletion(args: { if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Owner finalize DONE loop detected — requesting arbiter', @@ -116,10 +130,14 @@ function handleOwnerFinalizeCompletion(args: { return false; } - updatePairedTask(taskId, { - status: 'completed', - completion_reason: 'done', - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'completed', + updatedAt: now, + patch: { + completion_reason: 'done', + }, }); logger.info( { taskId, hasNewChanges, summary: summary?.slice(0, 100) }, @@ -161,6 +179,7 @@ export function handleOwnerCompletion(args: { if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Owner blocked/needs_context — requesting arbiter', escalateLogMessage: 'Owner blocked/needs_context — escalating to user', diff --git a/src/paired-execution-context-reviewer.ts b/src/paired-execution-context-reviewer.ts index 2ebfe19..b3dc496 100644 --- a/src/paired-execution-context-reviewer.ts +++ b/src/paired-execution-context-reviewer.ts @@ -5,6 +5,7 @@ import { classifyVerdict, requestArbiterOrEscalate, resolveCanonicalSourceRef, + transitionPairedTaskStatus, } from './paired-execution-context-shared.js'; import type { PairedTask } from './types.js'; @@ -29,11 +30,15 @@ export function handleFailedReviewerExecution(args: { verdict === 'done' && ownerWs?.workspace_dir ? resolveCanonicalSourceRef(ownerWs.workspace_dir) : task.source_ref; - updatePairedTask(taskId, { - status: verdict === 'done' ? 'merge_ready' : 'completed', - ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}), - ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}), - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: verdict === 'done' ? 'merge_ready' : 'completed', + updatedAt: now, + patch: { + ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}), + ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}), + }, }); logger.info( { @@ -53,7 +58,12 @@ export function handleFailedReviewerExecution(args: { ? 'review_ready' : task.status; if (fallbackStatus !== task.status) { - updatePairedTask(taskId, { status: fallbackStatus, updated_at: now }); + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: fallbackStatus, + updatedAt: now, + }); logger.warn( { taskId, @@ -81,10 +91,14 @@ export function handleReviewerCompletion(args: { const approvedSourceRef = ownerWs?.workspace_dir ? resolveCanonicalSourceRef(ownerWs.workspace_dir) : task.source_ref; - updatePairedTask(taskId, { - status: 'merge_ready', - source_ref: approvedSourceRef, - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'merge_ready', + updatedAt: now, + patch: { + source_ref: approvedSourceRef, + }, }); logger.info( { @@ -102,6 +116,7 @@ export function handleReviewerCompletion(args: { case 'needs_context': requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Reviewer blocked/needs_context — requesting arbiter before escalating', @@ -116,6 +131,7 @@ export function handleReviewerCompletion(args: { if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { requestArbiterOrEscalate({ taskId, + currentStatus: task.status, now, arbiterLogMessage: 'Deadlock detected — requesting arbiter intervention', @@ -129,7 +145,12 @@ export function handleReviewerCompletion(args: { }); return; } - updatePairedTask(taskId, { status: 'active', updated_at: now }); + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'active', + updatedAt: now, + }); logger.info( { taskId, verdict }, 'Reviewer has feedback, task set back to active for owner', diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts index a2e7a2f..59a1fd8 100644 --- a/src/paired-execution-context-shared.ts +++ b/src/paired-execution-context-shared.ts @@ -3,6 +3,7 @@ import { execFileSync } from 'child_process'; import { isArbiterEnabled } from './config.js'; import { updatePairedTask } from './db.js'; import { logger } from './logger.js'; +import type { PairedTaskStatus } from './types.js'; export type Verdict = | 'done' @@ -93,29 +94,108 @@ export function hasCodeChangesSinceRef( } } +const ALLOWED_PAIRED_STATUS_TRANSITIONS: Record< + PairedTaskStatus, + ReadonlySet +> = { + active: new Set(['review_ready', 'arbiter_requested', 'completed']), + review_ready: new Set([ + 'active', + 'in_review', + 'arbiter_requested', + 'completed', + ]), + in_review: new Set([ + 'active', + 'review_ready', + 'merge_ready', + 'arbiter_requested', + 'completed', + ]), + merge_ready: new Set(['active', 'arbiter_requested', 'completed']), + completed: new Set(), + arbiter_requested: new Set(['in_arbitration', 'completed']), + in_arbitration: new Set(['active', 'arbiter_requested', 'completed']), +}; + +export function assertPairedTaskStatusTransition(args: { + currentStatus: PairedTaskStatus; + nextStatus: PairedTaskStatus; +}): void { + const { currentStatus, nextStatus } = args; + if (currentStatus === nextStatus) { + return; + } + + if (ALLOWED_PAIRED_STATUS_TRANSITIONS[currentStatus].has(nextStatus)) { + return; + } + + throw new Error( + `Invalid paired task status transition: ${currentStatus} -> ${nextStatus}`, + ); +} + +export function transitionPairedTaskStatus(args: { + taskId: string; + currentStatus: PairedTaskStatus; + nextStatus: PairedTaskStatus; + updatedAt: string; + patch?: Omit< + Parameters[1], + 'status' | 'updated_at' + >; +}): void { + assertPairedTaskStatusTransition({ + currentStatus: args.currentStatus, + nextStatus: args.nextStatus, + }); + + updatePairedTask(args.taskId, { + ...args.patch, + status: args.nextStatus, + updated_at: args.updatedAt, + }); +} + export function requestArbiterOrEscalate(args: { taskId: string; + currentStatus: PairedTaskStatus; now: string; arbiterLogMessage: string; escalateLogMessage: string; logContext?: Record; }): void { - const { taskId, now, arbiterLogMessage, escalateLogMessage, logContext } = - args; + const { + taskId, + currentStatus, + now, + arbiterLogMessage, + escalateLogMessage, + logContext, + } = args; if (isArbiterEnabled()) { - updatePairedTask(taskId, { - status: 'arbiter_requested', - arbiter_requested_at: now, - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus, + nextStatus: 'arbiter_requested', + updatedAt: now, + patch: { + arbiter_requested_at: now, + }, }); logger.info(logContext ?? { taskId }, arbiterLogMessage); return; } - updatePairedTask(taskId, { - status: 'completed', - completion_reason: 'escalated', - updated_at: now, + transitionPairedTaskStatus({ + taskId, + currentStatus, + nextStatus: 'completed', + updatedAt: now, + patch: { + completion_reason: 'escalated', + }, }); logger.info(logContext ?? { taskId }, escalateLogMessage); } diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 38d8094..ff07c62 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -38,6 +38,7 @@ import { import { resolveCanonicalSourceRef, requestArbiterOrEscalate, + transitionPairedTaskStatus, } from './paired-execution-context-shared.js'; import { markPairedTaskReviewReady, @@ -226,11 +227,22 @@ export function preparePairedExecutionContext(args: { latestTask.status === 'review_ready' || latestTask.status === 'in_review'; if (hasHuman || needsStatusReset) { - updatePairedTask(latestTask.id, { - ...(hasHuman ? { round_trip_count: 0 } : {}), - ...(needsStatusReset ? { status: 'active' as const } : {}), - updated_at: now, - }); + if (needsStatusReset) { + transitionPairedTaskStatus({ + taskId: latestTask.id, + currentStatus: latestTask.status, + nextStatus: 'active', + updatedAt: now, + patch: { + ...(hasHuman ? { round_trip_count: 0 } : {}), + }, + }); + } else { + updatePairedTask(latestTask.id, { + ...(hasHuman ? { round_trip_count: 0 } : {}), + updated_at: now, + }); + } } // Use a stable per-channel worktree (not per-task) so the Claude SDK // session persists across tasks. Different channels still get isolation. @@ -253,9 +265,11 @@ export function preparePairedExecutionContext(args: { blockMessage = reviewerWorkspace.blockMessage; const refreshedTask = getPairedTaskById(latestTask.id) ?? latestTask; if (workspace && refreshedTask.status === 'review_ready') { - updatePairedTask(latestTask.id, { - status: 'in_review', - updated_at: now, + transitionPairedTaskStatus({ + taskId: latestTask.id, + currentStatus: refreshedTask.status, + nextStatus: 'in_review', + updatedAt: now, }); } } else if (roomRoleContext.role === 'arbiter') { @@ -265,9 +279,11 @@ export function preparePairedExecutionContext(args: { blockMessage = reviewerWorkspace.blockMessage; const refreshedTask = getPairedTaskById(latestTask.id) ?? latestTask; if (workspace && refreshedTask.status === 'arbiter_requested') { - updatePairedTask(latestTask.id, { - status: 'in_arbitration', - updated_at: now, + transitionPairedTaskStatus({ + taskId: latestTask.id, + currentStatus: refreshedTask.status, + nextStatus: 'in_arbitration', + updatedAt: now, }); } } diff --git a/src/paired-workspace-manager.ts b/src/paired-workspace-manager.ts index 3bfa486..327cfe8 100644 --- a/src/paired-workspace-manager.ts +++ b/src/paired-workspace-manager.ts @@ -13,6 +13,7 @@ import { } from './db.js'; import { resolvePairedTaskWorkspacePath } from './group-folder.js'; import { logger } from './logger.js'; +import { transitionPairedTaskStatus } from './paired-execution-context-shared.js'; import type { PairedTask, PairedWorkspace } from './types.js'; import { ensureWorkspaceDependenciesInstalled } from './workspace-package-manager.js'; @@ -798,9 +799,15 @@ export function markPairedTaskReviewReady(taskId: string): { ); const now = new Date().toISOString(); - updatePairedTask(taskId, { - status: 'review_ready', - updated_at: now, + const task = getPairedTaskById(taskId); + if (!task) { + return null; + } + transitionPairedTaskStatus({ + taskId, + currentStatus: task.status, + nextStatus: 'review_ready', + updatedAt: now, }); return { ownerWorkspace, reviewerWorkspace }; diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 2e0507c..a79bcc1 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -3,6 +3,7 @@ import { CronExpressionParser } from 'cron-parser'; import fs from 'fs'; import { getAgentOutputText } from './agent-output.js'; import { createEvaluatedOutputHandler } from './agent-attempt.js'; +import { resolveAttemptRetryAction } from './agent-attempt-retry.js'; import { getErrorMessage } from './utils.js'; import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js'; @@ -42,7 +43,6 @@ import { classifyRotationTrigger, type AgentTriggerReason, type CodexRotationReason, - isCodexRotationReason, } from './agent-error-detection.js'; import { getTokenCount, @@ -512,65 +512,34 @@ async function runTask( } }; - const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude'; + const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude'; - { - const attempt = await runTaskAttempt(provider); - result = attempt.attemptResult; - error = attempt.attemptError; + { + const attempt = await runTaskAttempt(provider); + result = attempt.attemptResult; + error = attempt.attemptError; - if ( - provider === 'claude' && - attempt.streamedTriggerReason && - !attempt.sawOutput - ) { - await retryClaudeTaskWithRotation(attempt.streamedTriggerReason); - } else if ( - provider === 'codex' && - attempt.streamedTriggerReason && - !attempt.sawOutput && - isCodexRotationReason(attempt.streamedTriggerReason.reason) - ) { - 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 - ? { - shouldRetry: true, - reason: attempt.streamedTriggerReason.reason, - retryAfterMs: attempt.streamedTriggerReason.retryAfterMs, - } - : classifyRotationTrigger(error); - if (trigger.shouldRetry) { - await retryClaudeTaskWithRotation({ - reason: trigger.reason, - retryAfterMs: trigger.retryAfterMs, - }); - } - } else if (attempt.output.status === 'error' && provider === 'codex') { - const trigger = - attempt.streamedTriggerReason && - isCodexRotationReason(attempt.streamedTriggerReason.reason) - ? { - shouldRotate: true as const, - reason: attempt.streamedTriggerReason.reason, - } - : detectCodexRotationTrigger(error); - if (trigger.shouldRotate) { - await retryCodexTaskWithRotation( - { reason: trigger.reason }, - error || undefined, + const retryAction = resolveAttemptRetryAction({ + provider, + canRetryClaudeCredentials: provider === 'claude' && getTokenCount() > 0, + canRetryCodex: provider === 'codex' && getCodexAccountCount() > 1, + attempt, + rotationMessage: error, + }); + + if (retryAction.kind === 'claude') { + await retryClaudeTaskWithRotation( + retryAction.trigger, + retryAction.rotationMessage, ); + } else if (retryAction.kind === 'codex') { + await retryCodexTaskWithRotation( + retryAction.trigger, + retryAction.rotationMessage, + ); + } else if (attempt.output.status === 'error') { + error = attempt.attemptError || 'Unknown error'; } - } else if (attempt.output.status === 'error') { - error = attempt.attemptError || 'Unknown error'; - } } // end else (non-exhausted path) log.info(