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

refactor: restore pre-v1 runtime follow-up flow

Browse Source
This commit is contained in:
Eyejoker
2026-03-23 05:54:48 +09:00
parent 77a7918063
commit 2c37103243
3 changed files with 178 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<boolean>((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 () => {

49
src/group-queue.ts
View File

@@ -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.
*/

137
src/message-runtime.ts
View File

@@ -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<string, number>();
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<typeof filterProcessableMessages>[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);