diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 893722a..dd4926e 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -1167,6 +1167,217 @@ describe('createMessageRuntime', () => { } }); + it('recreates a tracked progress message when editing the existing one fails', async () => { + vi.useFakeTimers(); + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(channel.sendAndTrack!) + .mockResolvedValueOnce('progress-1') + .mockResolvedValueOnce('progress-2'); + vi.mocked(channel.editMessage!).mockRejectedValueOnce( + new Error('discord edit failed'), + ); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '진행 중입니다.', + newSessionId: 'session-progress-recreate', + }); + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '진행 중입니다.', + newSessionId: 'session-progress-recreate', + }); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-progress-recreate', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-progress-recreate', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + try { + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-progress-recreate', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendAndTrack).toHaveBeenNthCalledWith( + 1, + chatJid, + '진행 중입니다.\n\n0초', + ); + expect(channel.editMessage).toHaveBeenCalledWith( + chatJid, + 'progress-1', + '진행 중입니다.\n\n10초', + ); + expect(channel.sendAndTrack).toHaveBeenNthCalledWith( + 2, + chatJid, + '진행 중입니다.\n\n10초', + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '진행 중입니다.', + ); + } finally { + vi.useRealTimers(); + } + }); + + it('requeues a queued follow-up as a fresh run when the active agent ends without follow-up output', async () => { + vi.useFakeTimers(); + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const closeStdin = vi.fn(); + const enqueueMessageCheck = vi.fn(); + let activityTouch: + | ((meta?: { + source: 'follow-up'; + textLength: number; + filename: string; + }) => void) + | null = null; + const lastAgentTimestamps: Record = { [chatJid]: '1' }; + const saveState = vi.fn(); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + seq: 2, + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '초기 턴 응답입니다.', + newSessionId: 'session-follow-up-rerun', + }); + activityTouch?.({ + source: 'follow-up', + textLength: 48, + filename: 'follow-up.json', + }); + lastAgentTimestamps[chatJid] = '3'; + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-follow-up-rerun', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-follow-up-rerun', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 60_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin, + notifyIdle: vi.fn(), + enqueueMessageCheck, + setActivityTouch: vi.fn( + (_jid, touch) => (activityTouch = touch as typeof activityTouch), + ), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => lastAgentTimestamps, + saveState, + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + try { + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-follow-up-rerun', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(closeStdin).toHaveBeenCalledWith(chatJid, { + runId: 'run-follow-up-rerun', + reason: 'follow-up-no-output-preemption', + }); + expect(lastAgentTimestamps[chatJid]).toBe('2'); + expect(saveState).toHaveBeenCalled(); + expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid, group.folder); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '초기 턴 응답입니다.', + ); + } finally { + vi.useRealTimers(); + } + }); + it('does not roll back when a streamed progress message was already posted before an error', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index b71ffff..9ca9989 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -627,6 +627,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let followUpQueuedAt: number | null = null; let followUpQueuedTextLength: number | null = null; let followUpQueuedFilename: string | null = null; + let followUpRollbackCursor: string | null = null; + let followUpRestartRequested = false; + let followUpEndedWithoutOutputReason: string | null = null; let followUpNoOutputWarnTimer: ReturnType | null = null; let hadError = false; let latestProgressText: string | null = null; @@ -665,6 +668,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 'Idle timeout reached while a queued follow-up still had no agent output', ); + followUpRestartRequested = true; } deps.queue.closeStdin(chatJid, { reason: 'idle-timeout' }); }, deps.idleTimeout); @@ -682,6 +686,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { followUpQueuedAt = null; followUpQueuedTextLength = null; followUpQueuedFilename = null; + followUpRollbackCursor = null; + followUpRestartRequested = false; + followUpEndedWithoutOutputReason = null; }; const scheduleFollowUpNoOutputWarning = () => { @@ -700,6 +707,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 'No agent output observed within 10s of queuing a follow-up message', ); + followUpRestartRequested = true; + deps.queue.closeStdin(chatJid, { + runId, + reason: 'follow-up-no-output-preemption', + }); }, FOLLOW_UP_NO_OUTPUT_WARN_MS); }; @@ -709,6 +721,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { filename: string; }) => { if (meta?.source !== 'follow-up') return; + if (followUpRollbackCursor === null) { + followUpRollbackCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; + } followUpQueuedAt = Date.now(); followUpQueuedTextLength = meta.textLength; followUpQueuedFilename = meta.filename; @@ -745,6 +760,22 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const warnFollowUpEndedWithoutOutput = (reasonText: string) => { if (followUpQueuedAt === null) return; + followUpEndedWithoutOutputReason = reasonText; + }; + + const requeuePendingFollowUp = (reasonText: string): boolean => { + if (followUpQueuedAt === null || followUpRollbackCursor === null) { + return false; + } + + const lastAgentTimestamps = deps.getLastAgentTimestamps(); + const currentCursor = lastAgentTimestamps[chatJid] || '0'; + lastAgentTimestamps[chatJid] = normalizeStoredSeqCursor( + followUpRollbackCursor, + chatJid, + ); + deps.saveState(); + logger.warn( { group: group.name, @@ -755,10 +786,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { followUpQueuedFilename, followUpWaitMs: Date.now() - followUpQueuedAt, reason: reasonText, + rollbackCursor: followUpRollbackCursor, + currentCursor, }, - 'Active agent ended a turn without any output after a queued follow-up', + 'Re-queueing queued follow-up after active agent ended without output', ); + clearPendingFollowUpDiagnostics(); + deps.queue.enqueueMessageCheck(chatJid, group.folder); + return true; }; deps.queue.setActivityTouch?.(chatJid, (meta) => { @@ -852,9 +888,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { try { await channel.editMessage(chatJid, progressMessageId, rendered); latestProgressRendered = rendered; - } catch { + } catch (err) { + logger.warn( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + progressMessageId, + err, + }, + 'Failed to edit tracked progress message; will recreate it on the next progress update', + ); clearProgressTicker(); progressMessageId = null; + latestProgressRendered = null; } }; @@ -885,7 +933,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }; const sendProgressMessage = async (text: string) => { - if (!text || text === latestProgressText) { + if (!text || (text === latestProgressText && progressMessageId)) { return; } @@ -1095,9 +1143,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); - clearPendingFollowUpDiagnostics(); deps.queue.setActivityTouch?.(chatJid, null); + const finalFollowUpRequeueReason = + followUpEndedWithoutOutputReason || + (followUpRestartRequested ? 'follow-up-no-output-timeout' : null); + if ( + finalFollowUpRequeueReason && + requeuePendingFollowUp(finalFollowUpRequeueReason) + ) { + return true; + } + + clearPendingFollowUpDiagnostics(); + if (hadError) { if ( finalOutputSentToUser ||