From dfd9eb0f4773f9901fcd4aff03055701847c5851 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Sat, 4 Apr 2026 05:21:25 +0900 Subject: [PATCH] fix: recover fallback delivery retries across agent types --- src/db.test.ts | 19 +++++ src/db.ts | 14 ++++ src/message-runtime.test.ts | 155 +++++++++++++++++++++++++++++++++++- src/message-runtime.ts | 16 ++-- 4 files changed, 192 insertions(+), 12 deletions(-) diff --git a/src/db.test.ts b/src/db.test.ts index 17200e4..000277e 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -31,6 +31,7 @@ import { getMessagesSinceSeq, getNewMessagesBySeq, getOpenWorkItem, + getOpenWorkItemForChat, getPendingServiceHandoffs, recallMemories, getRegisteredAgentTypesForJid, @@ -2603,6 +2604,24 @@ describe('work items', () => { expect(getOpenWorkItem('dc:123', 'claude-code')).toBeUndefined(); }); + it('finds pending delivery retries even when they were created by a fallback agent type', () => { + const fallbackItem = createProducedWorkItem({ + group_folder: 'discord_test', + chat_jid: 'dc:fallback', + agent_type: 'codex', + delivery_role: 'reviewer', + start_seq: 20, + end_seq: 22, + result_payload: 'fallback reviewer output', + }); + + expect(getOpenWorkItem('dc:fallback', 'claude-code')).toBeUndefined(); + expect(getOpenWorkItemForChat('dc:fallback')?.id).toBe(fallbackItem.id); + + markWorkItemDelivered(fallbackItem.id, 'msg-fallback'); + expect(getOpenWorkItemForChat('dc:fallback')).toBeUndefined(); + }); + it('stores service shadow as a derived compatibility field from role and agent type', () => { const reviewerItem = createProducedWorkItem({ group_folder: 'discord_test', diff --git a/src/db.ts b/src/db.ts index a6d4b8a..15be1fd 100644 --- a/src/db.ts +++ b/src/db.ts @@ -2039,6 +2039,20 @@ export function getOpenWorkItem( return row ? hydrateWorkItem(row) : undefined; } +export function getOpenWorkItemForChat(chatJid: string): WorkItem | undefined { + const row = db + .prepare( + `SELECT * + FROM work_items + WHERE chat_jid = ? + AND status IN ('produced', 'delivery_retry') + ORDER BY id ASC + LIMIT 1`, + ) + .get(chatJid) as StoredWorkItemRow | undefined; + return row ? hydrateWorkItem(row) : undefined; +} + export function createProducedWorkItem(input: { group_folder: string; chat_jid: string; diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 9f8032e..fbb317d 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -37,6 +37,13 @@ vi.mock('./paired-execution-context.js', () => ({ })); vi.mock('./db.js', () => { + const getOpenWorkItem = vi.fn( + ( + _chatJid?: string, + _agentType?: 'claude-code' | 'codex', + _serviceId?: string, + ) => undefined, + ); const getMessagesSince = vi.fn( ( _chatJid?: string, @@ -114,7 +121,8 @@ vi.mock('./db.js', () => { }; }, ), - getOpenWorkItem: vi.fn(() => undefined), + getOpenWorkItem, + getOpenWorkItemForChat: vi.fn((chatJid: string) => getOpenWorkItem(chatJid)), getLatestOpenPairedTaskForChat: vi.fn(() => undefined), getPairedTurnOutputs: vi.fn(() => []), getRecentChatMessages: vi.fn(() => []), @@ -782,6 +790,90 @@ describe('createMessageRuntime', () => { expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid); }); + it('retries a stale fallback reviewer work item even when the room owner agent type is different', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const ownerChannel = makeChannel(chatJid); + const reviewerChannel = makeChannel(chatJid, 'discord-review', false); + const enqueueMessageCheck = vi.fn(); + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + vi.mocked(db.getOpenWorkItem).mockReturnValue(undefined); + vi.mocked(db.getOpenWorkItemForChat).mockReturnValue({ + id: 102, + group_folder: group.folder, + chat_jid: chatJid, + agent_type: 'codex', + service_id: 'codex-review', + delivery_role: 'reviewer', + status: 'delivery_retry', + start_seq: 12, + end_seq: 13, + result_payload: 'fallback reviewer final retry', + delivery_attempts: 1, + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + delivered_at: null, + delivery_message_id: null, + last_error: 'discord send failed', + }); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({ + id: 'task-fallback-review-delivery-role', + chat_jid: chatJid, + group_folder: group.folder, + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: '2026-03-30T00:00:00.000Z', + round_trip_count: 1, + status: 'merge_ready', + 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', + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [ownerChannel, reviewerChannel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + enqueueMessageCheck, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-open-fallback-review-work-item-persisted-role', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid); + expect(reviewerChannel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'fallback reviewer final retry', + ); + expect(ownerChannel.sendMessage).not.toHaveBeenCalled(); + expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid); + }); + it('does not inject filtered raw bot finals into workspace-based review prompts', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); @@ -3388,4 +3480,65 @@ describe('createMessageRuntime', () => { expect(enqueueTask).not.toHaveBeenCalled(); expect(db.getMessagesSinceSeq).not.toHaveBeenCalled(); }); + + it('recovery also queues fallback delivery retries across agent types', () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const enqueueMessageCheck = vi.fn(); + const enqueueTask = vi.fn(); + + vi.mocked(db.getOpenWorkItem).mockReturnValue(undefined); + vi.mocked(db.getOpenWorkItemForChat).mockReturnValue({ + id: 199, + group_folder: group.folder, + chat_jid: chatJid, + agent_type: 'codex', + service_id: 'codex-review', + delivery_role: 'reviewer', + status: 'delivery_retry', + start_seq: 5, + end_seq: 6, + result_payload: '미전달 reviewer 결과', + delivery_attempts: 1, + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + delivered_at: null, + delivery_message_id: null, + last_error: 'discord send failed', + }); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [makeChannel(chatJid)], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + enqueueMessageCheck, + enqueueTask, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + runtime.recoverPendingMessages(); + + expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid); + expect(enqueueMessageCheck).toHaveBeenCalledWith( + chatJid, + resolveGroupIpcPath(group.folder), + ); + expect(enqueueTask).not.toHaveBeenCalled(); + expect(db.getMessagesSinceSeq).not.toHaveBeenCalled(); + }); }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 1e36767..f849544 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -6,7 +6,7 @@ import { createProducedWorkItem, failServiceHandoff, getAllPendingServiceHandoffs, - getOpenWorkItem, + getOpenWorkItemForChat, getMessagesSinceSeq, getNewMessagesBySeq, markWorkItemDelivered, @@ -830,12 +830,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); } - // Deliver pending work items through the correct channel. - // For paired rooms, check if the work item was from a reviewer turn. - const openWorkItem = getOpenWorkItem( - chatJid, - (group.agentType || 'claude-code') as 'claude-code' | 'codex', - ); + // Delivery retries can come from forced fallback runs whose agent_type + // differs from the room owner's registered agent type. + const openWorkItem = getOpenWorkItemForChat(chatJid); if (openWorkItem) { const pendingTask = hasReviewerLease(chatJid) ? getLatestOpenPairedTaskForChat(chatJid) @@ -1313,10 +1310,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const recoverPendingMessages = (): void => { const registeredGroups = deps.getRegisteredGroups(); for (const [chatJid, group] of Object.entries(registeredGroups)) { - const openWorkItem = getOpenWorkItem( - chatJid, - (group.agentType || 'claude-code') as 'claude-code' | 'codex', - ); + const openWorkItem = getOpenWorkItemForChat(chatJid); if (openWorkItem) { logger.info( { chatJid, group: group.name, workItemId: openWorkItem.id },