diff --git a/src/message-runtime-queue-cursor.test.ts b/src/message-runtime-queue-cursor.test.ts new file mode 100644 index 0000000..3e543ad --- /dev/null +++ b/src/message-runtime-queue-cursor.test.ts @@ -0,0 +1,186 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { _initTestDatabase, createPairedTask } from './db.js'; +import { runQueuedGroupTurn } from './message-runtime-queue.js'; +import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js'; +import type { + Channel, + NewMessage, + PairedTask, + RegisteredGroup, +} from './types.js'; + +type RunQueuedGroupTurnArgs = Parameters[0]; + +function makeGroup(): RegisteredGroup { + return { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: '2026-03-30T00:00:00.000Z', + requiresTrigger: false, + agentType: 'codex', + workDir: '/repo', + }; +} + +function makeTask(overrides: Partial = {}): PairedTask { + return { + id: 'task-queued-cursor', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + owner_agent_type: 'claude-code', + reviewer_agent_type: 'codex', + arbiter_agent_type: null, + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: null, + 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', + ...overrides, + }; +} + +function makeChannel(): Channel { + return { + name: 'discord-owner', + connect: vi.fn(), + sendMessage: vi.fn(), + isConnected: vi.fn(() => true), + ownsJid: vi.fn(() => true), + disconnect: vi.fn(), + } as unknown as Channel; +} + +function makeQueuedTurnArgs(overrides: { + task: PairedTask; + executeTurn: RunQueuedGroupTurnArgs['executeTurn']; + lastAgentTimestamps: Record; + saveState: () => void; + log?: RunQueuedGroupTurnArgs['log']; +}): RunQueuedGroupTurnArgs { + return { + chatJid: overrides.task.chat_jid, + group: makeGroup(), + runId: 'run-owner-queued-cursor', + log: + overrides.log ?? + ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any), + timezone: 'UTC', + missedMessages: [ + { + id: 'human-owner-cursor', + chat_jid: overrides.task.chat_jid, + sender: 'user@test', + sender_name: 'User', + content: '중간에 이 내용도 봐줘', + timestamp: '2026-03-30T00:00:04.000Z', + seq: 48, + is_bot_message: false, + }, + ], + task: overrides.task, + roleToChannel: { + owner: null, + reviewer: makeChannel(), + arbiter: null, + }, + ownerChannel: makeChannel(), + lastAgentTimestamps: overrides.lastAgentTimestamps, + saveState: overrides.saveState, + executeTurn: overrides.executeTurn, + getFixedRoleChannelName: () => 'discord-review', + labelPairedSenders: (_chatJid: string, messages: NewMessage[]) => messages, + formatMessages: () => 'formatted prompt', + }; +} + +describe('message-runtime queued cursor handling', () => { + beforeEach(() => { + _initTestDatabase(); + resetPairedFollowUpScheduleState(); + }); + + it('keeps the queued human cursor when the owner turn fails silently', async () => { + const task = makeTask({ + owner_service_id: 'codex', + owner_agent_type: 'codex', + }); + createPairedTask(task); + const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn( + async () => ({ + outputStatus: 'error' as const, + deliverySucceeded: true, + visiblePhase: 'silent', + }), + ); + const lastAgentTimestamps: Record = {}; + const saveState = vi.fn(() => undefined); + const logMocks = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }; + const log = logMocks as unknown as RunQueuedGroupTurnArgs['log']; + + const outcome = await runQueuedGroupTurn( + makeQueuedTurnArgs({ + task, + executeTurn, + lastAgentTimestamps, + saveState, + log, + }), + ); + + expect(outcome).toBe(false); + expect(executeTurn).toHaveBeenCalledTimes(1); + expect(lastAgentTimestamps).toEqual({}); + expect(saveState).toHaveBeenCalledTimes(2); + expect(logMocks.warn).toHaveBeenCalledWith( + { + messageSeqStart: 48, + messageSeqEnd: 48, + }, + 'Queued run failed before producing visible output; keeping cursor for retry', + ); + }); + + it('advances the queued human cursor when a silent owner turn succeeds', async () => { + const task = makeTask({ + owner_service_id: 'codex', + owner_agent_type: 'codex', + }); + createPairedTask(task); + const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn( + async () => ({ + outputStatus: 'success' as const, + deliverySucceeded: true, + visiblePhase: 'silent', + }), + ); + const lastAgentTimestamps: Record = {}; + const saveState = vi.fn(() => undefined); + + const outcome = await runQueuedGroupTurn( + makeQueuedTurnArgs({ + task, + executeTurn, + lastAgentTimestamps, + saveState, + }), + ); + + expect(outcome).toBe(true); + expect(lastAgentTimestamps).toEqual({ 'group@test': '48' }); + expect(saveState).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts index 79a8a00..3f83084 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -14,6 +14,7 @@ import { buildPairedTurnIdentity } from './paired-turn-identity.js'; import { resolveStoredVisibleVerdict } from './paired-verdict.js'; import { advanceLastAgentCursor, + finalizeQueuedRunCursor, resolveActiveRole, resolveCursorKeyForRole, resolveQueuedPairedTurnRole, @@ -28,6 +29,7 @@ import type { Channel, NewMessage, PairedTask, + PairedTaskStatus, PairedTurnReservationIntentKind, RegisteredGroup, } from './types.js'; @@ -52,6 +54,86 @@ function resolveQueuedTurnReservationIntent(args: { return 'owner-follow-up'; } +function buildQueuedGroupTurnPrompt(args: { + turnRole: 'owner' | 'reviewer' | 'arbiter'; + currentTask: PairedTask | null | undefined; + chatJid: string; + timezone: string; + missedMessages: NewMessage[]; + fallbackMessages: NewMessage[]; + turnOutputs: ReturnType; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; + formatMessages: (messages: NewMessage[], timezone: string) => string; +}): string { + if (args.turnRole === 'arbiter' && args.currentTask) { + const recentMessages = getRecentChatMessages(args.chatJid, 20); + return buildArbiterPromptForTask({ + task: args.currentTask, + chatJid: args.chatJid, + timezone: args.timezone, + turnOutputs: args.turnOutputs, + recentMessages, + labeledRecentMessages: args.labelPairedSenders( + args.chatJid, + recentMessages, + ), + }); + } + + if (args.currentTask) { + return buildPairedTurnPrompt({ + taskId: args.currentTask.id, + chatJid: args.chatJid, + timezone: args.timezone, + missedMessages: args.missedMessages, + labeledFallbackMessages: args.labelPairedSenders( + args.chatJid, + args.fallbackMessages, + ), + turnOutputs: args.turnOutputs, + }); + } + + return args.formatMessages( + args.labelPairedSenders(args.chatJid, args.missedMessages), + args.timezone, + ); +} + +function claimQueuedPairedTurn(args: { + chatJid: string; + runId: string; + task: PairedTask | null | undefined; + taskStatus: PairedTaskStatus | null | undefined; + intentKind: PairedTurnReservationIntentKind | null; + turnRole: 'owner' | 'reviewer' | 'arbiter'; + log: typeof logger; +}): boolean { + if (!args.task) { + return true; + } + const claimed = claimPairedTurnExecution({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + intentKind: args.intentKind!, + }); + if (claimed) { + return true; + } + args.log.info( + { + taskId: args.task.id, + taskStatus: args.taskStatus, + taskUpdatedAt: args.task.updated_at, + intentKind: args.intentKind, + turnRole: args.turnRole, + }, + 'Skipped queued paired turn because the task revision was already claimed elsewhere', + ); + return false; +} + export async function runPendingPairedTurnIfNeeded(args: { chatJid: string; group: RegisteredGroup; @@ -246,35 +328,17 @@ export async function runQueuedGroupTurn(args: { }) : undefined; - let prompt: string; - if (turnRole === 'arbiter' && currentTask) { - const recentMessages = getRecentChatMessages(chatJid, 20); - prompt = buildArbiterPromptForTask({ - task: currentTask, - chatJid, - timezone: args.timezone, - turnOutputs, - recentMessages, - labeledRecentMessages: args.labelPairedSenders(chatJid, recentMessages), - }); - } else if (currentTask) { - prompt = buildPairedTurnPrompt({ - taskId: currentTask.id, - chatJid, - timezone: args.timezone, - missedMessages, - labeledFallbackMessages: args.labelPairedSenders( - chatJid, - fallbackMessages, - ), - turnOutputs, - }); - } else { - prompt = args.formatMessages( - args.labelPairedSenders(chatJid, missedMessages), - args.timezone, - ); - } + const prompt = buildQueuedGroupTurnPrompt({ + turnRole, + currentTask, + chatJid, + timezone: args.timezone, + missedMessages, + fallbackMessages, + turnOutputs, + labelPairedSenders: args.labelPairedSenders, + formatMessages: args.formatMessages, + }); const startSeq = missedMessages[0].seq ?? null; const endSeq = missedMessages[missedMessages.length - 1].seq ?? null; @@ -300,29 +364,23 @@ export async function runQueuedGroupTurn(args: { return false; } - if (currentTask) { - const claimed = claimPairedTurnExecution({ + if ( + !claimQueuedPairedTurn({ chatJid, runId, task: currentTask, + taskStatus, intentKind: queuedIntentKind!, - }); - if (!claimed) { - log.info( - { - taskId: currentTask.id, - taskStatus, - taskUpdatedAt: currentTask.updated_at, - intentKind: queuedIntentKind, - turnRole, - }, - 'Skipped queued paired turn because the task revision was already claimed elsewhere', - ); - return true; - } + turnRole, + log, + }) + ) { + return true; } - if (endSeq !== null) { + const previousCursor = args.lastAgentTimestamps[cursorKey]; + const cursorAdvanced = endSeq !== null; + if (cursorAdvanced) { advanceLastAgentCursor( args.lastAgentTimestamps, args.saveState, @@ -332,19 +390,41 @@ export async function runQueuedGroupTurn(args: { ); } - const { deliverySucceeded, visiblePhase } = await args.executeTurn({ - group, - prompt, - chatJid, - runId, - channel: turnChannel, - deliveryRole: currentTask ? turnRole : undefined, - startSeq, - endSeq, - hasHumanMessage: hasHumanMsg, - forcedRole, - pairedTurnIdentity, - }); + const { outputStatus, deliverySucceeded, visiblePhase } = + await args.executeTurn({ + group, + prompt, + chatJid, + runId, + channel: turnChannel, + deliveryRole: currentTask ? turnRole : undefined, + startSeq, + endSeq, + hasHumanMessage: hasHumanMsg, + forcedRole, + pairedTurnIdentity, + }); + + if ( + !finalizeQueuedRunCursor({ + outputStatus, + visiblePhase, + startSeq, + endSeq, + rollbackOnSilentError: + hasHumanMsg && + turnRole === 'owner' && + currentTask?.owner_agent_type === 'codex', + cursorAdvanced, + previousCursor, + lastAgentTimestamps: args.lastAgentTimestamps, + saveState: args.saveState, + cursorKey, + log, + }) + ) { + return false; + } if (!deliverySucceeded) { log.warn( diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 9725e3c..49f2312 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -46,6 +46,47 @@ export function advanceLastAgentCursor( saveState(); } +export function finalizeQueuedRunCursor(args: { + outputStatus: 'success' | 'error'; + visiblePhase: unknown; + startSeq: number | null; + endSeq: number | null; + rollbackOnSilentError: boolean; + cursorAdvanced: boolean; + previousCursor: string | undefined; + lastAgentTimestamps: Record; + saveState: () => void; + cursorKey: string; + log: { + warn: (bindings: Record, message: string) => void; + }; +}): boolean { + if ( + args.rollbackOnSilentError && + args.outputStatus === 'error' && + args.visiblePhase === 'silent' + ) { + if (args.cursorAdvanced) { + if (args.previousCursor === undefined) { + delete args.lastAgentTimestamps[args.cursorKey]; + } else { + args.lastAgentTimestamps[args.cursorKey] = args.previousCursor; + } + args.saveState(); + } + args.log.warn( + { + messageSeqStart: args.startSeq, + messageSeqEnd: args.endSeq, + }, + 'Queued run failed before producing visible output; keeping cursor for retry', + ); + return false; + } + + return true; +} + /** Map task status to the active role. */ export function resolveActiveRole( taskStatus?: string | null,