From 10e10b499ca624e959628cb83ba9b9a957959521 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Tue, 7 Apr 2026 05:12:22 +0900 Subject: [PATCH] Complete paired-room SSOT cleanup and logger singleton --- src/agent-attempt-orchestration.ts | 182 +++++++++++++ src/codex-token-rotation.test.ts | 21 ++ src/codex-token-rotation.ts | 7 +- src/logger.test.ts | 33 +++ src/logger.ts | 46 +++- src/message-agent-executor.ts | 82 ++---- src/message-runtime-flow.ts | 32 +-- src/message-runtime.test.ts | 118 -------- src/message-runtime.ts | 363 +++++++++++++------------ src/paired-execution-context-owner.ts | 12 +- src/paired-execution-context-shared.ts | 11 + src/paired-execution-context.test.ts | 1 - src/paired-execution-context.ts | 20 +- src/paired-workspace-manager.test.ts | 25 ++ src/paired-workspace-manager.ts | 11 +- src/task-scheduler.ts | 86 +++--- src/token-rotation.test.ts | 40 +++ src/token-rotation.ts | 31 ++- 18 files changed, 648 insertions(+), 473 deletions(-) create mode 100644 src/agent-attempt-orchestration.ts create mode 100644 src/logger.test.ts diff --git a/src/agent-attempt-orchestration.ts b/src/agent-attempt-orchestration.ts new file mode 100644 index 0000000..acf56b6 --- /dev/null +++ b/src/agent-attempt-orchestration.ts @@ -0,0 +1,182 @@ +import type { AgentOutput } from './agent-runner.js'; +import { + resolveAttemptRetryAction, + type AttemptRetryAction, + type AttemptRetryState, +} from './agent-attempt-retry.js'; +import type { + AgentTriggerReason, + CodexRotationReason, +} from './agent-error-detection.js'; +import { + runClaudeRotationLoop, + runCodexRotationLoop, +} from './provider-retry.js'; + +type RotationLogContext = Record; + +type ClaudeRetryAttemptState = Pick< + AttemptRetryState, + 'sawOutput' | 'streamedTriggerReason' +> & { + output?: Pick; + error?: unknown; + sawSuccessNullResultWithoutOutput?: boolean; +}; + +type CodexRetryAttemptState = Pick< + AttemptRetryState, + 'sawOutput' | 'streamedTriggerReason' +> & { + output?: Pick; + error?: unknown; +}; + +export type ExecutedAttemptRetryAction = + | { kind: 'none' } + | { + kind: 'claude'; + trigger: { reason: AgentTriggerReason; retryAfterMs?: number }; + rotationMessage?: string; + result: 'success' | 'error'; + } + | { + kind: 'codex'; + trigger: { reason: CodexRotationReason }; + rotationMessage?: string; + result: 'success' | 'error'; + }; + +export async function runClaudeAttemptWithRotation< + TAttempt extends ClaudeRetryAttemptState, +>(args: { + initialTrigger: { + reason: AgentTriggerReason; + retryAfterMs?: number; + }; + runAttempt: () => Promise; + logContext: RotationLogContext; + rotationMessage?: string; + afterAttempt?: (attempt: TAttempt) => Promise | void; + onSuccess?: (outcome: { sawOutput: boolean }) => Promise | void; +}): Promise<'success' | 'error'> { + const outcome = await runClaudeRotationLoop( + args.initialTrigger, + async () => { + const attempt = await args.runAttempt(); + await args.afterAttempt?.(attempt); + return { + output: attempt.output, + thrownError: attempt.error, + sawOutput: attempt.sawOutput, + sawSuccessNullResult: attempt.sawSuccessNullResultWithoutOutput, + streamedTriggerReason: attempt.streamedTriggerReason, + }; + }, + args.logContext, + args.rotationMessage, + ); + + if (outcome.type === 'success') { + await args.onSuccess?.({ sawOutput: outcome.sawOutput }); + return 'success'; + } + + return 'error'; +} + +export async function runCodexAttemptWithRotation< + TAttempt extends CodexRetryAttemptState, +>(args: { + initialTrigger: { reason: CodexRotationReason }; + runAttempt: () => Promise; + logContext: RotationLogContext; + rotationMessage?: string; + afterAttempt?: (attempt: TAttempt) => Promise | void; +}): Promise<'success' | 'error'> { + const outcome = await runCodexRotationLoop( + args.initialTrigger, + async () => { + const attempt = await args.runAttempt(); + await args.afterAttempt?.(attempt); + return { + output: attempt.output, + thrownError: attempt.error, + sawOutput: attempt.sawOutput, + streamedTriggerReason: attempt.streamedTriggerReason, + }; + }, + args.logContext, + args.rotationMessage, + ); + + return outcome.type === 'success' ? 'success' : 'error'; +} + +export async function executeAttemptRetryAction(args: { + provider: 'claude' | 'codex'; + canRetryClaudeCredentials: boolean; + canRetryCodex: boolean; + attempt: Pick; + rotationMessage?: string | null; + runClaude: ( + trigger: { reason: AgentTriggerReason; retryAfterMs?: number }, + rotationMessage?: string, + ) => Promise<'success' | 'error'>; + runCodex: ( + trigger: { reason: CodexRotationReason }, + rotationMessage?: string, + ) => Promise<'success' | 'error'>; +}): Promise { + const retryAction = resolveAttemptRetryAction({ + provider: args.provider, + canRetryClaudeCredentials: args.canRetryClaudeCredentials, + canRetryCodex: args.canRetryCodex, + attempt: args.attempt, + rotationMessage: args.rotationMessage, + }); + + return executeResolvedAttemptRetryAction({ + retryAction, + runClaude: args.runClaude, + runCodex: args.runCodex, + }); +} + +async function executeResolvedAttemptRetryAction(args: { + retryAction: AttemptRetryAction; + runClaude: ( + trigger: { reason: AgentTriggerReason; retryAfterMs?: number }, + rotationMessage?: string, + ) => Promise<'success' | 'error'>; + runCodex: ( + trigger: { reason: CodexRotationReason }, + rotationMessage?: string, + ) => Promise<'success' | 'error'>; +}): Promise { + if (args.retryAction.kind === 'claude') { + return { + kind: 'claude', + trigger: args.retryAction.trigger, + rotationMessage: args.retryAction.rotationMessage, + result: await args.runClaude( + args.retryAction.trigger, + args.retryAction.rotationMessage, + ), + }; + } + + if (args.retryAction.kind === 'codex') { + return { + kind: 'codex', + trigger: args.retryAction.trigger, + rotationMessage: args.retryAction.rotationMessage, + result: await args.runCodex( + args.retryAction.trigger, + args.retryAction.rotationMessage, + ), + }; + } + + return { kind: 'none' }; +} diff --git a/src/codex-token-rotation.test.ts b/src/codex-token-rotation.test.ts index bdb89d5..8762a33 100644 --- a/src/codex-token-rotation.test.ts +++ b/src/codex-token-rotation.test.ts @@ -121,4 +121,25 @@ describe('codex-token-rotation d7 ≥ 100% auto-skip', () => { expect(active).toBeDefined(); expect(active!.index).toBe(1); // fallback picks next non-rate-limited }); + + it('warns when Codex rotation state cannot be persisted', async () => { + const mod = await import('./codex-token-rotation.js'); + const utils = await import('./utils.js'); + const { logger } = await import('./logger.js'); + + vi.mocked(utils.writeJsonFile).mockImplementation(() => { + throw new Error('disk full'); + }); + + mod.initCodexTokenRotation(); + expect(mod.rotateCodexToken('rate limit')).toBe(true); + + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + stateFile: '/tmp/ejclaw-codex-rot-data/codex-rotation-state.json', + err: expect.any(Error), + }), + 'Failed to persist Codex rotation state', + ); + }); }); diff --git a/src/codex-token-rotation.ts b/src/codex-token-rotation.ts index b0badc1..3c1c039 100644 --- a/src/codex-token-rotation.ts +++ b/src/codex-token-rotation.ts @@ -138,8 +138,11 @@ function saveCodexState(): void { resetD7Ats: accounts.map((a) => a.resetD7At ?? null), }; writeJsonFile(STATE_FILE, state); - } catch { - /* best effort */ + } catch (err) { + logger.warn( + { stateFile: STATE_FILE, err }, + 'Failed to persist Codex rotation state', + ); } } diff --git a/src/logger.test.ts b/src/logger.test.ts new file mode 100644 index 0000000..0428691 --- /dev/null +++ b/src/logger.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it, vi } from 'vitest'; + +describe('logger singleton', () => { + it('reuses the root logger and does not add duplicate process listeners across module reloads', async () => { + vi.resetModules(); + + const beforeUncaught = process.listeners('uncaughtException').length; + const beforeUnhandled = process.listeners('unhandledRejection').length; + const beforeExit = process.listeners('exit').length; + + const first = await import('./logger.js'); + + const afterFirstUncaught = process.listeners('uncaughtException').length; + const afterFirstUnhandled = process.listeners('unhandledRejection').length; + const afterFirstExit = process.listeners('exit').length; + + vi.resetModules(); + + const second = await import('./logger.js'); + + const afterSecondUncaught = process.listeners('uncaughtException').length; + const afterSecondUnhandled = process.listeners('unhandledRejection').length; + const afterSecondExit = process.listeners('exit').length; + + expect(second.logger).toBe(first.logger); + expect(afterFirstUncaught).toBeGreaterThanOrEqual(beforeUncaught); + expect(afterFirstUnhandled).toBeGreaterThanOrEqual(beforeUnhandled); + expect(afterSecondUncaught).toBe(afterFirstUncaught); + expect(afterSecondUnhandled).toBe(afterFirstUnhandled); + expect(afterFirstExit).toBeGreaterThanOrEqual(beforeExit); + expect(afterSecondExit).toBe(afterFirstExit); + }); +}); diff --git a/src/logger.ts b/src/logger.ts index 398583b..2da75a8 100644 --- a/src/logger.ts +++ b/src/logger.ts @@ -1,12 +1,34 @@ import pino, { type Logger } from 'pino'; const serviceName = (process.env.ASSISTANT_NAME || 'claude').toLowerCase(); +const isTestEnv = + process.env.VITEST === 'true' || process.env.NODE_ENV === 'test'; -export const logger = pino({ - level: process.env.LOG_LEVEL || 'info', - name: serviceName, - transport: { target: 'pino-pretty', options: { colorize: true } }, -}); +type LoggerGlobalState = typeof globalThis & { + __ejclawLogger?: Logger; + __ejclawProcessHandlersInstalled?: boolean; +}; + +const globalState = globalThis as LoggerGlobalState; + +function createRootLogger(): Logger { + const baseOptions = { + level: process.env.LOG_LEVEL || 'info', + name: serviceName, + }; + + if (isTestEnv) { + return pino(baseOptions); + } + + return pino({ + ...baseOptions, + transport: { target: 'pino-pretty', options: { colorize: true } }, + }); +} + +export const logger = + globalState.__ejclawLogger ?? (globalState.__ejclawLogger = createRootLogger()); type LogBindings = Record; @@ -30,11 +52,17 @@ export function createScopedLogger(bindings: LogBindings): Logger { } // Route uncaught errors through pino so they get timestamps in stderr -process.on('uncaughtException', (err) => { +function handleUncaughtException(err: unknown): void { logger.fatal({ err }, 'Uncaught exception'); process.exit(1); -}); +} -process.on('unhandledRejection', (reason) => { +function handleUnhandledRejection(reason: unknown): void { logger.error({ err: reason }, 'Unhandled rejection'); -}); +} + +if (!globalState.__ejclawProcessHandlersInstalled) { + process.on('uncaughtException', handleUncaughtException); + process.on('unhandledRejection', handleUnhandledRejection); + globalState.__ejclawProcessHandlersInstalled = true; +} diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index b165207..28ef15a 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -6,9 +6,11 @@ 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'; + executeAttemptRetryAction, + runClaudeAttemptWithRotation, + runCodexAttemptWithRotation, +} from './agent-attempt-orchestration.js'; +import { isRetryableClaudeSessionFailureAttempt } from './agent-attempt-retry.js'; import { AgentOutput, runAgentProcess, @@ -37,10 +39,6 @@ import { resolveExecutionTarget } from './message-runtime-rules.js'; import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js'; import { buildRoomRoleContext } from './room-role-context.js'; import { type AgentTriggerReason } from './agent-error-detection.js'; -import { - runClaudeRotationLoop, - runCodexRotationLoop, -} from './provider-retry.js'; import { shouldResetSessionOnAgentFailure, shouldRetryFreshSessionOnAgentFailure, @@ -649,27 +647,17 @@ export async function runAgentForGroup( initialTrigger: { reason: CodexRotationReason }, rotationMessage?: string, ): Promise<'success' | 'error'> => { - const outcome = await runCodexRotationLoop( + return runCodexAttemptWithRotation({ initialTrigger, - async () => { - const attempt = await runAttempt('codex'); - return { - output: attempt.output, - thrownError: attempt.error, - sawOutput: attempt.sawOutput, - streamedTriggerReason: attempt.streamedTriggerReason, - }; - }, - { + runAttempt: () => runAttempt('codex'), + logContext: { chatJid, group: group.name, groupFolder: group.folder, runId, }, rotationMessage, - ); - - return outcome.type === 'success' ? 'success' : 'error'; + }); }; type AgentAttempt = Awaited>; @@ -688,81 +676,63 @@ export async function runAgentForGroup( runId, }; - const outcome = await runClaudeRotationLoop( + return runClaudeAttemptWithRotation({ initialTrigger, - async () => { - const attempt = await runAttempt('claude'); - return { - output: attempt.output, - thrownError: attempt.error, - sawOutput: attempt.sawOutput, - sawSuccessNullResult: attempt.sawSuccessNullResultWithoutOutput, - streamedTriggerReason: attempt.streamedTriggerReason, - }; - }, - logCtx, + runAttempt: () => runAttempt('claude'), + logContext: logCtx, rotationMessage, - ); - - switch (outcome.type) { - case 'success': - pairedSawOutput = outcome.sawOutput; - return 'success'; - case 'error': - return 'error'; - } + onSuccess: ({ sawOutput }) => { + pairedSawOutput = sawOutput; + }, + }); }; const retryClaudeAttemptIfNeeded = async ( attempt: AgentAttempt, rotationMessage?: string | null, ): Promise<'success' | 'error' | null> => { - const retryAction = resolveAttemptRetryAction({ + const retryAction = await executeAttemptRetryAction({ provider, canRetryClaudeCredentials, canRetryCodex: false, attempt, rotationMessage, + runClaude: retryClaudeWithRotation, + runCodex: retryCodexWithRotation, }); if (retryAction.kind !== 'claude') { return null; } - const result = await retryClaudeWithRotation( - retryAction.trigger, - retryAction.rotationMessage, - ); - if (result === 'error') { + if (retryAction.result === 'error') { return maybeHandoffAfterError(retryAction.trigger.reason, attempt); } pairedExecutionStatus = 'succeeded'; - return result; + return retryAction.result; }; const retryCodexAttemptIfNeeded = async ( attempt: AgentAttempt, rotationMessage?: string | null, ): Promise<'success' | 'error' | null> => { - const retryAction = resolveAttemptRetryAction({ + const retryAction = await executeAttemptRetryAction({ provider, canRetryClaudeCredentials: false, canRetryCodex: !isClaudeCodeAgent && getCodexAccountCount() > 1, attempt, rotationMessage, + runClaude: retryClaudeWithRotation, + runCodex: retryCodexWithRotation, }); if (retryAction.kind !== 'codex') { return null; } - const result = await retryCodexWithRotation( - retryAction.trigger, - retryAction.rotationMessage, - ); - if (result === 'success') { + if (retryAction.result === 'success') { pairedExecutionStatus = 'succeeded'; } - return result; + return retryAction.result; }; const maybeHandoffAfterError = ( diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index 7f7311a..261b43a 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -68,25 +68,6 @@ export function isBotOnlyPairedRoomTurn( ); } -export function resolveLastDeliveredBotRole( - messages: NewMessage[], -): PairedRoomRole | null { - for (let index = messages.length - 1; index >= 0; index -= 1) { - const message = messages[index]; - if (!message?.is_bot_message) { - continue; - } - if ( - message.sender_name === 'owner' || - message.sender_name === 'reviewer' || - message.sender_name === 'arbiter' - ) { - return message.sender_name; - } - } - return null; -} - export function buildPendingPairedTurn(args: { chatJid: string; timezone: string; @@ -98,7 +79,6 @@ export function buildPendingPairedTurn(args: { labeledRecentMessages: Parameters< typeof buildArbiterPromptForTask >[0]['labeledRecentMessages']; - lastDeliveredMessages?: NewMessage[]; resolveChannel: (taskStatus?: string | null) => Channel | null; }): PendingPairedTurn { const { @@ -117,10 +97,7 @@ export function buildPendingPairedTurn(args: { const lastTurnOutput = turnOutputs[turnOutputs.length - 1]; const nextTurnAction = resolveNextTurnAction({ taskStatus, - lastTurnOutputRole: - resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ?? - lastTurnOutput?.role ?? - null, + lastTurnOutputRole: lastTurnOutput?.role ?? null, }); const recentMessages = getRecentChatMessages(chatJid, 20); const lastHumanMessage = getLastHumanMessageContent(chatJid); @@ -255,7 +232,6 @@ export function resolveBotOnlyPairedFollowUpAction(args: { chatJid: string; task: PairedTask | null | undefined; isBotOnlyPairedFollowUp: boolean; - lastDeliveredMessages?: NewMessage[]; pendingCursorSource: | { seq?: number | null; timestamp?: string | null } | undefined; @@ -270,10 +246,7 @@ export function resolveBotOnlyPairedFollowUpAction(args: { const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1); const nextTurnAction = resolveNextTurnAction({ taskStatus: task.status, - lastTurnOutputRole: - resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ?? - lastTurnOutput?.role ?? - null, + lastTurnOutputRole: lastTurnOutput?.role ?? null, }); if (nextTurnAction.kind === 'finalize-owner-turn') { @@ -440,7 +413,6 @@ export function buildQueuedTurnDispatch(args: { chatJid: args.chatJid, task: args.loopPendingTask, isBotOnlyPairedFollowUp, - lastDeliveredMessages: args.labeledMessagesToSend, pendingCursorSource, }); diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 5199a6e..009bcfe 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -1827,124 +1827,6 @@ describe('createMessageRuntime', () => { expect(sendMessage).not.toHaveBeenCalled(); }); - it('requeues owner follow-ups from reviewer bot-only messages even when paired turn output storage is stale', async () => { - const chatJid = 'group@test'; - const group = makeGroup('codex'); - const ownerChannel = makeChannel(chatJid); - const reviewerChannel = { - ...makeChannel(chatJid, 'discord-review', false), - isOwnMessage: vi.fn((message) => message.sender === 'reviewer-bot@test'), - }; - const enqueueMessageCheck = vi.fn(); - const closeStdin = vi.fn(); - const sendMessage = vi.fn(() => false); - const setLastTimestamp = vi.fn(); - const saveState = vi.fn(); - const lastAgentTimestamps: Record = {}; - const stopLoop = new Error('stop-message-loop'); - - vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); - vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ - id: 'task-active-owner-follow-up-stale-output', - chat_jid: chatJid, - group_folder: group.folder, - owner_service_id: 'codex-main', - reviewer_service_id: 'claude', - title: null, - source_ref: 'HEAD', - plan_notes: null, - review_requested_at: '2026-03-30T00:00:00.000Z', - round_trip_count: 1, - status: 'active', - arbiter_verdict: null, - arbiter_requested_at: null, - completion_reason: null, - created_at: '2026-03-30T00:00:00.000Z', - updated_at: '2026-03-30T00:00:00.000Z', - }); - vi.mocked(db.getPairedTurnOutputs).mockReturnValue([ - { - id: 1, - task_id: 'task-active-owner-follow-up-stale-output', - turn_number: 1, - role: 'owner', - output_text: 'owner 초안', - created_at: '2026-03-30T00:00:01.000Z', - }, - ]); - vi.mocked(db.getNewMessages).mockReturnValue({ - messages: [ - { - id: 'reviewer-bot-message-owner-follow-up-stale-output', - chat_jid: chatJid, - sender: 'reviewer-bot@test', - sender_name: '리뷰어', - content: 'DONE_WITH_CONCERNS\n\nreviewer direct message', - timestamp: '2026-03-30T00:00:04.000Z', - seq: 42, - is_bot_message: true, - } as any, - ], - newTimestamp: '42', - }); - - const runtime = createMessageRuntime({ - assistantName: 'Andy', - idleTimeout: 60_000, - pollInterval: 123, - timezone: 'UTC', - triggerPattern: /^@Andy\b/i, - channels: [ownerChannel, reviewerChannel], - queue: { - registerProcess: vi.fn(), - closeStdin, - enqueueMessageCheck, - notifyIdle: vi.fn(), - sendMessage, - } as any, - getRegisteredGroups: () => ({ [chatJid]: group }), - getSessions: () => ({}), - getLastTimestamp: () => '', - setLastTimestamp, - getLastAgentTimestamps: () => lastAgentTimestamps, - saveState, - persistSession: vi.fn(), - clearSession: vi.fn(), - }); - - const originalSetTimeout = global.setTimeout; - const setTimeoutSpy = vi.spyOn(global, 'setTimeout').mockImplementation((( - handler: any, - timeout?: number, - ...args: any[] - ) => { - if (timeout === 123) { - throw stopLoop; - } - return (originalSetTimeout as any)(handler, timeout, ...args); - }) as typeof setTimeout); - - try { - await expect(runtime.startMessageLoop()).rejects.toThrow( - stopLoop.message, - ); - } finally { - setTimeoutSpy.mockRestore(); - } - - expect(setLastTimestamp).toHaveBeenCalledWith('42'); - expect(agentRunner.runAgentProcess).not.toHaveBeenCalled(); - expect(closeStdin).toHaveBeenCalledWith( - chatJid, - expect.objectContaining({ reason: 'paired-pending-turn-follow-up' }), - ); - expect(enqueueMessageCheck).toHaveBeenCalledWith( - chatJid, - resolveGroupIpcPath(group.folder), - ); - expect(sendMessage).not.toHaveBeenCalled(); - }); - it('auto-runs an owner follow-up when a task returns to active after reviewer feedback', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index d822a76..9e53d75 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -851,10 +851,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { chatJid, getRecentChatMessages(chatJid, 20), ), - lastDeliveredMessages: labelPairedSenders( - chatJid, - missedMessages, - ), resolveChannel, }) : null; @@ -1072,6 +1068,191 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } }; + const processQueuedGroupDispatch = async (args: { + chatJid: string; + group: RegisteredGroup; + channel: Channel; + processableGroupMessages: NewMessage[]; + }): Promise => { + const { chatJid, group, channel, processableGroupMessages } = args; + const loopPendingTask = hasReviewerLease(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null; + const loopCursorKey = resolveCursorKey(chatJid, loopPendingTask?.status); + const rawPendingMessages = getMessagesSinceSeq( + chatJid, + deps.getLastAgentTimestamps()[loopCursorKey] || '0', + deps.assistantName, + ); + const pendingMessages = filterLoopingPairedBotMessages( + chatJid, + getProcessableMessages(chatJid, rawPendingMessages, channel), + FAILURE_FINAL_TEXT, + ); + const messagesToSend = + pendingMessages.length > 0 ? pendingMessages : processableGroupMessages; + const labeledMessagesToSend = labelPairedSenders(chatJid, messagesToSend); + const { + formatted, + botOnlyFollowUpAction, + isBotOnlyPairedFollowUp, + loopCursorKey: dispatchCursorKey, + endSeq, + } = buildQueuedTurnDispatch({ + chatJid, + timezone: deps.timezone, + loopPendingTask, + rawPendingMessages, + messagesToSend, + labeledMessagesToSend, + formatMessages, + }); + + if ( + await executeBotOnlyPairedFollowUpAction({ + action: botOnlyFollowUpAction, + chatJid, + group, + runId: `loop-merge-ready-${Date.now().toString(36)}`, + channel, + log: logger, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, + enqueueGroupMessageCheck: () => + enqueueScopedGroupMessageCheck(chatJid, group.folder), + closeStdin: () => + deps.queue.closeStdin(chatJid, { + reason: 'paired-pending-turn-follow-up', + }), + }) + ) { + return; + } + + if (deps.queue.sendMessage(chatJid, formatted)) { + if (endSeq != null) { + advanceLastAgentCursor( + deps.getLastAgentTimestamps(), + deps.saveState, + chatJid, + endSeq, + dispatchCursorKey, + ); + } + logger.debug( + { + transition: 'typing:on', + source: 'follow-up-queued', + chatJid, + group: group.name, + groupFolder: group.folder, + endSeq: endSeq ?? null, + suppressed: isBotOnlyPairedFollowUp, + }, + 'Typing indicator transition', + ); + if (!isBotOnlyPairedFollowUp) { + await channel + .setTyping?.(chatJid, true) + ?.catch((err) => + logger.warn( + { chatJid, err }, + 'Failed to set typing indicator', + ), + ); + } + return; + } + + enqueueScopedGroupMessageCheck(chatJid, group.folder); + }; + + const processLoopGroupMessages = async (args: { + chatJid: string; + group: RegisteredGroup; + groupMessages: NewMessage[]; + channel: Channel; + }): Promise => { + const { chatJid, group, groupMessages, channel } = args; + const isMainGroup = group.isMain === true; + const processableGroupMessages = getProcessableMessages( + chatJid, + groupMessages, + channel, + ); + + if (processableGroupMessages.length === 0) { + const lastIgnored = groupMessages[groupMessages.length - 1]; + if (lastIgnored?.seq != null) { + advanceLastAgentCursor( + deps.getLastAgentTimestamps(), + deps.saveState, + chatJid, + lastIgnored.seq, + ); + } + return; + } + + if (shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages)) { + const lastIgnored = + processableGroupMessages[processableGroupMessages.length - 1]; + if (lastIgnored?.seq != null) { + advanceLastAgentCursor( + deps.getLastAgentTimestamps(), + deps.saveState, + chatJid, + lastIgnored.seq, + ); + } + logger.info( + { chatJid, group: group.name, groupFolder: group.folder }, + 'Bot-collaboration timeout: no recent human message, skipping', + ); + return; + } + + const loopCmdMsg = groupMessages.find( + (msg) => extractSessionCommand(msg.content, deps.triggerPattern) !== null, + ); + + if (loopCmdMsg) { + if ( + isSessionCommandAllowed( + isMainGroup, + loopCmdMsg.is_from_me === true, + isSessionCommandSenderAllowed(loopCmdMsg.sender), + ) + ) { + deps.queue.closeStdin(chatJid, { + reason: 'session-command-detected', + }); + } + deps.queue.enqueueMessageCheck(chatJid); + return; + } + + if ( + !hasAllowedTrigger({ + chatJid, + messages: processableGroupMessages, + group, + triggerPattern: deps.triggerPattern, + hasImplicitContinuationWindow: continuationTracker.has, + }) + ) { + return; + } + + await processQueuedGroupDispatch({ + chatJid, + group, + channel, + processableGroupMessages, + }); + }; + const startMessageLoop = async (): Promise => { if (messageLoopRunning) { logger.debug('Message loop already running, skipping duplicate start'); @@ -1120,182 +1301,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); continue; } - - const isMainGroup = group.isMain === true; - const processableGroupMessages = getProcessableMessages( + await processLoopGroupMessages({ chatJid, + group, groupMessages, channel, - ); - - if (processableGroupMessages.length === 0) { - const lastIgnored = groupMessages[groupMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - continue; - } - - if ( - shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages) - ) { - const lastIgnored = - processableGroupMessages[processableGroupMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - logger.info( - { chatJid, group: group.name, groupFolder: group.folder }, - 'Bot-collaboration timeout: no recent human message, skipping', - ); - continue; - } - - const loopCmdMsg = groupMessages.find( - (msg) => - extractSessionCommand(msg.content, deps.triggerPattern) !== - null, - ); - - if (loopCmdMsg) { - if ( - isSessionCommandAllowed( - isMainGroup, - loopCmdMsg.is_from_me === true, - isSessionCommandSenderAllowed(loopCmdMsg.sender), - ) - ) { - deps.queue.closeStdin(chatJid, { - reason: 'session-command-detected', - }); - } - deps.queue.enqueueMessageCheck(chatJid); - continue; - } - - if ( - !hasAllowedTrigger({ - chatJid, - messages: processableGroupMessages, - group, - triggerPattern: deps.triggerPattern, - hasImplicitContinuationWindow: continuationTracker.has, - }) - ) { - continue; - } - - // Use role-aware cursor for paired rooms so the reviewer - // always sees the owner's last messages and vice versa. - const loopPendingTask = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const loopCursorKey = resolveCursorKey( - chatJid, - loopPendingTask?.status, - ); - - const rawPendingMessages = getMessagesSinceSeq( - chatJid, - deps.getLastAgentTimestamps()[loopCursorKey] || '0', - deps.assistantName, - ); - const pendingMessages = filterLoopingPairedBotMessages( - chatJid, - getProcessableMessages(chatJid, rawPendingMessages, channel), - FAILURE_FINAL_TEXT, - ); - const messagesToSend = - pendingMessages.length > 0 - ? pendingMessages - : processableGroupMessages; - const labeledMessagesToSend = labelPairedSenders( - chatJid, - messagesToSend, - ); - const { - formatted, - botOnlyFollowUpAction, - isBotOnlyPairedFollowUp, - loopCursorKey: dispatchCursorKey, - endSeq, - } = buildQueuedTurnDispatch({ - chatJid, - timezone: deps.timezone, - loopPendingTask, - rawPendingMessages, - messagesToSend, - labeledMessagesToSend, - formatMessages, }); - if ( - await executeBotOnlyPairedFollowUpAction({ - action: botOnlyFollowUpAction, - chatJid, - group, - runId: `loop-merge-ready-${Date.now().toString(36)}`, - channel, - log: logger, - saveState: deps.saveState, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - executeTurn, - enqueueGroupMessageCheck: () => - enqueueScopedGroupMessageCheck(chatJid, group.folder), - closeStdin: () => - deps.queue.closeStdin(chatJid, { - reason: 'paired-pending-turn-follow-up', - }), - }) - ) { - continue; - } - - if (deps.queue.sendMessage(chatJid, formatted)) { - if (endSeq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - endSeq, - dispatchCursorKey, - ); - } - logger.debug( - { - transition: 'typing:on', - source: 'follow-up-queued', - chatJid, - group: group.name, - groupFolder: group.folder, - endSeq: endSeq ?? null, - suppressed: isBotOnlyPairedFollowUp, - }, - 'Typing indicator transition', - ); - if (!isBotOnlyPairedFollowUp) { - await channel - .setTyping?.(chatJid, true) - ?.catch((err) => - logger.warn( - { chatJid, err }, - 'Failed to set typing indicator', - ), - ); - } - continue; - } - - enqueueScopedGroupMessageCheck(chatJid, group.folder); } } } catch (err) { diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index a7dca12..80899f6 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -5,11 +5,11 @@ import { import { getPairedWorkspace, hasActiveCiWatcherForChat, - updatePairedTask, } from './db.js'; import { logger } from './logger.js'; import { markPairedTaskReviewReady } from './paired-workspace-manager.js'; import { + applyPairedTaskPatch, classifyVerdict, hasCodeChangesSinceRef, requestArbiterOrEscalate, @@ -207,10 +207,12 @@ export function handleOwnerCompletion(args: { const result = markPairedTaskReviewReady(taskId); if (result) { - updatePairedTask(taskId, { - round_trip_count: task.round_trip_count + 1, - review_requested_at: now, - updated_at: now, + applyPairedTaskPatch({ + taskId, + updatedAt: now, + patch: { + round_trip_count: task.round_trip_count + 1, + }, }); logger.info( { taskId, roundTrip: task.round_trip_count + 1 }, diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts index 6c1266b..fa8328c 100644 --- a/src/paired-execution-context-shared.ts +++ b/src/paired-execution-context-shared.ts @@ -155,6 +155,17 @@ export function transitionPairedTaskStatus(args: { }); } +export function applyPairedTaskPatch(args: { + taskId: string; + updatedAt: string; + patch: Omit[1], 'updated_at'>; +}): void { + updatePairedTask(args.taskId, { + ...args.patch, + updated_at: args.updatedAt, + }); +} + export function requestArbiterOrEscalate(args: { taskId: string; currentStatus: PairedTaskStatus; diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index 74cfc96..1f57b57 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -565,7 +565,6 @@ describe('paired execution context', () => { 'task-1', expect.objectContaining({ round_trip_count: 2, - review_requested_at: expect.any(String), }), ); }); diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index ff07c62..3f2a3be 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -19,7 +19,6 @@ import { getPairedTaskById, getPairedWorkspace, hasActiveCiWatcherForChat, - updatePairedTask, upsertPairedProject, } from './db.js'; import { logger } from './logger.js'; @@ -36,6 +35,7 @@ import { handleReviewerCompletion, } from './paired-execution-context-reviewer.js'; import { + applyPairedTaskPatch, resolveCanonicalSourceRef, requestArbiterOrEscalate, transitionPairedTaskStatus, @@ -238,9 +238,12 @@ export function preparePairedExecutionContext(args: { }, }); } else { - updatePairedTask(latestTask.id, { - ...(hasHuman ? { round_trip_count: 0 } : {}), - updated_at: now, + applyPairedTaskPatch({ + taskId: latestTask.id, + updatedAt: now, + patch: { + ...(hasHuman ? { round_trip_count: 0 } : {}), + }, }); } } @@ -253,9 +256,12 @@ export function preparePairedExecutionContext(args: { if (workspace?.workspace_dir && latestTask.status === 'active') { const wsRef = resolveCanonicalSourceRef(workspace.workspace_dir); if (wsRef !== latestTask.source_ref) { - updatePairedTask(latestTask.id, { - source_ref: wsRef, - updated_at: now, + applyPairedTaskPatch({ + taskId: latestTask.id, + updatedAt: now, + patch: { + source_ref: wsRef, + }, }); } } diff --git a/src/paired-workspace-manager.test.ts b/src/paired-workspace-manager.test.ts index 929547f..1d20003 100644 --- a/src/paired-workspace-manager.test.ts +++ b/src/paired-workspace-manager.test.ts @@ -258,6 +258,31 @@ describe('paired workspace manager', () => { ); }); + it('leaves review_requested_at untouched when review handoff aborts before an owner workspace exists', async () => { + const { db, manager } = await loadModules(); + db._initTestDatabase(); + + const canonicalDir = path.join(tempRoot, 'canonical'); + initCanonicalRepo(canonicalDir); + seedPairedTask(db, canonicalDir, { + taskId: 'paired-task-no-owner-workspace', + groupFolder: 'no-owner-workspace-room', + }); + + const result = manager.markPairedTaskReviewReady( + 'paired-task-no-owner-workspace', + ); + + expect(result).toBeNull(); + expect( + db.getPairedTaskById('paired-task-no-owner-workspace') + ?.review_requested_at, + ).toBeNull(); + expect( + db.getPairedTaskById('paired-task-no-owner-workspace')?.status, + ).toBe('active'); + }); + it('uses the shared DB owner workspace across service-local data dirs', async () => { const canonicalDir = path.join(tempRoot, 'canonical'); fs.mkdirSync(canonicalDir, { recursive: true }); diff --git a/src/paired-workspace-manager.ts b/src/paired-workspace-manager.ts index 327cfe8..7019ac9 100644 --- a/src/paired-workspace-manager.ts +++ b/src/paired-workspace-manager.ts @@ -8,7 +8,6 @@ import { getPairedProject, getPairedTaskById, getPairedWorkspace, - updatePairedTask, upsertPairedWorkspace, } from './db.js'; import { resolvePairedTaskWorkspacePath } from './group-folder.js'; @@ -758,10 +757,6 @@ export function markPairedTaskReviewReady(taskId: string): { reviewerWorkspace: PairedWorkspace; } | null { const requestedAt = new Date().toISOString(); - updatePairedTask(taskId, { - review_requested_at: requestedAt, - updated_at: requestedAt, - }); const ownerWorkspace = getPairedWorkspace(taskId, 'owner'); if (!ownerWorkspace) { @@ -798,7 +793,6 @@ export function markPairedTaskReviewReady(taskId: string): { 'Reviewer will mount owner workspace directly', ); - const now = new Date().toISOString(); const task = getPairedTaskById(taskId); if (!task) { return null; @@ -807,7 +801,10 @@ export function markPairedTaskReviewReady(taskId: string): { taskId, currentStatus: task.status, nextStatus: 'review_ready', - updatedAt: now, + updatedAt: requestedAt, + patch: { + review_requested_at: requestedAt, + }, }); return { ownerWorkspace, reviewerWorkspace }; diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index ff44b0a..14a2134 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -3,7 +3,11 @@ 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 { + executeAttemptRetryAction, + runClaudeAttemptWithRotation, + runCodexAttemptWithRotation, +} from './agent-attempt-orchestration.js'; import { getErrorMessage } from './utils.js'; import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js'; @@ -29,10 +33,6 @@ import { } from './group-folder.js'; import { createScopedLogger, logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; -import { - runClaudeRotationLoop, - runCodexRotationLoop, -} from './provider-retry.js'; import { detectCodexRotationTrigger, rotateCodexToken, @@ -447,69 +447,53 @@ async function runTask( retryAfterMs?: number; }, rotationMessage?: string, - ): Promise => { - const logCtx = { + ): Promise<'success' | 'error'> => { + const logContext = { taskId: task.id, group: context.group.name, groupFolder: task.group_folder, }; - const outcome = await runClaudeRotationLoop( + const outcome = await runClaudeAttemptWithRotation({ initialTrigger, - async () => { - const attempt = await runTaskAttempt('claude'); + runAttempt: () => runTaskAttempt('claude'), + logContext, + rotationMessage, + afterAttempt: (attempt) => { result = attempt.attemptResult; error = attempt.attemptError; - return { - output: attempt.output, - sawOutput: attempt.sawOutput, - streamedTriggerReason: attempt.streamedTriggerReason, - }; }, - logCtx, - rotationMessage, - ); + }); - switch (outcome.type) { - case 'success': - error = null; - return; - case 'error': - if (outcome.trigger) { - error = `Claude ${outcome.trigger.reason}`; - } - return; + if (outcome === 'success') { + error = null; } + return outcome; }; const retryCodexTaskWithRotation = async ( initialTrigger: { reason: CodexRotationReason }, rotationMessage?: string, - ): Promise => { - const outcome = await runCodexRotationLoop( + ): Promise<'success' | 'error'> => { + const outcome = await runCodexAttemptWithRotation({ initialTrigger, - async () => { - const retryAttempt = await runTaskAttempt('codex'); - result = retryAttempt.attemptResult; - error = retryAttempt.attemptError; - return { - output: retryAttempt.output, - thrownError: null, - sawOutput: retryAttempt.sawOutput, - streamedTriggerReason: retryAttempt.streamedTriggerReason, - }; - }, - { + runAttempt: () => runTaskAttempt('codex'), + logContext: { taskId: task.id, group: context.group.name, groupFolder: task.group_folder, }, rotationMessage, - ); + afterAttempt: (attempt) => { + result = attempt.attemptResult; + error = attempt.attemptError; + }, + }); - if (outcome.type === 'success') { + if (outcome === 'success') { error = null; } + return outcome; }; const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude'; @@ -519,25 +503,17 @@ async function runTask( result = attempt.attemptResult; error = attempt.attemptError; - const retryAction = resolveAttemptRetryAction({ + const retryAction = await executeAttemptRetryAction({ provider, canRetryClaudeCredentials: provider === 'claude' && getTokenCount() > 0, canRetryCodex: provider === 'codex' && getCodexAccountCount() > 1, attempt, rotationMessage: error, + runClaude: retryClaudeTaskWithRotation, + runCodex: retryCodexTaskWithRotation, }); - 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') { + if (retryAction.kind === 'none' && attempt.output.status === 'error') { error = attempt.attemptError || 'Unknown error'; } } // end else (non-exhausted path) diff --git a/src/token-rotation.test.ts b/src/token-rotation.test.ts index e659508..8da8b61 100644 --- a/src/token-rotation.test.ts +++ b/src/token-rotation.test.ts @@ -65,4 +65,44 @@ describe('token-rotation runtime reselection', () => { rateLimited: 0, }); }); + + it('reloads Claude rotation state from disk written by another service', async () => { + const mod = await import('./token-rotation.js'); + const utils = await import('./utils.js'); + const readJsonFile = vi.mocked(utils.readJsonFile); + + readJsonFile.mockReturnValueOnce(null); + mod.initTokenRotation(); + expect(mod.getCurrentToken()).toBe('token-1'); + + readJsonFile.mockReturnValueOnce({ + currentIndex: 1, + rateLimits: [Date.now() + 60_000, null], + }); + mod.reloadTokenRotationStateFromDisk(); + + expect(mod.getCurrentTokenIndex()).toBe(1); + expect(mod.getCurrentToken()).toBe('token-2'); + }); + + it('warns when Claude rotation state cannot be persisted', async () => { + const mod = await import('./token-rotation.js'); + const utils = await import('./utils.js'); + const { logger } = await import('./logger.js'); + + vi.mocked(utils.writeJsonFile).mockImplementation(() => { + throw new Error('disk full'); + }); + + mod.initTokenRotation(); + expect(mod.rotateToken('rate limit')).toBe(true); + + expect(logger.warn).toHaveBeenCalledWith( + expect.objectContaining({ + stateFile: '/tmp/ejclaw-claude-rot-data/token-rotation-state.json', + err: expect.any(Error), + }), + 'Failed to persist Claude token rotation state', + ); + }); }); diff --git a/src/token-rotation.ts b/src/token-rotation.ts index 4bc88e0..9cf6d1d 100644 --- a/src/token-rotation.ts +++ b/src/token-rotation.ts @@ -69,12 +69,15 @@ function saveState(): void { rateLimits: tokens.map((t) => t.rateLimitedUntil), }; writeJsonFile(STATE_FILE, state); - } catch { - /* best effort */ + } catch (err) { + logger.warn( + { stateFile: STATE_FILE, err }, + 'Failed to persist Claude token rotation state', + ); } } -function loadState(): void { +function loadState(quiet = false): void { const state = readJsonFile<{ currentIndex?: number; rateLimits?: (number | null)[]; @@ -93,13 +96,27 @@ function loadState(): void { const until = state.rateLimits[i]; if (typeof until === 'number' && until > now) { tokens[i].rateLimitedUntil = until; + } else { + tokens[i].rateLimitedUntil = null; } } } - logger.info( - { currentIndex, tokenCount: tokens.length }, - 'Token rotation state restored', - ); + if (!quiet) { + logger.info( + { currentIndex, tokenCount: tokens.length }, + 'Token rotation state restored', + ); + } +} + +/** + * Re-read the on-disk rotation state (written by any service). + * Call before dashboard renders so the renderer picks up rotations + * performed by another Claude service process. + */ +export function reloadTokenRotationStateFromDisk(): void { + if (tokens.length <= 1) return; + loadState(true); } function refreshRuntimeTokenSelection(): void {