claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: resolve session race condition and follow-up turn output loss

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Eyejoker
2026-03-21 12:28:44 +09:00
parent 767ab20d76
commit 9b741be902
5 changed files with 549 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/agent-runner.test.ts
View File

@@ -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<void>((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' }),
);
});
});

13
src/agent-runner.ts
View File

@@ -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;
}

67
src/index.ts
View File

@@ -298,6 +298,8 @@ async function processGroupMessages(chatJid: string): Promise<boolean> {
let progressTicker: ReturnType<typeof setInterval> | 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<boolean> {
}
};
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<boolean> {
};
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<boolean> {
return;
}
latestModelProgressTextForFinalFallback = text;
latestProgressText = text;
if (progressStartedAt === null) {
progressStartedAt = Date.now();
@@ -372,6 +387,10 @@ async function processGroupMessages(chatJid: string): Promise<boolean> {
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<boolean> {
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<boolean> {
? 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<boolean> {
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);

317
src/message-runtime.test.ts
View File

@@ -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: '<internal>hidden final</internal>',
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');

110
src/message-runtime.ts
View File

@@ -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);