From 2c37103243639b198ce95c7da2ca0cc307f1661a Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Mon, 23 Mar 2026 05:54:48 +0900 Subject: [PATCH] refactor: restore pre-v1 runtime follow-up flow --- src/group-queue.test.ts | 27 ++++++++ src/group-queue.ts | 49 ++++++++++++++ src/message-runtime.ts | 137 ++++++++++++++++++++++++++++++---------- 3 files changed, 178 insertions(+), 35 deletions(-) diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 7dc4198..374fdfe 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -1,4 +1,5 @@ import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; +import fs from 'fs'; import { GroupQueue } from './group-queue.js'; @@ -62,6 +63,32 @@ describe('GroupQueue', () => { expect(maxConcurrent).toBe(1); }); + it('pipes follow-up messages to an active non-task process', async () => { + let releaseRun!: (value: boolean) => void; + const blocker = new Promise((resolve) => { + releaseRun = resolve; + }); + + const processMessages = vi.fn( + async () => await blocker, + ); + + queue.setProcessMessagesFn(processMessages); + queue.enqueueMessageCheck('group1@g.us', 'group-folder'); + await vi.advanceTimersByTimeAsync(10); + + expect(queue.sendMessage('group1@g.us', '후속 메시지')).toBe(true); + expect(fs.mkdirSync).toHaveBeenCalledWith( + '/tmp/ejclaw-test-data/ipc/group-folder/input', + { recursive: true }, + ); + expect(fs.writeFileSync).toHaveBeenCalled(); + expect(fs.renameSync).toHaveBeenCalled(); + + releaseRun(true); + await vi.advanceTimersByTimeAsync(10); + }); + // --- Global concurrency limit --- it('respects global concurrency limit', async () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index ad1052c..a10c732 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -197,6 +197,55 @@ export class GroupQueue { ); } + 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, + groupFolder: state.groupFolder, + isTaskProcess: state.isTaskProcess, + }, + 'Cannot pipe follow-up message to active agent', + ); + return false; + } + + 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); + 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. */ diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 10d4aa5..3165fef 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -8,6 +8,7 @@ import { import { getAllChats, getAllTasks, + getLastHumanMessageTimestamp, getLatestMessageSeqAtOrBefore, getMessagesSinceSeq, getNewMessagesBySeq, @@ -87,6 +88,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } { let messageLoopRunning = false; const implicitContinuationUntil = new Map(); + const BOT_COLLABORATION_WINDOW_MS = 12 * 60 * 60 * 1000; + const FAILURE_FINAL_TEXT = '요청을 완료하지 못했습니다. 다시 시도해 주세요.'; const getCurrentAvailableGroups = (): AvailableGroup[] => getAvailableGroups(deps.getRegisteredGroups()); @@ -134,6 +137,40 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return messages.some((message) => message.is_from_me !== true); }; + const shouldSkipBotOnlyCollaboration = ( + chatJid: string, + messages: NewMessage[], + ): boolean => { + if (isPairedRoomJid(chatJid)) return false; + const allFromBots = messages.every( + (message) => message.is_from_me || !!message.is_bot_message, + ); + if (!allFromBots) return false; + const lastHuman = getLastHumanMessageTimestamp(chatJid); + if (!lastHuman) return true; + return ( + Date.now() - new Date(lastHuman).getTime() > BOT_COLLABORATION_WINDOW_MS + ); + }; + + const hasAllowedTrigger = ( + chatJid: string, + messages: NewMessage[], + group: RegisteredGroup, + ): boolean => { + if (group.isMain === true || group.requiresTrigger === false) { + return true; + } + const allowlistCfg = loadSenderAllowlist(); + const hasTrigger = messages.some( + (message) => + deps.triggerPattern.test(message.content.trim()) && + (message.is_from_me || + isTriggerAllowed(chatJid, message.sender, allowlistCfg)), + ); + return hasTrigger || hasImplicitContinuationWindow(chatJid, messages); + }; + const getProcessableMessages = ( chatJid: string, messages: Parameters[0], @@ -602,9 +639,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const isMainGroup = group.isMain === true; const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; - const FAILURE_FINAL_TEXT = - '요청을 완료하지 못했습니다. 다시 시도해 주세요.'; - while (true) { const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; const rawMissedMessages = getMessagesSinceSeq( @@ -626,6 +660,18 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return true; } + if (shouldSkipBotOnlyCollaboration(chatJid, missedMessages)) { + const lastMessage = missedMessages[missedMessages.length - 1]; + if (lastMessage?.seq != null) { + advanceLastAgentCursor(chatJid, lastMessage.seq); + } + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Skipping bot-only collaboration because no recent human message exists', + ); + return true; + } + const cmdResult = await handleSessionCommand({ missedMessages, isMainGroup, @@ -665,24 +711,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }); if (cmdResult.handled) return cmdResult.success; - if (!isMainGroup && group.requiresTrigger !== false) { - const allowlistCfg = loadSenderAllowlist(); - const hasTrigger = missedMessages.some( - (msg) => - deps.triggerPattern.test(msg.content.trim()) && - (msg.is_from_me || - isTriggerAllowed(chatJid, msg.sender, allowlistCfg)), + if (!hasAllowedTrigger(chatJid, missedMessages, group)) { + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Skipping queued run because no allowed trigger was found', ); - if ( - !hasTrigger && - !hasImplicitContinuationWindow(chatJid, missedMessages) - ) { - logger.info( - { chatJid, group: group.name, groupFolder: group.folder, runId }, - 'Skipping queued run because no allowed trigger was found', - ); - return true; - } + return true; } const prompt = formatMessages(missedMessages, deps.timezone); @@ -1192,6 +1226,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { continue; } + if ( + shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages) + ) { + const lastIgnored = + processableGroupMessages[processableGroupMessages.length - 1]; + if (lastIgnored?.seq != null) { + advanceLastAgentCursor(chatJid, lastIgnored.seq); + } + logger.info( + { chatJid, group: group.name, groupFolder: group.folder }, + 'Bot-collaboration timeout: no recent human message, skipping', + ); + continue; + } + const loopCmdMsg = groupMessages.find( (msg) => extractSessionCommand(msg.content, deps.triggerPattern) !== @@ -1214,22 +1263,40 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { continue; } - const needsTrigger = - !isMainGroup && group.requiresTrigger !== false; - if (needsTrigger) { - const allowlistCfg = loadSenderAllowlist(); - const hasTrigger = processableGroupMessages.some( - (msg) => - deps.triggerPattern.test(msg.content.trim()) && - (msg.is_from_me || - isTriggerAllowed(chatJid, msg.sender, allowlistCfg)), - ); - if ( - !hasTrigger && - !hasImplicitContinuationWindow(chatJid, processableGroupMessages) - ) { - continue; + if (!hasAllowedTrigger(chatJid, processableGroupMessages, group)) { + continue; + } + + const rawPendingMessages = getMessagesSinceSeq( + chatJid, + deps.getLastAgentTimestamps()[chatJid] || '0', + deps.assistantName, + ); + const pendingMessages = filterLoopingPairedBotMessages( + chatJid, + getProcessableMessages(chatJid, rawPendingMessages, channel), + FAILURE_FINAL_TEXT, + ); + const messagesToSend = + pendingMessages.length > 0 + ? pendingMessages + : processableGroupMessages; + const formatted = formatMessages(messagesToSend, deps.timezone); + + if (deps.queue.sendMessage(chatJid, formatted)) { + 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', + ), + ); + continue; } deps.queue.enqueueMessageCheck(chatJid, group.folder);