diff --git a/src/db.ts b/src/db.ts index d80a362..dc60ec4 100644 --- a/src/db.ts +++ b/src/db.ts @@ -507,10 +507,6 @@ export function getAllChats(): ChatInfo[] { * Only call this for registered groups where message history is needed. */ export function storeMessage(msg: NewMessage): void { - const existing = db - .prepare('SELECT seq FROM messages WHERE id = ? AND chat_jid = ?') - .get(msg.id, msg.chat_jid) as { seq: number | null } | undefined; - const nextSeq = () => { const result = db .prepare('INSERT INTO message_sequence DEFAULT VALUES') @@ -519,36 +515,21 @@ export function storeMessage(msg: NewMessage): void { }; db.transaction(() => { - if (existing?.seq != null) { - db.prepare( - `INSERT INTO messages ( - id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(id, chat_jid) DO UPDATE SET - sender = excluded.sender, - sender_name = excluded.sender_name, - content = excluded.content, - timestamp = excluded.timestamp, - is_from_me = excluded.is_from_me, - is_bot_message = excluded.is_bot_message`, - ).run( - msg.id, - msg.chat_jid, - msg.sender, - msg.sender_name, - msg.content, - msg.timestamp, - existing.seq, - msg.is_from_me ? 1 : 0, - msg.is_bot_message ? 1 : 0, - ); - return; - } - + const existing = db + .prepare('SELECT seq FROM messages WHERE id = ? AND chat_jid = ?') + .get(msg.id, msg.chat_jid) as { seq: number | null } | undefined; + const seq = existing?.seq ?? nextSeq(); db.prepare( `INSERT INTO messages ( id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id, chat_jid) DO UPDATE SET + sender = excluded.sender, + sender_name = excluded.sender_name, + content = excluded.content, + timestamp = excluded.timestamp, + is_from_me = excluded.is_from_me, + is_bot_message = excluded.is_bot_message`, ).run( msg.id, msg.chat_jid, @@ -556,7 +537,7 @@ export function storeMessage(msg: NewMessage): void { msg.sender_name, msg.content, msg.timestamp, - nextSeq(), + seq, msg.is_from_me ? 1 : 0, msg.is_bot_message ? 1 : 0, ); diff --git a/src/index.ts b/src/index.ts index da8d0ae..61a1f35 100644 --- a/src/index.ts +++ b/src/index.ts @@ -50,7 +50,9 @@ import { buildRestartAnnouncement, buildInterruptedRestartAnnouncement, consumeRestartContext, + getInterruptedRecoveryCandidates, inferRecentRestartContext, + type RestartContext, writeShutdownRestartContext, } from './restart-context.js'; import { @@ -980,7 +982,7 @@ async function startUsageDashboard(): Promise { async function announceRestartRecovery( processStartedAtMs: number, -): Promise { +): Promise { const explicitContext = consumeRestartContext(); const dedupeSince = new Date(processStartedAtMs - 60_000).toISOString(); if (explicitContext) { @@ -989,7 +991,7 @@ async function announceRestartRecovery( { chatJid: explicitContext.chatJid }, 'Skipped duplicate restart recovery announcement', ); - return; + return explicitContext; } await sendFormattedChannelMessage( @@ -1013,21 +1015,21 @@ async function announceRestartRecovery( buildInterruptedRestartAnnouncement(interrupted), ); } - return; + return explicitContext; } const inferred = inferRecentRestartContext( registeredGroups, processStartedAtMs, ); - if (!inferred) return; + if (!inferred) return null; if (hasRecentRestartAnnouncement(inferred.chatJid, dedupeSince)) { logger.info( { chatJid: inferred.chatJid }, 'Skipped duplicate inferred restart recovery announcement', ); - return; + return null; } await sendFormattedChannelMessage( @@ -1039,6 +1041,7 @@ async function announceRestartRecovery( { chatJid: inferred.chatJid }, 'Sent inferred restart recovery announcement', ); + return null; } async function main(): Promise { @@ -1175,7 +1178,23 @@ async function main(): Promise { }); queue.setProcessMessagesFn(runtime.processGroupMessages); runtime.recoverPendingMessages(); - await announceRestartRecovery(processStartedAtMs); + const restartContext = await announceRestartRecovery(processStartedAtMs); + for (const candidate of getInterruptedRecoveryCandidates( + restartContext, + registeredGroups, + )) { + queue.enqueueMessageCheck(candidate.chatJid, candidate.groupFolder); + logger.info( + { + chatJid: candidate.chatJid, + groupFolder: candidate.groupFolder, + status: candidate.status, + pendingMessages: candidate.pendingMessages, + pendingTasks: candidate.pendingTasks, + }, + 'Queued interrupted group for restart recovery', + ); + } // Purge old messages in status channel before creating fresh dashboards if (STATUS_CHANNEL_ID && SERVICE_AGENT_TYPE === 'claude-code') { const statusJid = `dc:${STATUS_CHANNEL_ID}`; diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 4ee99f5..893722a 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -1411,4 +1411,55 @@ describe('createMessageRuntime', () => { 'success-null-result 폴백 응답입니다.', ); }); + + it('recovery queues a group when an open work item is waiting for delivery', () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const enqueueMessageCheck = vi.fn(); + + vi.mocked(db.getOpenWorkItem).mockReturnValue({ + id: 99, + group_folder: group.folder, + chat_jid: chatJid, + agent_type: 'claude-code', + status: 'produced', + start_seq: 1, + end_seq: 1, + result_payload: '미전달 결과', + delivery_attempts: 0, + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + delivered_at: null, + delivery_message_id: null, + last_error: null, + }); + + 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, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + runtime.recoverPendingMessages(); + + expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid, group.folder); + expect(db.getMessagesSinceSeq).not.toHaveBeenCalled(); + }); }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 825c7df..b71ffff 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1288,6 +1288,19 @@ 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', + ); + if (openWorkItem) { + logger.info( + { chatJid, group: group.name, workItemId: openWorkItem.id }, + 'Recovery: found open work item awaiting delivery', + ); + deps.queue.enqueueMessageCheck(chatJid, group.folder); + continue; + } + const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || ''; const rawPending = getMessagesSinceSeq( chatJid, diff --git a/src/restart-context.test.ts b/src/restart-context.test.ts new file mode 100644 index 0000000..c1baf4b --- /dev/null +++ b/src/restart-context.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from 'vitest'; + +import { + getInterruptedRecoveryCandidates, + type RestartContext, +} from './restart-context.js'; +import type { RegisteredGroup } from './types.js'; + +function makeGroup(folder: string): RegisteredGroup { + return { + name: folder, + folder, + trigger: '@Andy', + added_at: new Date().toISOString(), + requiresTrigger: false, + }; +} + +describe('getInterruptedRecoveryCandidates', () => { + it('returns only registered interrupted groups and deduplicates by chatJid', () => { + const registeredGroups: Record = { + 'dc:1': makeGroup('group-one'), + 'dc:2': makeGroup('group-two'), + }; + + const context: RestartContext = { + chatJid: 'dc:main', + summary: 'restart', + verify: [], + writtenAt: new Date().toISOString(), + interruptedGroups: [ + { + chatJid: 'dc:1', + groupName: 'one', + status: 'processing', + elapsedMs: 1000, + pendingMessages: true, + pendingTasks: 0, + }, + { + chatJid: 'dc:1', + groupName: 'one-duplicate', + status: 'waiting', + elapsedMs: null, + pendingMessages: false, + pendingTasks: 1, + }, + { + chatJid: 'dc:2', + groupName: 'two', + status: 'idle', + elapsedMs: null, + pendingMessages: false, + pendingTasks: 0, + }, + { + chatJid: 'dc:3', + groupName: 'missing', + status: 'processing', + elapsedMs: 500, + pendingMessages: true, + pendingTasks: 0, + }, + ], + }; + + expect( + getInterruptedRecoveryCandidates(context, registeredGroups), + ).toEqual([ + { + chatJid: 'dc:1', + groupFolder: 'group-one', + status: 'processing', + pendingMessages: true, + pendingTasks: 0, + }, + { + chatJid: 'dc:2', + groupFolder: 'group-two', + status: 'idle', + pendingMessages: false, + pendingTasks: 0, + }, + ]); + }); + + it('returns empty when there is no explicit restart context', () => { + expect(getInterruptedRecoveryCandidates(null, {})).toEqual([]); + }); +}); diff --git a/src/restart-context.ts b/src/restart-context.ts index 914fa73..480f1bc 100644 --- a/src/restart-context.ts +++ b/src/restart-context.ts @@ -30,6 +30,14 @@ export interface InferredRestartContext { lines: string[]; } +export interface RestartRecoveryCandidate { + chatJid: string; + groupFolder: string; + status: RestartInterruptedGroup['status']; + pendingMessages: boolean; + pendingTasks: number; +} + const INFER_WINDOW_MS = 3 * 60 * 1000; function getRestartContextPath(serviceId: string = SERVICE_ID): string { @@ -229,6 +237,32 @@ export function buildRestartAnnouncement(context: RestartContext): string { return lines.join('\n'); } +export function getInterruptedRecoveryCandidates( + context: RestartContext | null, + registeredGroups: Record, +): RestartRecoveryCandidate[] { + if (!context?.interruptedGroups?.length) return []; + + const seen = new Set(); + const candidates: RestartRecoveryCandidate[] = []; + + for (const interrupted of context.interruptedGroups) { + if (seen.has(interrupted.chatJid)) continue; + const group = registeredGroups[interrupted.chatJid]; + if (!group) continue; + seen.add(interrupted.chatJid); + candidates.push({ + chatJid: interrupted.chatJid, + groupFolder: group.folder, + status: interrupted.status, + pendingMessages: interrupted.pendingMessages, + pendingTasks: interrupted.pendingTasks, + }); + } + + return candidates; +} + export function inferRecentRestartContext( registeredGroups: Record, processStartedAtMs: number,