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

refactor: drop idle queue state

Browse Source
This commit is contained in:
Eyejoker
2026-03-23 04:57:13 +09:00
parent 02408aae8e
commit 7c6019dea3
6 changed files with 158 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/dashboard.ts
View File

@@ -33,7 +33,6 @@ interface CodexRateLimit {
const STATUS_ICONS: Record<string, string> = {
processing: '🟡',
idle: '🟢',
waiting: '🔵',
inactive: '⚪',
};
@@ -107,7 +106,6 @@ function getStatusLabel(status: GroupStatus): string {
if (status.status === 'processing') {
return `처리 중 (${formatElapsed(status.elapsedMs || 0)})`;
}
if (status.status === 'idle') return '대기 중';
if (status.status === 'waiting') {
return status.pendingTasks > 0
? `큐 대기 (태스크 ${status.pendingTasks}개)`
@@ -152,7 +150,7 @@ export function buildStatusContent(opts: DashboardOptions): string {
const sections: string[] = [];
let totalActive = 0;
let totalIdle = 0;
let totalWaiting = 0;
let total = 0;
for (const [categoryName, categoryEntries] of sortedCategories) {
@@ -177,13 +175,13 @@ export function buildStatusContent(opts: DashboardOptions): string {
totalActive += categoryEntries.filter(
(entry) => entry.status.status === 'processing',
).length;
totalIdle += categoryEntries.filter(
(entry) => entry.status.status === 'idle',
totalWaiting += categoryEntries.filter(
(entry) => entry.status.status === 'waiting',
).length;
total += categoryEntries.length;
}
const header = `**에이전트 상태** (${opts.assistantName}) — 활성 ${totalActive} | 대기 ${totalIdle} | 전체 ${total}`;
const header = `**에이전트 상태** (${opts.assistantName}) — 활성 ${totalActive} | 큐대기 ${totalWaiting} | 전체 ${total}`;
return `${header}\n\n${sections.join('\n\n')}\n\n_${new Date().toLocaleTimeString('ko-KR')}_`;
}

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

@@ -301,9 +301,9 @@ describe('GroupQueue', () => {
expect(taskCallCount).toBe(1);
});
// --- Idle preemption ---
// --- Active runs queue work without preemption ---
it('does NOT preempt active agent when not idle', async () => {
it('does not preempt an active agent when a task is queued', async () => {
const fs = await import('fs');
let resolveProcess: () => void;
@@ -337,153 +337,4 @@ describe('GroupQueue', () => {
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
it('preempts idle agent when task is enqueued', 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);
// Start processing
queue.enqueueMessageCheck('group1@g.us');
await vi.advanceTimersByTimeAsync(10);
// Register process and mark idle
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
queue.notifyIdle('group1@g.us');
// Clear previous writes, then enqueue a task
const writeFileSync = vi.mocked(fs.default.writeFileSync);
writeFileSync.mockClear();
const taskFn = vi.fn(async () => {});
queue.enqueueTask('group1@g.us', 'task-1', taskFn);
// _close SHOULD have been written (agent is idle)
const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
it('preempts idle agent when a fresh message run is queued', 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');
queue.notifyIdle('group1@g.us');
const writeFileSync = vi.mocked(fs.default.writeFileSync);
writeFileSync.mockClear();
queue.enqueueMessageCheck('group1@g.us');
const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
it('preempts an idle run so the next message is handled in a fresh run', 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');
queue.notifyIdle('group1@g.us');
const writeFileSync = vi.mocked(fs.default.writeFileSync);
writeFileSync.mockClear();
queue.enqueueMessageCheck('group1@g.us', 'test-group');
const closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
expect(processMessages).toHaveBeenCalledTimes(2);
});
it('preempts when idle arrives with pending tasks', 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);
// Start processing
queue.enqueueMessageCheck('group1@g.us');
await vi.advanceTimersByTimeAsync(10);
// Register process and enqueue a task (no idle yet — no preemption)
queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
const writeFileSync = vi.mocked(fs.default.writeFileSync);
writeFileSync.mockClear();
const taskFn = vi.fn(async () => {});
queue.enqueueTask('group1@g.us', 'task-1', taskFn);
let closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(0);
// Now agent becomes idle — should preempt because task is pending
writeFileSync.mockClear();
queue.notifyIdle('group1@g.us');
closeWrites = writeFileSync.mock.calls.filter(
(call) => typeof call[0] === 'string' && call[0].endsWith('_close'),
);
expect(closeWrites).toHaveLength(1);
resolveProcess!();
await vi.advanceTimersByTimeAsync(10);
});
});

49
src/group-queue.ts
View File

@@ -21,7 +21,6 @@ const BASE_RETRY_MS = 5000;
interface GroupState {
active: boolean;
idleWaiting: boolean;
closingStdin: boolean;
isTaskProcess: boolean;
runningTaskId: string | null;
@@ -39,7 +38,7 @@ interface GroupState {
export interface GroupStatus {
jid: string;
status: 'processing' | 'idle' | 'waiting' | 'inactive';
status: 'processing' | 'waiting' | 'inactive';
elapsedMs: number | null;
pendingMessages: boolean;
pendingTasks: number;
@@ -59,7 +58,6 @@ export class GroupQueue {
if (!state) {
state = {
active: false,
idleWaiting: false,
closingStdin: false,
isTaskProcess: false,
runningTaskId: null,
@@ -85,10 +83,6 @@ export class GroupQueue {
this.processMessagesFn = fn;
}
private canPreemptIdleActiveRun(state: GroupState): boolean {
return state.idleWaiting;
}
private createRunId(): string {
return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
}
@@ -105,12 +99,6 @@ export class GroupQueue {
if (state.active) {
state.pendingMessages = true;
if (this.canPreemptIdleActiveRun(state)) {
this.closeStdin(groupJid, {
runId: state.currentRunId ?? undefined,
reason: 'pending-message-preemption',
});
}
logger.debug({ groupJid }, 'Agent active, message queued');
return;
}
@@ -165,9 +153,6 @@ export class GroupQueue {
if (state.active) {
state.pendingTasks.push({ id: taskId, groupJid, fn });
if (this.canPreemptIdleActiveRun(state)) {
this.closeStdin(groupJid);
}
logger.debug({ groupJid, taskId }, 'Agent active, task queued');
return;
}
@@ -212,29 +197,6 @@ export class GroupQueue {
);
}
/**
* Mark the agent process as idle-waiting (finished work, waiting for IPC input).
* If tasks are pending, preempt the idle agent process immediately.
*/
notifyIdle(groupJid: string, runId?: string): void {
const state = this.getGroup(groupJid);
state.idleWaiting = true;
logger.info(
{
groupJid,
runId: runId ?? state.currentRunId,
pendingTasks: state.pendingTasks.length,
},
'Agent entered idle wait state',
);
if (state.pendingTasks.length > 0) {
this.closeStdin(groupJid, {
runId: runId ?? state.currentRunId ?? undefined,
reason: 'pending-task-preemption',
});
}
}
/**
* Signal the active agent process to wind down by writing a close sentinel.
*/
@@ -245,7 +207,6 @@ export class GroupQueue {
const state = this.getGroup(groupJid);
if (!state.active || !state.groupFolder) return;
state.closingStdin = true;
state.idleWaiting = false;
const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input');
try {
@@ -281,7 +242,6 @@ export class GroupQueue {
const state = this.getGroup(groupJid);
const runId = this.createRunId();
state.active = true;
state.idleWaiting = false;
state.closingStdin = false;
state.isTaskProcess = false;
state.currentRunId = runId;
@@ -332,7 +292,6 @@ export class GroupQueue {
);
state.active = false;
state.startedAt = null;
state.idleWaiting = false;
state.closingStdin = false;
state.process = null;
state.processName = null;
@@ -346,7 +305,6 @@ export class GroupQueue {
private async runTask(groupJid: string, task: QueuedTask): Promise<void> {
const state = this.getGroup(groupJid);
state.active = true;
state.idleWaiting = false;
state.closingStdin = false;
state.isTaskProcess = true;
state.runningTaskId = task.id;
@@ -367,7 +325,6 @@ export class GroupQueue {
state.isTaskProcess = false;
state.runningTaskId = null;
state.startedAt = null;
state.idleWaiting = false;
state.closingStdin = false;
state.process = null;
state.processName = null;
@@ -495,10 +452,8 @@ export class GroupQueue {
};
}
let status: GroupStatus['status'];
if (state.active && !state.idleWaiting) {
if (state.active) {
status = 'processing';
} else if (state.active && state.idleWaiting) {
status = 'idle';
} else if (
state.pendingMessages ||
state.pendingTasks.length > 0 ||

9
src/index.ts
View File

@@ -355,14 +355,13 @@ async function refreshChannelMeta(): Promise<void> {
}
function getStatusLabel(s: {
status: 'processing' | 'idle' | 'waiting' | 'inactive';
status: 'processing' | 'waiting' | 'inactive';
elapsedMs: number | null;
pendingMessages: boolean;
pendingTasks: number;
}): string {
if (s.status === 'processing')
return `처리 중 (${formatElapsed(s.elapsedMs || 0)})`;
if (s.status === 'idle') return '대기 중';
if (s.status === 'waiting')
return s.pendingTasks > 0
? `큐 대기 (태스크 ${s.pendingTasks}개)`
@@ -425,7 +424,7 @@ function writeLocalStatusSnapshot(): void {
name: string;
folder: string;
agentType: 'claude-code' | 'codex';
status: 'processing' | 'idle' | 'waiting' | 'inactive';
status: 'processing' | 'waiting' | 'inactive';
elapsedMs: number | null;
pendingMessages: boolean;
pendingTasks: number;
@@ -443,7 +442,7 @@ function buildStatusContent(): string {
// Collect all entries keyed by jid, with agent type info
interface RoomEntry {
agentType: 'claude-code' | 'codex';
status: 'processing' | 'idle' | 'waiting' | 'inactive';
status: 'processing' | 'waiting' | 'inactive';
elapsedMs: number | null;
pendingMessages: boolean;
pendingTasks: number;
@@ -1059,7 +1058,7 @@ async function main(): Promise<void> {
(
status,
): status is typeof status & {
status: 'processing' | 'idle' | 'waiting';
status: 'processing' | 'waiting';
} => status.status !== 'inactive',
)
.map((status) => ({

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

@@ -190,6 +190,130 @@ describe('createMessageRuntime', () => {
});
});
it('ignores generic failure bot messages in paired rooms', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
const saveState = vi.fn();
const lastAgentTimestamps: Record<string, string> = {};
vi.mocked(db.isPairedRoomJid).mockReturnValue(true);
vi.mocked(db.getMessagesSince).mockReturnValue([
{
id: 'msg-1',
chat_jid: chatJid,
sender: 'other-bot@test',
sender_name: 'Other Bot',
content: '요청을 완료하지 못했습니다. 다시 시도해 주세요.',
timestamp: '2026-03-18T09:00:00.000Z',
is_bot_message: true,
},
]);
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: () => lastAgentTimestamps,
saveState,
persistSession: vi.fn(),
clearSession: vi.fn(),
});
const result = await runtime.processGroupMessages(chatJid, {
runId: 'run-ignore-bot-failure-loop',
reason: 'messages',
});
expect(result).toBe(true);
expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
expect(lastAgentTimestamps[chatJid]).toBe('0');
expect(saveState).toHaveBeenCalled();
});
it('keeps mentionless substantive bot messages in paired rooms', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
const saveState = vi.fn();
const lastAgentTimestamps: Record<string, string> = {};
vi.mocked(db.isPairedRoomJid).mockReturnValue(true);
vi.mocked(db.getMessagesSince).mockReturnValue([
{
id: 'msg-1',
chat_jid: chatJid,
sender: 'other-bot@test',
sender_name: 'Other Bot',
content: '정리해보면 Reaction Engine이 1순위 같아.',
timestamp: '2026-03-18T09:00:00.000Z',
is_bot_message: true,
},
]);
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
async (_group, _input, _onProcess, onOutput) => {
await onOutput?.({
status: 'success',
result: '그 방향이 맞습니다.',
newSessionId: 'session-paired-bot',
});
return {
status: 'success',
result: '그 방향이 맞습니다.',
newSessionId: 'session-paired-bot',
};
},
);
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: () => lastAgentTimestamps,
saveState,
persistSession: vi.fn(),
clearSession: vi.fn(),
});
const result = await runtime.processGroupMessages(chatJid, {
runId: 'run-mentionless-paired-bot-message',
reason: 'messages',
});
expect(result).toBe(true);
expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(
chatJid,
'그 방향이 맞습니다.',
);
expect(lastAgentTimestamps[chatJid]).toBe('1');
expect(saveState).toHaveBeenCalled();
});
it('clears Claude sessions and closes stdin immediately on poisoned output', async () => {
const chatJid = 'group@test';
const group = makeGroup('claude-code');

25
src/message-runtime.ts
View File

@@ -126,6 +126,25 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
channel?.isOwnMessage?.bind(channel),
);
const filterLoopingPairedBotMessages = (
chatJid: string,
messages: Parameters<typeof filterProcessableMessages>[0],
failureText: string,
) => {
if (!isPairedRoomJid(chatJid)) return messages;
// Keep mentionless paired-room collaboration intact, but drop the generic
// failure boilerplate from other bots so paired agents do not amplify it
// into a reply loop.
return messages.filter(
(message) =>
!(
message.is_bot_message &&
message.content.trim() === failureText
),
);
};
const deliverOpenWorkItem = async (
channel: Channel,
item: WorkItem,
@@ -575,10 +594,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
sinceSeqCursor,
deps.assistantName,
);
const missedMessages = getProcessableMessages(
const missedMessages = filterLoopingPairedBotMessages(
chatJid,
rawMissedMessages,
channel,
getProcessableMessages(chatJid, rawMissedMessages, channel),
FAILURE_FINAL_TEXT,
);
if (missedMessages.length === 0) {