diff --git a/src/group-queue.ts b/src/group-queue.ts index 042f7a3..4e888a4 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -42,6 +42,7 @@ interface GroupState { retryScheduledAt: number | null; startedAt: number | null; activityTouch: ((meta?: GroupActivityTouchMeta) => void) | null; + followUpPipeAllowed: (() => boolean) | null; } export interface GroupStatus { @@ -81,6 +82,7 @@ export class GroupQueue { retryScheduledAt: null, startedAt: null, activityTouch: null, + followUpPipeAllowed: null, }; this.groups.set(groupJid, state); } @@ -101,6 +103,14 @@ export class GroupQueue { state.activityTouch = touch; } + setFollowUpPipeAllowed( + groupJid: string, + allowed: (() => boolean) | null, + ): void { + const state = this.getGroup(groupJid); + state.followUpPipeAllowed = allowed; + } + private createRunId(): string { return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; } @@ -117,6 +127,12 @@ export class GroupQueue { if (state.active) { state.pendingMessages = true; + if (state.idleWaiting) { + this.closeStdin(groupJid, { + runId: state.currentRunId ?? undefined, + reason: 'pending-message-preemption', + }); + } logger.debug({ groupJid }, 'Agent active, message queued'); return; } @@ -268,6 +284,13 @@ export class GroupQueue { ); return false; } + if (state.followUpPipeAllowed && !state.followUpPipeAllowed()) { + logger.info( + { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder }, + 'Skipping follow-up IPC because active agent requires a fresh logical run', + ); + return false; + } state.idleWaiting = false; // Agent is about to receive work, no longer idle const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input'); @@ -412,6 +435,7 @@ export class GroupQueue { state.groupFolder = null; state.currentRunId = null; state.activityTouch = null; + state.followUpPipeAllowed = null; this.activeCount--; this.drainGroup(groupJid); } @@ -447,6 +471,7 @@ export class GroupQueue { state.processName = null; state.groupFolder = null; state.activityTouch = null; + state.followUpPipeAllowed = null; this.activeCount--; this.drainGroup(groupJid); } diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index dd4926e..13c98cb 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -335,7 +335,10 @@ describe('createMessageRuntime', () => { expect(result).toBe(true); expect(clearSession).not.toHaveBeenCalled(); - expect(closeStdin).not.toHaveBeenCalled(); + expect(closeStdin).toHaveBeenCalledWith(chatJid, { + runId: 'run-2', + reason: 'output-delivered-close', + }); expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-2'); }); @@ -419,10 +422,12 @@ describe('createMessageRuntime', () => { 'progress-1', 'CI 상태 확인 중입니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenCalledWith( + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-1', 'CI 상태 확인 중입니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); expect(notifyIdle).toHaveBeenCalledTimes(1); expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-progress'); expect(persistSession).toHaveBeenCalledWith( @@ -434,6 +439,152 @@ describe('createMessageRuntime', () => { } }); + it('falls back to a plain progress message when tracked progress creation throws', async () => { + 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!).mockRejectedValueOnce( + new Error('discord send failed'), + ); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '진행 중입니다.', + newSessionId: 'session-progress-fallback', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-progress-fallback', + }; + }, + ); + + 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(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-progress-fallback-throw', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendAndTrack).toHaveBeenCalledWith( + chatJid, + '진행 중입니다.\n\n0초', + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '진행 중입니다.\n\n0초', + ); + }); + + it('falls back to a plain progress message when tracked progress creation returns null', async () => { + 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(null as any); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '진행 중입니다.', + newSessionId: 'session-progress-null-fallback', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-progress-null-fallback', + }; + }, + ); + + 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(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-progress-fallback-null', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendAndTrack).toHaveBeenCalledWith( + chatJid, + '진행 중입니다.\n\n0초', + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '진행 중입니다.\n\n0초', + ); + }); + it('resets the idle timeout when follow-up Codex progress keeps arriving', async () => { vi.useFakeTimers(); const chatJid = 'group@test'; @@ -460,7 +611,6 @@ describe('createMessageRuntime', () => { newSessionId: 'session-follow-up', }); await vi.advanceTimersByTimeAsync(800); - expect(closeStdin).not.toHaveBeenCalled(); await onOutput?.({ status: 'success', @@ -469,7 +619,6 @@ describe('createMessageRuntime', () => { newSessionId: 'session-follow-up', }); await vi.advanceTimersByTimeAsync(800); - expect(closeStdin).not.toHaveBeenCalled(); await onOutput?.({ status: 'success', @@ -478,7 +627,6 @@ describe('createMessageRuntime', () => { newSessionId: 'session-follow-up', }); await vi.advanceTimersByTimeAsync(800); - expect(closeStdin).not.toHaveBeenCalled(); await onOutput?.({ status: 'success', @@ -523,7 +671,10 @@ describe('createMessageRuntime', () => { }); expect(result).toBe(true); - expect(closeStdin).not.toHaveBeenCalled(); + expect(closeStdin).toHaveBeenCalledWith(chatJid, { + runId: 'run-follow-up-progress', + reason: 'output-delivered-close', + }); expect(channel.sendMessage).toHaveBeenCalledWith( chatJid, '초기 응답입니다.', @@ -721,10 +872,12 @@ describe('createMessageRuntime', () => { 'progress-1', '테스트를 돌리는 중입니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenCalledWith( + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-1', '테스트가 끝났습니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); expect(notifyIdle).toHaveBeenCalledTimes(1); expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-final'); } finally { @@ -837,19 +990,22 @@ describe('createMessageRuntime', () => { expect(channel.editMessage).toHaveBeenNthCalledWith( 2, chatJid, + 'progress-1', + '첫 번째 결과입니다.', + ); + expect(channel.editMessage).toHaveBeenNthCalledWith( + 3, + chatJid, 'progress-2', '두 번째 진행상황입니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenNthCalledWith( - 1, - chatJid, - '첫 번째 결과입니다.', - ); - expect(channel.sendMessage).toHaveBeenNthCalledWith( - 2, + expect(channel.editMessage).toHaveBeenNthCalledWith( + 4, chatJid, + 'progress-2', '두 번째 결과입니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); } @@ -962,10 +1118,12 @@ describe('createMessageRuntime', () => { 'progress-2', '두 번째 진행상황입니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenCalledWith( + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-2', '두 번째 진행상황입니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); } @@ -1056,10 +1214,12 @@ describe('createMessageRuntime', () => { 'progress-1', '커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenCalledWith( + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-1', '커밋은 정상 들어갔고 pre-commit도 통과했습니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); } @@ -1157,11 +1317,12 @@ describe('createMessageRuntime', () => { 'progress-follow-up', '두 번째 턴 진행상황입니다.\n\n10초', ); - expect(channel.sendMessage).toHaveBeenNthCalledWith( - 2, + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-follow-up', '두 번째 턴 진행상황입니다.', ); + expect(channel.sendMessage).toHaveBeenCalledTimes(1); } finally { vi.useRealTimers(); } @@ -1258,15 +1419,13 @@ describe('createMessageRuntime', () => { 'progress-1', '진행 중입니다.\n\n10초', ); - expect(channel.sendAndTrack).toHaveBeenNthCalledWith( - 2, - chatJid, - '진행 중입니다.\n\n10초', - ); - expect(channel.sendMessage).toHaveBeenCalledWith( + expect(channel.sendAndTrack).toHaveBeenCalledTimes(1); + expect(channel.editMessage).toHaveBeenCalledWith( chatJid, + 'progress-1', '진행 중입니다.', ); + expect(channel.sendMessage).not.toHaveBeenCalled(); } finally { vi.useRealTimers(); } diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 9ca9989..12ad8c7 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -129,7 +129,43 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const deliverOpenWorkItem = async ( channel: Channel, item: WorkItem, + options?: { + replaceMessageId?: string | null; + }, ): Promise => { + const replaceMessageId = options?.replaceMessageId ?? null; + try { + if (replaceMessageId && channel.editMessage) { + await channel.editMessage( + item.chat_jid, + replaceMessageId, + item.result_payload, + ); + markWorkItemDelivered(item.id, replaceMessageId); + logger.info( + { + chatJid: item.chat_jid, + workItemId: item.id, + deliveryAttempts: item.delivery_attempts + 1, + replacedMessageId: replaceMessageId, + }, + 'Delivered produced work item by replacing tracked progress message', + ); + return true; + } + } catch (err) { + logger.warn( + { + chatJid: item.chat_jid, + workItemId: item.id, + deliveryAttempts: item.delivery_attempts + 1, + replacedMessageId: replaceMessageId, + err, + }, + 'Failed to replace tracked progress message; falling back to a new message', + ); + } + try { await channel.sendMessage(item.chat_jid, item.result_payload); markWorkItemDelivered(item.id); @@ -637,6 +673,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let progressMessageId: string | null = null; let progressStartedAt: number | null = null; let progressTicker: ReturnType | null = null; + let progressEditFailCount = 0; + let promotableTrackedProgressMessageId: string | null = null; let finalOutputSentToUser = false; let progressOutputSentToUser = false; let latestModelProgressTextForFinalFallback: string | null = null; @@ -644,6 +682,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let producedDeliverySucceeded = true; let isFirstLogicalTurn = true; let poisonedSessionDetected = false; + let canPipeFollowUps = true; const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; const FOLLOW_UP_NO_OUTPUT_WARN_MS = 10_000; @@ -801,6 +840,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { noteFollowUpQueued(meta); resetIdleTimer(); }); + deps.queue.setFollowUpPipeAllowed?.(chatJid, () => canPipeFollowUps); const clearProgressTicker = () => { if (progressTicker) { @@ -817,7 +857,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { producedFinalText = null; }; - const flushProducedFinalText = async () => { + const flushProducedFinalText = async ( + replaceMessageId?: string | null, + ) => { if (!producedFinalText) { return; } @@ -831,7 +873,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { end_seq: isFirstLogicalTurn ? endSeq : null, result_payload: producedFinalText, }); - const delivered = await deliverOpenWorkItem(channel, workItem); + const delivered = await deliverOpenWorkItem(channel, workItem, { + replaceMessageId, + }); if (delivered) { finalOutputSentToUser = true; } else { @@ -846,6 +890,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } finally { producedFinalText = null; latestModelProgressTextForFinalFallback = null; + promotableTrackedProgressMessageId = null; isFirstLogicalTurn = false; } }; @@ -856,6 +901,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { latestProgressRendered = null; progressMessageId = null; progressStartedAt = null; + progressEditFailCount = 0; }; const renderProgressMessage = (text: string) => { @@ -888,7 +934,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { try { await channel.editMessage(chatJid, progressMessageId, rendered); latestProgressRendered = rendered; + progressEditFailCount = 0; } catch (err) { + progressEditFailCount++; logger.warn( { chatJid, @@ -896,13 +944,16 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { groupFolder: group.folder, runId, progressMessageId, + progressEditFailCount, err, }, - 'Failed to edit tracked progress message; will recreate it on the next progress update', + 'Failed to edit tracked progress message; will retry before recreating', ); - clearProgressTicker(); - progressMessageId = null; latestProgressRendered = null; + if (progressEditFailCount >= 3) { + clearProgressTicker(); + progressMessageId = null; + } } }; @@ -916,7 +967,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 10_000); }; - const finalizeProgressMessage = async () => { + const finalizeProgressMessage = async (options?: { + preserveTrackedMessage?: boolean; + }): Promise => { logger.info( { chatJid, @@ -928,8 +981,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 'Finalizing tracked progress message', ); - await syncTrackedProgressMessage(); + const trackedMessageId = progressMessageId; + if (!options?.preserveTrackedMessage) { + await syncTrackedProgressMessage(); + } resetProgressState(); + return trackedMessageId; }; const sendProgressMessage = async (text: string) => { @@ -963,6 +1020,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } if (!channel.sendAndTrack) { + latestProgressRendered = rendered; + await channel.sendMessage(chatJid, rendered); + progressOutputSentToUser = true; return; } @@ -979,6 +1039,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 'Failed to send tracked progress message', ); + latestProgressRendered = rendered; + await channel.sendMessage(chatJid, rendered); + progressOutputSentToUser = true; return; } @@ -997,7 +1060,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { latestProgressRendered = rendered; ensureProgressTicker(); progressOutputSentToUser = true; + return; } + + latestProgressRendered = rendered; + await channel.sendMessage(chatJid, rendered); + progressOutputSentToUser = true; }; await channel.setTyping?.(chatJid, true); @@ -1032,6 +1100,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ? result.result : JSON.stringify(result.result); const text = formatOutbound(raw); + if (text) { + canPipeFollowUps = false; + } logger.info( { chatJid, @@ -1073,9 +1144,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { sawNonProgressOutput = true; if (text) { - await finalizeProgressMessage(); + const trackedProgressMessageId = await finalizeProgressMessage({ + preserveTrackedMessage: !!( + progressMessageId && channel.editMessage + ), + }); producedFinalText = text; - await flushProducedFinalText(); + await flushProducedFinalText(trackedProgressMessageId); } else { logger.info( { @@ -1096,7 +1171,14 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (result.status === 'success') { warnFollowUpEndedWithoutOutput('success-null-result'); } - await finalizeProgressMessage(); + promotableTrackedProgressMessageId = await finalizeProgressMessage({ + preserveTrackedMessage: !!( + !sawNonProgressOutput && + latestModelProgressTextForFinalFallback && + progressMessageId && + channel.editMessage + ), + }); } await channel.setTyping?.(chatJid, false); @@ -1106,6 +1188,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (result.status === 'success' && !poisonedSessionDetected) { deps.queue.notifyIdle(chatJid, runId); + if (finalOutputSentToUser || progressOutputSentToUser) { + deps.queue.closeStdin(chatJid, { + runId, + reason: 'output-delivered-close', + }); + } } if (result.status === 'error') { @@ -1136,14 +1224,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, 'Promoting last progress output to final message after agent completion', ); + const trackedProgressMessageId = await finalizeProgressMessage({ + preserveTrackedMessage: !!(progressMessageId && channel.editMessage), + }); producedFinalText = latestModelProgressTextForFinalFallback; - await flushProducedFinalText(); + await flushProducedFinalText( + trackedProgressMessageId || promotableTrackedProgressMessageId, + ); } clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); deps.queue.setActivityTouch?.(chatJid, null); + deps.queue.setFollowUpPipeAllowed?.(chatJid, null); const finalFollowUpRequeueReason = followUpEndedWithoutOutputReason ||