diff --git a/src/dashboard.ts b/src/dashboard.ts index 7bf756a..4f7134c 100644 --- a/src/dashboard.ts +++ b/src/dashboard.ts @@ -33,7 +33,6 @@ interface CodexRateLimit { const STATUS_ICONS: Record = { 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')}_`; } diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 6fffbd0..7dc4198 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -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((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((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((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((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); - }); }); diff --git a/src/group-queue.ts b/src/group-queue.ts index d82c9e7..ad1052c 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -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 { 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 || diff --git a/src/index.ts b/src/index.ts index 61a1f35..35a2854 100644 --- a/src/index.ts +++ b/src/index.ts @@ -355,14 +355,13 @@ async function refreshChannelMeta(): Promise { } 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 { ( status, ): status is typeof status & { - status: 'processing' | 'idle' | 'waiting'; + status: 'processing' | 'waiting'; } => status.status !== 'inactive', ) .map((status) => ({ diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index cfa7378..09b9651 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -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 = {}; + + 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 = {}; + + 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'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 54d7f72..be705d5 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -126,6 +126,25 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { channel?.isOwnMessage?.bind(channel), ); + const filterLoopingPairedBotMessages = ( + chatJid: string, + messages: Parameters[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) {