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

refactor: remove active follow-up piping

Browse Source
This commit is contained in:
Eyejoker
2026-03-23 04:13:53 +09:00
parent 3435b99cfc
commit 1974f8a6dc
4 changed files with 23 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
src/group-queue.test.ts
View File

@@ -8,7 +8,7 @@ vi.mock('./config.js', () => ({
MAX_CONCURRENT_AGENTS: 2,
}));
// Mock fs operations used by sendMessage/closeStdin
// Mock fs operations used by closeStdin
vi.mock('fs', async () => {
const actual = await vi.importActual<typeof import('fs')>('fs');
return {
@@ -409,7 +409,7 @@ describe('GroupQueue', () => {
await vi.advanceTimersByTimeAsync(10);
});
it('keeps a post-final follow-up queued instead of preempting the idle run', async () => {
it('preempts an idle run so the next message is handled in a fresh run', async () => {
const fs = await import('fs');
let resolveProcess: () => void;
@@ -425,7 +425,6 @@ describe('GroupQueue', () => {
await vi.advanceTimersByTimeAsync(10);
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
queue.setFollowUpPipeAllowed('group1@g.us', () => false);
queue.notifyIdle('group1@g.us');
const writeFileSync = vi.mocked(fs.default.writeFileSync);
@@ -436,119 +435,7 @@ describe('GroupQueue', () => {
const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(0);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
expect(processMessages).toHaveBeenCalledTimes(2);
});
it('sendMessage resets idleWaiting so a subsequent task enqueue does not preempt', async () => {
const fs = await import('fs');
let resolveProcess: () => void;
const processMessages = vi.fn(async () => {
await new Promise<void>((resolve) => {
resolveProcess = resolve;
});
return true;
});
queue.setProcessMessagesFn(processMessages);
queue.enqueueMessageCheck('group1@g.us');
await vi.advanceTimersByTimeAsync(10);
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
// Agent becomes idle
queue.notifyIdle('group1@g.us');
// A new user message arrives — resets idleWaiting
queue.sendMessage('group1@g.us', 'hello');
// Task enqueued after message reset — should NOT preempt (agent is working)
const writeFileSync = vi.mocked(fs.default.writeFileSync);
writeFileSync.mockClear();
const taskFn = vi.fn(async () => {});
queue.enqueueTask('group1@g.us', 'task-1', taskFn);
const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(0);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
it('sendMessage touches active run activity after piping follow-up', async () => {
let resolveProcess: () => void;
const touch = vi.fn();
const processMessages = vi.fn(async () => {
await new Promise<void>((resolve) => {
resolveProcess = resolve;
});
return true;
});
queue.setProcessMessagesFn(processMessages);
queue.enqueueMessageCheck('group1@g.us');
await vi.advanceTimersByTimeAsync(10);
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
queue.setActivityTouch('group1@g.us', touch);
expect(queue.sendMessage('group1@g.us', 'hello')).toBe(true);
expect(touch).toHaveBeenCalledTimes(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
it('sendMessage returns false for task agents so user messages queue up', async () => {
let resolveTask: () => void;
const taskFn = vi.fn(async () => {
await new Promise<void>((resolve) => {
resolveTask = resolve;
});
});
// Start a task (sets isTaskProcess = true)
queue.enqueueTask('group1@g.us', 'task-1', taskFn);
await vi.advanceTimersByTimeAsync(10);
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
// sendMessage should return false — user messages must not go to task agents
const result = queue.sendMessage('group1@g.us', 'hello');
expect(result).toBe(false);
resolveTask!();
await vi.advanceTimersByTimeAsync(10);
});
it('does not pipe follow-up messages to an agent after closeStdin', async () => {
let resolveProcess: () => void;
const processMessages = vi.fn(async () => {
await new Promise<void>((resolve) => {
resolveProcess = resolve;
});
return true;
});
queue.setProcessMessagesFn(processMessages);
queue.enqueueMessageCheck('group1@g.us', 'test-group');
await vi.advanceTimersByTimeAsync(10);
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
queue.notifyIdle('group1@g.us');
queue.closeStdin('group1@g.us');
expect(queue.sendMessage('group1@g.us', 'hello after clear')).toBe(false);
queue.enqueueMessageCheck('group1@g.us', 'test-group');
expect(closeWrites).toHaveLength(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);

114
src/group-queue.ts
View File

@@ -11,12 +11,6 @@ interface QueuedTask {
fn: () => Promise<void>;
}
export interface GroupActivityTouchMeta {
source: 'follow-up';
textLength: number;
filename: string;
}
export interface GroupRunContext {
runId: string;
reason: 'messages' | 'drain';
@@ -41,8 +35,6 @@ interface GroupState {
retryTimer: ReturnType<typeof setTimeout> | null;
retryScheduledAt: number | null;
startedAt: number | null;
activityTouch: ((meta?: GroupActivityTouchMeta) => void) | null;
followUpPipeAllowed: (() => boolean) | null;
}
export interface GroupStatus {
@@ -81,8 +73,6 @@ export class GroupQueue {
retryTimer: null,
retryScheduledAt: null,
startedAt: null,
activityTouch: null,
followUpPipeAllowed: null,
};
this.groups.set(groupJid, state);
}
@@ -95,32 +85,8 @@ export class GroupQueue {
this.processMessagesFn = fn;
}
setActivityTouch(
groupJid: string,
touch: ((meta?: GroupActivityTouchMeta) => void) | null,
): void {
const state = this.getGroup(groupJid);
state.activityTouch = touch;
}
setFollowUpPipeAllowed(
groupJid: string,
allowed: (() => boolean) | null,
): void {
const state = this.getGroup(groupJid);
state.followUpPipeAllowed = allowed;
}
private canPreemptIdleActiveRun(state: GroupState): boolean {
if (!state.idleWaiting) {
return false;
}
if (!state.followUpPipeAllowed) {
return true;
}
return state.followUpPipeAllowed();
return state.idleWaiting;
}
private createRunId(): string {
@@ -269,80 +235,6 @@ export class GroupQueue {
}
}
/**
* Send a follow-up message to the active agent process via IPC file.
* Returns true if the message was written, false if no active agent process.
*/
sendMessage(groupJid: string, text: string): boolean {
const state = this.getGroup(groupJid);
if (!state.active || !state.groupFolder || state.isTaskProcess) {
logger.debug(
{
groupJid,
runId: state.currentRunId,
active: state.active,
closingStdin: state.closingStdin,
groupFolder: state.groupFolder,
isTaskProcess: state.isTaskProcess,
},
'Cannot pipe follow-up message to active agent',
);
return false;
}
if (state.closingStdin) {
logger.info(
{ groupJid, runId: state.currentRunId, groupFolder: state.groupFolder },
'Skipping follow-up IPC because active agent is closing',
);
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');
try {
fs.mkdirSync(inputDir, { recursive: true });
const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 6)}.json`;
const filepath = path.join(inputDir, filename);
const tempPath = `${filepath}.tmp`;
fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text }));
fs.renameSync(tempPath, filepath);
state.activityTouch?.({
source: 'follow-up',
textLength: text.length,
filename,
});
logger.info(
{
groupJid,
runId: state.currentRunId,
groupFolder: state.groupFolder,
textLength: text.length,
filename,
},
'Queued follow-up message for active agent',
);
return true;
} catch (err) {
logger.warn(
{
groupJid,
runId: state.currentRunId,
groupFolder: state.groupFolder,
err,
},
'Failed to queue follow-up message for active agent',
);
return false;
}
}
/**
* Signal the active agent process to wind down by writing a close sentinel.
*/
@@ -446,8 +338,6 @@ export class GroupQueue {
state.processName = null;
state.groupFolder = null;
state.currentRunId = null;
state.activityTouch = null;
state.followUpPipeAllowed = null;
this.activeCount--;
this.drainGroup(groupJid);
}
@@ -482,8 +372,6 @@ export class GroupQueue {
state.process = null;
state.processName = null;
state.groupFolder = null;
state.activityTouch = null;
state.followUpPipeAllowed = null;
this.activeCount--;
this.drainGroup(groupJid);
}

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

@@ -335,7 +335,7 @@ describe('createMessageRuntime', () => {
expect(result).toBe(true);
expect(clearSession).not.toHaveBeenCalled();
expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-2');
expect(notifyIdle).not.toHaveBeenCalled();
expect(closeStdin).toHaveBeenCalledWith(chatJid, {
runId: 'run-2',
reason: 'output-delivered-close',
@@ -432,8 +432,7 @@ describe('createMessageRuntime', () => {
chatJid,
'CI 상태 확인 중입니다.',
);
expect(notifyIdle).toHaveBeenCalledTimes(1);
expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-progress');
expect(notifyIdle).not.toHaveBeenCalled();
expect(persistSession).toHaveBeenCalledWith(
group.folder,
'session-progress',
@@ -889,8 +888,7 @@ describe('createMessageRuntime', () => {
chatJid,
'테스트가 끝났습니다.',
);
expect(notifyIdle).toHaveBeenCalledTimes(1);
expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-final');
expect(notifyIdle).not.toHaveBeenCalled();
} finally {
vi.useRealTimers();
}
@@ -956,129 +954,6 @@ describe('createMessageRuntime', () => {
expect(channel.sendAndTrack).not.toHaveBeenCalled();
});
it('does not emit a synthetic failure final when follow-up activity happens during a silent run', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
const lastAgentTimestamps: Record<string, string> = {};
const saveState = vi.fn();
let activityTouch:
| ((meta?: {
source: 'follow-up';
textLength: number;
filename: string;
}) => void)
| null = null;
let followUpPersisted = false;
vi.mocked(db.getMessagesSince).mockImplementation(
(_chatJid, sinceSeqCursor) => {
if (String(sinceSeqCursor || '0') === '0') {
return followUpPersisted
? [
{
id: 'msg-1',
chat_jid: chatJid,
sender: 'user@test',
sender_name: 'User',
content: 'hello',
timestamp: '2026-03-19T00:00:00.000Z',
seq: 1,
},
{
id: 'msg-2',
chat_jid: chatJid,
sender: 'codex@test',
sender_name: 'Codex',
content: 'follow-up',
timestamp: '2026-03-19T00:00:05.000Z',
seq: 2,
},
]
: [
{
id: 'msg-1',
chat_jid: chatJid,
sender: 'user@test',
sender_name: 'User',
content: 'hello',
timestamp: '2026-03-19T00:00:00.000Z',
seq: 1,
},
];
}
if (String(sinceSeqCursor || '0') === '1' && followUpPersisted) {
return [
{
id: 'msg-2',
chat_jid: chatJid,
sender: 'codex@test',
sender_name: 'Codex',
content: 'follow-up',
timestamp: '2026-03-19T00:00:05.000Z',
seq: 2,
},
];
}
return [];
},
);
vi.mocked(agentRunner.runAgentProcess).mockImplementationOnce(async () => {
lastAgentTimestamps[chatJid] = '2';
followUpPersisted = true;
activityTouch?.({
source: 'follow-up',
textLength: 9,
filename: 'follow-up.json',
});
return {
status: 'success',
result: null,
newSessionId: 'session-silent-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(),
setActivityTouch: vi.fn(
(_jid, touch) => (activityTouch = touch as typeof activityTouch),
),
setFollowUpPipeAllowed: vi.fn(),
} as any,
getRegisteredGroups: () => ({ [chatJid]: group }),
getSessions: () => ({}),
getLastTimestamp: () => '',
setLastTimestamp: vi.fn(),
getLastAgentTimestamps: () => lastAgentTimestamps,
saveState,
persistSession: vi.fn(),
clearSession: vi.fn(),
});
const result = await runtime.processGroupMessages(chatJid, {
runId: 'run-silent-follow-up',
reason: 'messages',
});
expect(result).toBe(true);
expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1);
expect(saveState).toHaveBeenCalled();
expect(lastAgentTimestamps[chatJid]).toBe('2');
expect(channel.sendMessage).not.toHaveBeenCalled();
});
it('resets tracked progress after a final output that becomes empty after formatting', async () => {
vi.useFakeTimers();
const chatJid = 'group@test';
@@ -1301,111 +1176,6 @@ describe('createMessageRuntime', () => {
}
});
it('ignores follow-up activity after a final was already observed in the same run', async () => {
vi.useFakeTimers();
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
const closeStdin = vi.fn();
const enqueueMessageCheck = vi.fn();
const lastAgentTimestamps: Record<string, string> = { [chatJid]: '1' };
let activityTouch:
| ((meta?: {
source: 'follow-up';
textLength: number;
filename: string;
}) => void)
| null = null;
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-final-boundary',
});
activityTouch?.({
source: 'follow-up',
textLength: 32,
filename: 'late-follow-up.json',
});
lastAgentTimestamps[chatJid] = '3';
await onOutput?.({
status: 'success',
result: null,
newSessionId: 'session-final-boundary',
});
return {
status: 'success',
result: null,
newSessionId: 'session-final-boundary',
};
},
);
const runtime = createMessageRuntime({
assistantName: 'Andy',
idleTimeout: 1_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: vi.fn(),
persistSession: vi.fn(),
clearSession: vi.fn(),
});
try {
const result = await runtime.processGroupMessages(chatJid, {
runId: 'run-final-boundary',
reason: 'messages',
});
expect(result).toBe(true);
expect(closeStdin).toHaveBeenCalledWith(chatJid, {
runId: 'run-final-boundary',
reason: 'output-delivered-close',
});
expect(channel.sendMessage).toHaveBeenCalledWith(
chatJid,
'첫 번째 턴 최종 답변',
);
expect(channel.sendMessage).toHaveBeenCalledTimes(1);
expect(channel.sendAndTrack).not.toHaveBeenCalled();
expect(enqueueMessageCheck).not.toHaveBeenCalled();
expect(lastAgentTimestamps[chatJid]).toBe('3');
} finally {
vi.useRealTimers();
}
});
it('retries editing progress message instead of creating a duplicate when edit fails', async () => {
vi.useFakeTimers();
const chatJid = 'group@test';

157
src/message-runtime.ts
View File

@@ -664,11 +664,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
);
type VisiblePhase = 'silent' | 'progress' | 'final';
type PendingFollowUp = {
queuedAt: number;
textLength: number;
filename: string;
} | null;
let idleTimer: ReturnType<typeof setTimeout> | null = null;
let hadError = false;
let producedDeliverySucceeded = true;
@@ -681,8 +676,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
let progressEditFailCount = 0;
let latestProgressTextForFinal: string | null = null;
let poisonedSessionDetected = false;
let canPipeFollowUps = true;
let pendingFollowUp: PendingFollowUp = null;
let closeRequested = false;
const hasVisibleOutput = () => visiblePhase !== 'silent';
const terminalObserved = () => visiblePhase === 'final';
@@ -703,10 +697,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
progressEditFailCount = 0;
};
const clearPendingFollowUp = () => {
pendingFollowUp = null;
};
const renderProgressMessage = (text: string) => {
const elapsedSeconds =
progressStartedAt === null
@@ -830,6 +820,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
await deliverFinalText(FAILURE_FINAL_TEXT);
};
const requestAgentClose = (reason: string) => {
if (closeRequested) return;
closeRequested = true;
deps.queue.closeStdin(chatJid, { runId, reason });
};
const sendProgressMessage = async (text: string) => {
if (!text || (text === latestProgressText && progressMessageId)) {
return;
@@ -914,84 +910,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
idleTimer = null;
return;
}
idleTimer = setTimeout(
() => {
const closeReason =
!hasVisibleOutput() && pendingFollowUp
? 'follow-up-no-output-preemption'
: 'idle-timeout';
if (!hasVisibleOutput() && pendingFollowUp) {
logger.warn(
{
group: group.name,
chatJid,
runId,
followUpQueuedAt: new Date(
pendingFollowUp.queuedAt,
).toISOString(),
followUpQueuedTextLength: pendingFollowUp.textLength,
followUpQueuedFilename: pendingFollowUp.filename,
followUpWaitMs: Date.now() - pendingFollowUp.queuedAt,
},
'Idle timeout reached while a queued follow-up still had no visible output',
);
} else {
logger.debug(
{ group: group.name, chatJid, runId, closeReason },
'Idle timeout, closing agent stdin',
);
}
deps.queue.closeStdin(chatJid, { runId, reason: closeReason });
},
deps.idleTimeout,
);
idleTimer = setTimeout(() => {
logger.debug(
{ group: group.name, chatJid, runId },
'Idle timeout, closing agent stdin',
);
requestAgentClose('idle-timeout');
}, deps.idleTimeout);
};
const noteFollowUpQueued = (meta?: {
source: 'follow-up';
textLength: number;
filename: string;
}) => {
if (meta?.source !== 'follow-up' || terminalObserved()) return;
pendingFollowUp = {
queuedAt: Date.now(),
textLength: meta.textLength,
filename: meta.filename,
};
logger.info(
{
group: group.name,
chatJid,
runId,
followUpQueuedTextLength: pendingFollowUp.textLength,
followUpQueuedFilename: pendingFollowUp.filename,
},
'Registered follow-up activity for active agent',
);
resetIdleTimer();
};
const noteVisibleOutputObserved = (phase?: string) => {
if (!pendingFollowUp) return;
logger.info(
{
group: group.name,
chatJid,
runId,
followUpQueuedAt: new Date(pendingFollowUp.queuedAt).toISOString(),
followUpQueuedTextLength: pendingFollowUp.textLength,
followUpQueuedFilename: pendingFollowUp.filename,
followUpWaitMs: Date.now() - pendingFollowUp.queuedAt,
resultPhase: phase,
},
'Agent produced visible output after a queued follow-up',
);
clearPendingFollowUp();
};
deps.queue.setActivityTouch?.(chatJid, noteFollowUpQueued);
deps.queue.setFollowUpPipeAllowed?.(chatJid, () => canPipeFollowUps);
resetIdleTimer();
await channel.setTyping?.(chatJid, true);
@@ -1059,8 +986,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
if (result.phase === 'progress') {
if (text) {
canPipeFollowUps = false;
noteVisibleOutputObserved(result.phase);
await sendProgressMessage(text);
}
if (!poisonedSessionDetected) {
@@ -1073,8 +998,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
}
if (text) {
canPipeFollowUps = false;
noteVisibleOutputObserved(result.phase);
await finalizeProgressMessage();
await deliverFinalText(text);
} else if (raw) {
@@ -1097,18 +1020,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
}
await channel.setTyping?.(chatJid, false);
if (!poisonedSessionDetected) {
resetIdleTimer();
}
if (result.status === 'success' && !poisonedSessionDetected) {
deps.queue.notifyIdle(chatJid, runId);
if (hasVisibleOutput()) {
deps.queue.closeStdin(chatJid, {
runId,
reason: 'output-delivered-close',
});
}
requestAgentClose('output-delivered-close');
}
if (result.status === 'error') {
@@ -1152,9 +1065,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
clearProgressTicker();
if (idleTimer) clearTimeout(idleTimer);
deps.queue.setActivityTouch?.(chatJid, null);
deps.queue.setFollowUpPipeAllowed?.(chatJid, null);
clearPendingFollowUp();
if (!producedDeliverySucceeded) {
logger.warn(
@@ -1277,42 +1187,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
if (!hasTrigger) continue;
}
const allPending = getMessagesSinceSeq(
chatJid,
deps.getLastAgentTimestamps()[chatJid] || '',
deps.assistantName,
);
const processablePending = getProcessableMessages(
chatJid,
allPending,
channel,
);
const messagesToSend =
processablePending.length > 0
? processablePending
: processableGroupMessages;
const formatted = formatMessages(messagesToSend, deps.timezone);
if (deps.queue.sendMessage(chatJid, formatted)) {
logger.debug(
{ chatJid, count: messagesToSend.length },
'Piped messages to active agent',
);
const endSeq = messagesToSend[messagesToSend.length - 1].seq;
if (endSeq != null) {
advanceLastAgentCursor(chatJid, endSeq);
}
channel
.setTyping?.(chatJid, true)
?.catch((err) =>
logger.warn(
{ chatJid, err },
'Failed to set typing indicator',
),
);
} else {
deps.queue.enqueueMessageCheck(chatJid, group.folder);
}
deps.queue.enqueueMessageCheck(chatJid, group.folder);
}
}
} catch (err) {