From 9b741be902880b82d2d6a81f34ea36aa4e2d48bd Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Sat, 21 Mar 2026 12:28:44 +0900 Subject: [PATCH] fix: resolve session race condition and follow-up turn output loss Two bugs fixed: 1. agent-runner error path race condition: when agent process exited with error, the promise resolved immediately without waiting for outputChain, allowing a late wrappedOnOutput to re-persist a stale session ID after clearSession had already run. 2. follow-up turn state not resetting: when a follow-up IPC message joined an active run, turn-level flags (finalOutputSentToUser, sawNonProgressOutput) from the first turn prevented the last progress from being promoted to a final message. Co-Authored-By: Claude Opus 4.6 --- src/agent-runner.test.ts | 50 ++++++ src/agent-runner.ts | 13 +- src/index.ts | 67 +++++++- src/message-runtime.test.ts | 317 +++++++++++++++++++++++++++++++++++- src/message-runtime.ts | 110 +++++++++++++ 5 files changed, 549 insertions(+), 8 deletions(-) diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 8df241d..24b41cc 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -252,4 +252,54 @@ describe('agent-runner timeout behavior', () => { }), ); }); + + it('waits for queued streamed output before resolving an error exit', async () => { + let releaseOutputs: (() => void) | undefined; + const outputsFlushed = new Promise((resolve) => { + releaseOutputs = resolve; + }); + const onOutput = vi.fn(async () => { + await outputsFlushed; + }); + const resultPromise = runAgentProcess( + testGroup, + testInput, + () => {}, + onOutput, + ); + + emitOutputMarker(fakeProc, { + status: 'error', + result: null, + error: 'No conversation found with session ID: stale', + }); + emitOutputMarker(fakeProc, { + status: 'error', + result: null, + error: 'Claude Code process exited with code 1', + newSessionId: 'stale-session', + }); + + await vi.advanceTimersByTimeAsync(10); + fakeProc.emit('close', 1); + + let settled = false; + resultPromise.then(() => { + settled = true; + }); + + await vi.advanceTimersByTimeAsync(10); + expect(settled).toBe(false); + + releaseOutputs?.(); + await vi.advanceTimersByTimeAsync(10); + + const result = await resultPromise; + expect(result.status).toBe('error'); + expect(onOutput).toHaveBeenCalledTimes(2); + expect(onOutput).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ newSessionId: 'stale-session' }), + ); + }); }); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 4c307f7..af61785 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -681,10 +681,15 @@ export async function runAgentProcess( }, 'Agent exited with error', ); - resolve({ - status: 'error', - result: null, - error: `Agent exited with code ${code}: ${stderr.slice(-200)}`, + // Wait for any queued streamed-output handlers to finish so a late + // newSessionId cannot be persisted after the caller clears a poisoned + // session. + outputChain.then(() => { + resolve({ + status: 'error', + result: null, + error: `Agent exited with code ${code}: ${stderr.slice(-200)}`, + }); }); return; } diff --git a/src/index.ts b/src/index.ts index f8f0653..e860581 100644 --- a/src/index.ts +++ b/src/index.ts @@ -298,6 +298,8 @@ async function processGroupMessages(chatJid: string): Promise { let progressTicker: ReturnType | null = null; let finalOutputSentToUser = false; let progressOutputSentToUser = false; + let latestModelProgressTextForFinalFallback: string | null = null; + let sawNonProgressOutput = false; let poisonedSessionDetected = false; const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; @@ -309,6 +311,14 @@ async function processGroupMessages(chatJid: string): Promise { } }; + const resetProgressState = () => { + clearProgressTicker(); + latestProgressText = null; + latestProgressRendered = null; + progressMessageId = null; + progressStartedAt = null; + }; + const renderProgressMessage = (text: string) => { const elapsedSeconds = progressStartedAt === null @@ -356,8 +366,12 @@ async function processGroupMessages(chatJid: string): Promise { }; const finalizeProgressMessage = async () => { + logger.info( + { group: group.name, chatJid, progressMessageId, latestProgressText }, + 'Finalizing tracked progress message', + ); await syncTrackedProgressMessage(); - clearProgressTicker(); + resetProgressState(); }; const sendProgressMessage = async (text: string) => { @@ -365,6 +379,7 @@ async function processGroupMessages(chatJid: string): Promise { return; } + latestModelProgressTextForFinalFallback = text; latestProgressText = text; if (progressStartedAt === null) { progressStartedAt = Date.now(); @@ -372,6 +387,10 @@ async function processGroupMessages(chatJid: string): Promise { const rendered = renderProgressMessage(text); if (progressMessageId && channel.editMessage) { + logger.info( + { group: group.name, chatJid, progressMessageId, text }, + 'Updating tracked progress message', + ); await syncTrackedProgressMessage(); progressOutputSentToUser = true; return; @@ -383,6 +402,10 @@ async function processGroupMessages(chatJid: string): Promise { progressMessageId = await channel.sendAndTrack(chatJid, rendered); if (progressMessageId) { + logger.info( + { group: group.name, chatJid, progressMessageId, text }, + 'Created tracked progress message', + ); latestProgressRendered = rendered; ensureProgressTicker(); progressOutputSentToUser = true; @@ -413,17 +436,41 @@ async function processGroupMessages(chatJid: string): Promise { ? result.result : JSON.stringify(result.result); const text = formatOutbound(raw); - logger.info({ group: group.name }, `Agent output: ${raw.slice(0, 200)}`); + logger.info( + { + group: group.name, + chatJid, + resultStatus: result.status, + resultPhase: result.phase, + progressMessageId, + }, + `Agent output: ${raw.slice(0, 200)}`, + ); if (result.phase === 'progress') { if (text) { await sendProgressMessage(text); } return; } + sawNonProgressOutput = true; if (text) { await finalizeProgressMessage(); await channel.sendMessage(chatJid, text); finalOutputSentToUser = true; + latestModelProgressTextForFinalFallback = null; + } else { + logger.info( + { + group: group.name, + chatJid, + resultStatus: result.status, + resultPhase: result.phase, + progressMessageId, + }, + 'Agent output became empty after formatting; resetting tracked progress state', + ); + await finalizeProgressMessage(); + latestModelProgressTextForFinalFallback = null; } } else { await finalizeProgressMessage(); @@ -450,6 +497,22 @@ async function processGroupMessages(chatJid: string): Promise { hadError = true; } + if ( + output === 'success' && + !hadError && + !finalOutputSentToUser && + !sawNonProgressOutput && + latestModelProgressTextForFinalFallback + ) { + logger.info( + { group: group.name, chatJid }, + 'Promoting last progress output to final message after agent completion', + ); + await channel.sendMessage(chatJid, latestModelProgressTextForFinalFallback); + finalOutputSentToUser = true; + latestModelProgressTextForFinalFallback = null; + } + clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer); diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 9752ce4..bd292b8 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -223,7 +223,7 @@ describe('createMessageRuntime', () => { expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-2'); }); - it('tracks Codex progress in one editable message and updates elapsed time every 10 seconds', async () => { + it('tracks Codex progress in one editable message and promotes the last progress when the run ends without a final phase', async () => { vi.useFakeTimers(); const chatJid = 'group@test'; const group = makeGroup('codex'); @@ -303,7 +303,10 @@ describe('createMessageRuntime', () => { 'progress-1', 'CI 상태 확인 중입니다.\n\n10초', ); - expect(channel.sendMessage).not.toHaveBeenCalled(); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'CI 상태 확인 중입니다.', + ); expect(notifyIdle).toHaveBeenCalledTimes(1); expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-progress'); expect(persistSession).toHaveBeenCalledWith( @@ -628,6 +631,316 @@ describe('createMessageRuntime', () => { } }); + it('resets tracked progress after a final output that becomes empty after formatting', 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(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '첫 번째 진행상황입니다.', + newSessionId: 'session-empty-final', + }); + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'hidden final', + newSessionId: 'session-empty-final', + }); + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '두 번째 진행상황입니다.', + newSessionId: 'session-empty-final', + }); + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-empty-final', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-empty-final', + }; + }, + ); + + 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-empty-final', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendAndTrack).toHaveBeenNthCalledWith( + 1, + chatJid, + '첫 번째 진행상황입니다.\n\n0초', + ); + expect(channel.sendAndTrack).toHaveBeenNthCalledWith( + 2, + chatJid, + '두 번째 진행상황입니다.\n\n0초', + ); + expect(channel.editMessage).toHaveBeenNthCalledWith( + 1, + chatJid, + 'progress-1', + '첫 번째 진행상황입니다.\n\n10초', + ); + expect(channel.editMessage).toHaveBeenNthCalledWith( + 2, + chatJid, + 'progress-2', + '두 번째 진행상황입니다.\n\n10초', + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '두 번째 진행상황입니다.', + ); + } finally { + vi.useRealTimers(); + } + }); + + it('promotes the last progress output to a final message when the agent completes without a final phase', 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'); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '검증 중입니다.', + newSessionId: 'session-progress-only', + }); + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '커밋은 정상 들어갔고 pre-commit도 통과했습니다.', + newSessionId: 'session-progress-only', + }); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-progress-only', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-progress-only', + }; + }, + ); + + 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-only', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendAndTrack).toHaveBeenCalledWith( + chatJid, + '검증 중입니다.\n\n0초', + ); + expect(channel.editMessage).toHaveBeenCalledWith( + chatJid, + 'progress-1', + '커밋은 정상 들어갔고 pre-commit도 통과했습니다.\n\n10초', + ); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '커밋은 정상 들어갔고 pre-commit도 통과했습니다.', + ); + } finally { + vi.useRealTimers(); + } + }); + + it('promotes progress-only output for a follow-up turn after a prior final in the same run', 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-follow-up'); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'final', + result: '첫 번째 턴 최종 답변', + newSessionId: 'session-follow-up', + }); + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '두 번째 턴 진행상황입니다.', + newSessionId: 'session-follow-up', + }); + await vi.advanceTimersByTimeAsync(10_000); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-follow-up', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-follow-up', + }; + }, + ); + + 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-follow-up-progress-only', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendMessage).toHaveBeenNthCalledWith( + 1, + chatJid, + '첫 번째 턴 최종 답변', + ); + expect(channel.sendAndTrack).toHaveBeenCalledWith( + chatJid, + '두 번째 턴 진행상황입니다.\n\n0초', + ); + expect(channel.editMessage).toHaveBeenCalledWith( + chatJid, + 'progress-follow-up', + '두 번째 턴 진행상황입니다.\n\n10초', + ); + expect(channel.sendMessage).toHaveBeenNthCalledWith( + 2, + 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 253c06f..bd3ca7b 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -337,6 +337,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { let hadError = false; let finalOutputSentToUser = false; let progressOutputSentToUser = false; + let latestModelProgressTextForFinalFallback: string | null = null; + let sawNonProgressOutput = false; let poisonedSessionDetected = false; const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; @@ -348,6 +350,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } }; + const resetTurnOutputState = () => { + finalOutputSentToUser = false; + progressOutputSentToUser = false; + latestModelProgressTextForFinalFallback = null; + sawNonProgressOutput = false; + }; + const resetProgressState = () => { clearProgressTicker(); latestProgressText = null; @@ -404,6 +413,17 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }; const finalizeProgressMessage = async () => { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + progressMessageId, + latestProgressText, + }, + 'Finalizing tracked progress message', + ); await syncTrackedProgressMessage(); resetProgressState(); }; @@ -414,12 +434,29 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } if (progressStartedAt === null) { + // A fresh progress cycle after a prior final in the same active agent + // session represents a new logical turn (follow-up IPC message). Reset + // turn-scoped output flags so progress-only completion can still be + // promoted to a final message. + resetTurnOutputState(); progressStartedAt = Date.now(); } + latestModelProgressTextForFinalFallback = text; latestProgressText = text; const rendered = renderProgressMessage(text); if (progressMessageId && channel.editMessage) { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + progressMessageId, + text, + }, + 'Updating tracked progress message', + ); await syncTrackedProgressMessage(); progressOutputSentToUser = true; return; @@ -428,6 +465,17 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (channel.sendAndTrack) { progressMessageId = await channel.sendAndTrack(chatJid, rendered); if (progressMessageId) { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + progressMessageId, + text, + }, + 'Created tracked progress message', + ); latestProgressRendered = rendered; ensureProgressTicker(); progressOutputSentToUser = true; @@ -479,20 +527,61 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { groupFolder: group.folder, runId, resultStatus: result.status, + resultPhase: result.phase, + progressMessageId, }, `Agent output: ${raw.slice(0, 200)}`, ); if (result.phase === 'progress') { + // Detect new logical turn (follow-up IPC): if a final was already + // sent in this run, a new progress output means a follow-up turn + // has started. Reset turn-level state so the promotion logic at + // end-of-run works correctly for the new turn. + if (finalOutputSentToUser || sawNonProgressOutput) { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + }, + 'New logical turn detected (follow-up), resetting turn state', + ); + finalOutputSentToUser = false; + sawNonProgressOutput = false; + progressOutputSentToUser = false; + latestModelProgressTextForFinalFallback = null; + resetProgressState(); + await channel.setTyping?.(chatJid, true); + } if (text) { await sendProgressMessage(text); } return; } + sawNonProgressOutput = true; + if (text) { await finalizeProgressMessage(); await channel.sendMessage(chatJid, text); finalOutputSentToUser = true; + latestModelProgressTextForFinalFallback = null; + } else { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + resultStatus: result.status, + resultPhase: result.phase, + progressMessageId, + }, + 'Agent output became empty after formatting; resetting tracked progress state', + ); + await finalizeProgressMessage(); + latestModelProgressTextForFinalFallback = null; } } else { await finalizeProgressMessage(); @@ -520,6 +609,27 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { hadError = true; } + if ( + output === 'success' && + !hadError && + !finalOutputSentToUser && + !sawNonProgressOutput && + latestModelProgressTextForFinalFallback + ) { + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + }, + 'Promoting last progress output to final message after agent completion', + ); + await channel.sendMessage(chatJid, latestModelProgressTextForFinalFallback); + finalOutputSentToUser = true; + latestModelProgressTextForFinalFallback = null; + } + clearProgressTicker(); if (idleTimer) clearTimeout(idleTimer);