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

simplify runtime turn and progress handling

Browse Source
This commit is contained in:
Eyejoker
2026-03-23 02:19:48 +09:00
parent b83cc95963
commit 1a702d61e3
3 changed files with 313 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/group-queue.ts
View File

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

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

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

116
src/message-runtime.ts
View File

@@ -129,7 +129,43 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const deliverOpenWorkItem = async (
channel: Channel,
item: WorkItem,
options?: {
replaceMessageId?: string | null;
},
): Promise<boolean> => {
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<typeof setInterval> | 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<string | null> => {
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 ||