From e154c6ec2e77a44b4891623773315e7ca818df6d Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 27 May 2026 08:05:07 +0900 Subject: [PATCH] refactor: split message runtime processing --- quality/code-quality-budgets.json | 4 - src/message-runtime-group-processing.ts | 499 ++++++++++++++++++++++++ src/message-runtime.ts | 456 ++++------------------ 3 files changed, 566 insertions(+), 393 deletions(-) create mode 100644 src/message-runtime-group-processing.ts diff --git a/quality/code-quality-budgets.json b/quality/code-quality-budgets.json index b40aeaf..f324e5b 100644 --- a/quality/code-quality-budgets.json +++ b/quality/code-quality-budgets.json @@ -160,10 +160,6 @@ "maxNesting": 1, "owner": "message runtime test hotspot" }, - "src/message-runtime.ts": { - "maxFunctionLines": 620, - "owner": "message runtime hotspot" - }, "src/message-turn-controller.test.ts": { "maxFunctionLines": 730, "owner": "message turn controller test hotspot" diff --git a/src/message-runtime-group-processing.ts b/src/message-runtime-group-processing.ts new file mode 100644 index 0000000..c8a8294 --- /dev/null +++ b/src/message-runtime-group-processing.ts @@ -0,0 +1,499 @@ +import { + getLatestOpenPairedTaskForChat, + getMessagesSinceSeq, + getOpenWorkItemForChat, + markWorkItemDelivered, + type WorkItem, +} from './db.js'; +import { + isSessionCommandSenderAllowed, + SERVICE_SESSION_SCOPE, +} from './config.js'; +import { resolveRuntimeAttachmentBaseDirs } from './attachment-base-dirs.js'; +import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js'; +import { processOpenWorkItemDelivery } from './message-runtime-delivery.js'; +import { handleQueuedRunGates } from './message-runtime-gating.js'; +import { + runPendingPairedTurnIfNeeded, + runQueuedGroupTurn, +} from './message-runtime-queue.js'; +import { + advanceLastAgentCursor, + filterLoopingPairedBotMessages, + getProcessableMessages, + shouldSkipBotOnlyCollaboration, +} from './message-runtime-rules.js'; +import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js'; +import { transitionPairedTaskStatus } from './paired-task-status.js'; +import { hasReviewerLease } from './service-routing.js'; +import { + getFixedRoleChannelName, + getMissingRoleChannelMessage, +} from './message-runtime-shared.js'; +import { resolvePairedRoleChannels } from './message-runtime-role-channels.js'; +import { + getFreshHumanPreflightMessages, + hasHumanMessageAfterWorkItem, +} from './message-runtime-preflight-messages.js'; +import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js'; +import { findChannel, formatMessages } from './router.js'; +import { createScopedLogger, logger } from './logger.js'; +import type { AgentOutput } from './agent-runner.js'; +import type { GroupQueue, GroupRunContext } from './group-queue.js'; +import type { + ExecuteTurnFn, + RoleToChannelMap, +} from './message-runtime-types.js'; +import type { + Channel, + NewMessage, + PairedTask, + RegisteredGroup, +} from './types.js'; + +type RunAgentFn = ( + group: RegisteredGroup, + prompt: string, + chatJid: string, + runId: string, + onOutput?: (output: AgentOutput) => Promise, +) => Promise<'success' | 'error'>; + +interface ProcessGroupMessagesDeps { + assistantName: string; + failureFinalText: string; + timezone: string; + triggerPattern: RegExp; + channels: Channel[]; + queue: Pick; + getRoomBindings: () => Record; + getLastAgentTimestamps: () => Record; + saveState: () => void; + clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; + runAgent: RunAgentFn; + executeTurn: ExecuteTurnFn; + hasImplicitContinuationWindow: ( + chatJid: string, + messages: NewMessage[], + ) => boolean; + openContinuation: (chatJid: string) => void; + isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; +} + +interface RuntimeContext { + chatJid: string; + runId: string; + group: RegisteredGroup; + channel: Channel; + log: ReturnType; + roleToChannel: RoleToChannelMap; +} + +interface PendingDeliveryState { + pendingTask: PairedTask | null; + openWorkItem: WorkItem | undefined; +} + +export function createProcessGroupMessages( + args: ProcessGroupMessagesDeps, +): (chatJid: string, context: GroupRunContext) => Promise { + return async (chatJid, context): Promise => { + const runtime = resolveRuntimeContext(args, chatJid, context.runId); + if (!runtime) return true; + + const pendingDelivery = resolvePendingDeliveryState(args, runtime); + const deliveryOutcome = await deliverPendingWorkItem( + args, + runtime, + pendingDelivery, + ); + if (deliveryOutcome !== null) { + return deliveryOutcome; + } + + return processMissedMessages(args, runtime); + }; +} + +function resolveRuntimeContext( + args: ProcessGroupMessagesDeps, + chatJid: string, + runId: string, +): RuntimeContext | null { + const group = args.getRoomBindings()[chatJid]; + if (!group) return null; + const log = createScopedLogger({ + chatJid, + groupName: group.name, + groupFolder: group.folder, + runId, + }); + + const channel = findChannel(args.channels, chatJid); + if (!channel) { + log.warn('No channel owns JID, skipping messages'); + return null; + } + + const { + roleToChannel, + reviewerChannelName, + foundReviewerChannel, + arbiterChannelName, + foundArbiterChannel, + } = resolvePairedRoleChannels(args.channels, channel); + if (hasReviewerLease(chatJid)) { + log.info( + { + reviewerChannelName, + foundChannel: foundReviewerChannel?.name ?? null, + arbiterChannelName, + foundArbiterChannel: foundArbiterChannel?.name ?? null, + availableChannels: args.channels.map((c) => c.name), + }, + 'Paired room reviewer/arbiter channel resolution', + ); + } + + return { chatJid, runId, group, channel, log, roleToChannel }; +} + +function resolvePendingDeliveryState( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, +): PendingDeliveryState { + const { chatJid, channel, group, log } = runtime; + let pendingTask: PairedTask | null = hasReviewerLease(chatJid) + ? (getLatestOpenPairedTaskForChat(chatJid) ?? null) + : null; + let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE); + if (openWorkItem?.delivery_role !== 'owner' || !pendingTask) { + return { pendingTask, openWorkItem }; + } + + const freshHumanMessages = getFreshHumanPreflightMessages({ + chatJid, + channel, + lastAgentTimestamps: args.getLastAgentTimestamps(), + assistantName: args.assistantName, + failureFinalText: args.failureFinalText, + }); + if (pendingTask.status === 'merge_ready' && freshHumanMessages.length > 0) { + const resolvedTask = resolveOwnerTaskForHumanMessage({ + group, + chatJid, + existingTask: pendingTask, + }); + pendingTask = resolvedTask.task ?? null; + if (resolvedTask.supersededTask) { + markWorkItemDelivered(openWorkItem.id); + log.info( + { + chatJid, + workItemId: openWorkItem.id, + supersededTaskId: resolvedTask.supersededTask.id, + replacementTaskId: resolvedTask.task?.id ?? null, + }, + 'Suppressed stale owner delivery retry because a new human message superseded the merge_ready task', + ); + openWorkItem = undefined; + } + } else if ( + pendingTask.status === 'active' && + hasHumanMessageAfterWorkItem(openWorkItem, freshHumanMessages) + ) { + markWorkItemDelivered(openWorkItem.id); + log.info( + { + chatJid, + workItemId: openWorkItem.id, + taskId: pendingTask.id, + taskStatus: pendingTask.status, + workItemEndSeq: openWorkItem.end_seq ?? null, + freshHumanMessageCount: freshHumanMessages.length, + }, + 'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active', + ); + openWorkItem = undefined; + } + + return { pendingTask, openWorkItem }; +} + +async function deliverPendingWorkItem( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, + pendingDelivery: PendingDeliveryState, +): Promise { + const { chatJid, runId, group, channel, roleToChannel, log } = runtime; + const openWorkItemOutcome = await processOpenWorkItemDelivery({ + chatJid, + runId, + openWorkItem: pendingDelivery.openWorkItem, + pendingTask: pendingDelivery.pendingTask, + channel, + roleToChannel, + log, + attachmentBaseDirs: resolveRuntimeAttachmentBaseDirs(group), + isPairedRoom: hasReviewerLease(chatJid), + getMissingRoleChannelMessage, + isDuplicateOfLastBotFinal: args.isDuplicateOfLastBotFinal, + openContinuation: args.openContinuation, + enqueueFollowUpAfterDeliveryRetry: ({ + deliveryRole, + pendingTask, + workItemId, + }) => + enqueueDeliveryRetryFollowUp({ + chatJid, + runId, + deliveryRole, + pendingTask, + workItemId, + log, + enqueueMessageCheck: () => args.queue.enqueueMessageCheck(chatJid), + }), + }); + if (openWorkItemOutcome === 'failed') { + return false; + } + if (openWorkItemOutcome === 'delivered') { + return true; + } + return null; +} + +async function processMissedMessages( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, +): Promise { + while (true) { + const rawMissedMessages = getMessagesSinceSeq( + runtime.chatJid, + args.getLastAgentTimestamps()[runtime.chatJid] || '0', + args.assistantName, + ); + const missedMessages = filterLoopingPairedBotMessages( + runtime.chatJid, + getProcessableMessages( + runtime.chatJid, + rawMissedMessages, + runtime.channel, + ), + args.failureFinalText, + ); + + if (missedMessages.length === 0) { + return handleNoMissedMessages(args, runtime, rawMissedMessages); + } + + const pendingTurnOutcome = await runBotOnlyPendingTurn( + args, + runtime, + rawMissedMessages, + missedMessages, + ); + if (pendingTurnOutcome !== null) { + return pendingTurnOutcome; + } + + if (shouldSkipBotOnlyCollaboration(runtime.chatJid, missedMessages)) { + advancePastBotOnlyCollaboration(args, runtime, missedMessages); + return true; + } + + const gateOutcome = await runQueuedRunGates(args, runtime, missedMessages); + if (gateOutcome !== null) { + return gateOutcome; + } + + return runQueuedGroupTurn({ + chatJid: runtime.chatJid, + group: runtime.group, + runId: runtime.runId, + log: runtime.log, + timezone: args.timezone, + missedMessages, + task: hasReviewerLease(runtime.chatJid) + ? (getLatestOpenPairedTaskForChat(runtime.chatJid) ?? null) + : undefined, + roleToChannel: runtime.roleToChannel, + ownerChannel: runtime.channel, + lastAgentTimestamps: args.getLastAgentTimestamps(), + saveState: args.saveState, + executeTurn: args.executeTurn, + getFixedRoleChannelName, + labelPairedSenders: args.labelPairedSenders, + formatMessages, + }); + } +} + +async function handleNoMissedMessages( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, + rawMissedMessages: NewMessage[], +): Promise { + const pendingTurnOutcome = await runPendingPairedTurnIfNeeded({ + chatJid: runtime.chatJid, + group: runtime.group, + runId: runtime.runId, + log: runtime.log, + timezone: args.timezone, + task: hasReviewerLease(runtime.chatJid) + ? getLatestOpenPairedTaskForChat(runtime.chatJid) + : null, + rawMissedMessages, + saveState: args.saveState, + lastAgentTimestamps: args.getLastAgentTimestamps(), + executeTurn: args.executeTurn, + getFixedRoleChannelName, + roleToChannel: runtime.roleToChannel, + labelPairedSenders: args.labelPairedSenders, + mode: 'idle', + }); + if (pendingTurnOutcome !== null) { + return pendingTurnOutcome; + } + + const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; + if (lastIgnored) { + advanceLastAgentCursor( + args.getLastAgentTimestamps(), + args.saveState, + runtime.chatJid, + lastIgnored.timestamp, + ); + } + return true; +} + +function runBotOnlyPendingTurn( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, + rawMissedMessages: NewMessage[], + missedMessages: NewMessage[], +): Promise { + return runPendingPairedTurnIfNeeded({ + chatJid: runtime.chatJid, + group: runtime.group, + runId: runtime.runId, + log: runtime.log, + timezone: args.timezone, + task: hasReviewerLease(runtime.chatJid) + ? getLatestOpenPairedTaskForChat(runtime.chatJid) + : null, + rawMissedMessages, + saveState: args.saveState, + lastAgentTimestamps: args.getLastAgentTimestamps(), + executeTurn: args.executeTurn, + getFixedRoleChannelName, + roleToChannel: runtime.roleToChannel, + labelPairedSenders: args.labelPairedSenders, + mode: 'bot-only', + missedMessages, + }); +} + +function advancePastBotOnlyCollaboration( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, + missedMessages: NewMessage[], +): void { + const lastMessage = missedMessages[missedMessages.length - 1]; + if (lastMessage?.seq != null) { + advanceLastAgentCursor( + args.getLastAgentTimestamps(), + args.saveState, + runtime.chatJid, + lastMessage.seq, + ); + } + runtime.log.info( + 'Skipping bot-only collaboration because no recent human message exists', + ); +} + +async function runQueuedRunGates( + args: ProcessGroupMessagesDeps, + runtime: RuntimeContext, + missedMessages: NewMessage[], +): Promise { + const gateResult = await handleQueuedRunGates({ + chatJid: runtime.chatJid, + group: runtime.group, + runId: runtime.runId, + missedMessages, + triggerPattern: args.triggerPattern, + timezone: args.timezone, + hasImplicitContinuationWindow: args.hasImplicitContinuationWindow, + sessionCommandDeps: { + sendMessage: (text) => + deliverSessionCommandMessage(args, runtime.chatJid, text), + setTyping: (typing) => + runtime.channel.setTyping?.(runtime.chatJid, typing) ?? + Promise.resolve(), + runAgent: (prompt, onOutput) => + args.runAgent( + runtime.group, + prompt, + runtime.chatJid, + runtime.runId, + onOutput, + ), + closeStdin: () => + args.queue.closeStdin(runtime.chatJid, { + reason: 'session-command', + }), + clearSession: (opts) => args.clearSession(runtime.group.folder, opts), + advanceCursor: (cursorOrTimestamp) => { + advanceLastAgentCursor( + args.getLastAgentTimestamps(), + args.saveState, + runtime.chatJid, + cursorOrTimestamp, + ); + }, + formatMessages, + isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender), + canSenderInteract: () => true, + resetPairedTask: () => completePairedTaskAsStopped(runtime.chatJid), + killProcess: () => args.queue.killProcess(runtime.chatJid), + }, + }); + return gateResult.handled ? gateResult.success : null; +} + +function completePairedTaskAsStopped(chatJid: string): void { + if (!hasReviewerLease(chatJid)) return; + const task = getLatestOpenPairedTaskForChat(chatJid); + if (!task) return; + + const now = new Date().toISOString(); + transitionPairedTaskStatus({ + taskId: task.id, + currentStatus: task.status, + nextStatus: 'completed', + expectedUpdatedAt: task.updated_at, + updatedAt: now, + patch: { + completion_reason: 'stopped', + }, + }); +} + +async function deliverSessionCommandMessage( + deps: Pick, + chatJid: string, + text: string, +): Promise { + await deliverCanonicalOutboundMessage( + { jid: chatJid, text }, + { + channels: deps.channels, + roomBindings: deps.getRoomBindings, + log: logger, + }, + ); +} diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 9245310..9a5e868 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,61 +1,24 @@ -import { - getOpenWorkItemForChat, - getMessagesSinceSeq, - getLatestOpenPairedTaskForChat, - markWorkItemDelivered, - updatePairedTurnProgressText, -} from './db.js'; -import { - isSessionCommandSenderAllowed, - SERVICE_SESSION_SCOPE, -} from './config.js'; -import { resolveRuntimeAttachmentBaseDirs } from './attachment-base-dirs.js'; +import { updatePairedTurnProgressText } from './db.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; -import { findChannel, formatMessages } from './router.js'; -import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js'; -import { processOpenWorkItemDelivery } from './message-runtime-delivery.js'; -import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js'; -import { handleQueuedRunGates } from './message-runtime-gating.js'; import { enqueueMessageRuntimePendingHandoffs } from './message-runtime-handoffs.js'; import { buildScopedMessageCheckEnqueuer, processMessageLoopTick, recoverPendingMessages as recoverRuntimePendingMessages, } from './message-runtime-loop.js'; -import { - runPendingPairedTurnIfNeeded, - runQueuedGroupTurn, -} from './message-runtime-queue.js'; -import { - advanceLastAgentCursor, - createImplicitContinuationTracker, - filterLoopingPairedBotMessages, - getProcessableMessages, - shouldSkipBotOnlyCollaboration, -} from './message-runtime-rules.js'; -import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js'; +import { createImplicitContinuationTracker } from './message-runtime-rules.js'; import { schedulePairedFollowUpWithMessageCheck } from './message-runtime-follow-up.js'; -import { transitionPairedTaskStatus } from './paired-task-status.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; -import { createScopedLogger, logger } from './logger.js'; -import { hasReviewerLease } from './service-routing.js'; -import { - getFixedRoleChannelName, - getMissingRoleChannelMessage, -} from './message-runtime-shared.js'; +import { logger } from './logger.js'; import { createRunAgent, createExecuteTurn, isDuplicateOfLastBotFinal, labelPairedSenders, } from './message-runtime-turns.js'; -import { resolvePairedRoleChannels } from './message-runtime-role-channels.js'; -import { - getFreshHumanPreflightMessages, - hasHumanMessageAfterWorkItem, -} from './message-runtime-preflight-messages.js'; import { handleMessageRuntimeAfterDeliverySuccess } from './message-runtime-after-delivery.js'; import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js'; +import { createProcessGroupMessages } from './message-runtime-group-processing.js'; export { isDuplicateOfLastBotFinal }; @@ -77,19 +40,30 @@ export interface MessageRuntimeDeps { clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; } -async function deliverSessionCommandMessage( - deps: Pick, - chatJid: string, - text: string, -): Promise { - await deliverCanonicalOutboundMessage( - { jid: chatJid, text }, - { - channels: deps.channels, - roomBindings: deps.getRoomBindings, - log: logger, - }, - ); +function createStartMessageLoop(args: { + pollInterval: number; + processTick: () => Promise; +}): () => Promise { + let messageLoopRunning = false; + + return async (): Promise => { + if (messageLoopRunning) { + logger.debug('Message loop already running, skipping duplicate start'); + return; + } + messageLoopRunning = true; + + logger.info('EJClaw running'); + + while (true) { + try { + await args.processTick(); + } catch (err) { + logger.error({ err }, 'Error in message loop'); + } + await new Promise((resolve) => setTimeout(resolve, args.pollInterval)); + } + }; } export function createMessageRuntime(deps: MessageRuntimeDeps): { @@ -100,7 +74,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { recoverPendingMessages: () => void; startMessageLoop: () => Promise; } { - let messageLoopRunning = false; const FAILURE_FINAL_TEXT = '요청을 완료하지 못했습니다. 다시 시도해 주세요.'; const continuationTracker = createImplicitContinuationTracker( deps.idleTimeout, @@ -113,10 +86,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.queue, ); - /** - * Check if a message is a duplicate of the last bot final message in a paired room. - * Returns true if duplicate (should be suppressed). - */ const checkDuplicateOfLastBotFinal = ( chatJid: string, text: string, @@ -208,346 +177,55 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }); }; - const processGroupMessages = async ( - chatJid: string, - context: GroupRunContext, - ): Promise => { - const { runId } = context; - const group = deps.getRoomBindings()[chatJid]; - if (!group) return true; - const log = createScopedLogger({ - chatJid, - groupName: group.name, - groupFolder: group.folder, - runId, - }); + const processGroupMessages = createProcessGroupMessages({ + assistantName: deps.assistantName, + failureFinalText: FAILURE_FINAL_TEXT, + timezone: deps.timezone, + triggerPattern: deps.triggerPattern, + channels: deps.channels, + queue: deps.queue, + getRoomBindings: deps.getRoomBindings, + getLastAgentTimestamps: deps.getLastAgentTimestamps, + saveState: deps.saveState, + clearSession: deps.clearSession, + runAgent, + executeTurn, + hasImplicitContinuationWindow: continuationTracker.has, + openContinuation: continuationTracker.open, + isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, + labelPairedSenders: labelPairedRuntimeSenders, + }); - const channel = findChannel(deps.channels, chatJid); - if (!channel) { - log.warn('No channel owns JID, skipping messages'); - return true; - } - - const { - roleToChannel, - reviewerChannelName, - foundReviewerChannel, - arbiterChannelName, - foundArbiterChannel, - } = resolvePairedRoleChannels(deps.channels, channel); - if (hasReviewerLease(chatJid)) { - log.info( - { - reviewerChannelName, - foundChannel: foundReviewerChannel?.name ?? null, - arbiterChannelName, - foundArbiterChannel: foundArbiterChannel?.name ?? null, - availableChannels: deps.channels.map((c) => c.name), - }, - 'Paired room reviewer/arbiter channel resolution', - ); - } - - // Delivery retries can come from forced fallback runs whose agent_type - // differs from the room owner's registered agent type. - let pendingTask = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE); - if (openWorkItem?.delivery_role === 'owner' && pendingTask) { - const freshHumanMessages = getFreshHumanPreflightMessages({ - chatJid, - channel, - lastAgentTimestamps: deps.getLastAgentTimestamps(), + const startMessageLoop = createStartMessageLoop({ + pollInterval: deps.pollInterval, + processTick: async () => { + await processMessageLoopTick({ assistantName: deps.assistantName, failureFinalText: FAILURE_FINAL_TEXT, - }); - if ( - pendingTask.status === 'merge_ready' && - freshHumanMessages.length > 0 - ) { - const resolvedTask = resolveOwnerTaskForHumanMessage({ - group, - chatJid, - existingTask: pendingTask, - }); - pendingTask = resolvedTask.task; - if (resolvedTask.supersededTask) { - markWorkItemDelivered(openWorkItem.id); - log.info( - { - chatJid, - workItemId: openWorkItem.id, - supersededTaskId: resolvedTask.supersededTask.id, - replacementTaskId: resolvedTask.task?.id ?? null, - }, - 'Suppressed stale owner delivery retry because a new human message superseded the merge_ready task', - ); - openWorkItem = undefined; - } - } else if ( - pendingTask.status === 'active' && - hasHumanMessageAfterWorkItem(openWorkItem, freshHumanMessages) - ) { - markWorkItemDelivered(openWorkItem.id); - log.info( - { - chatJid, - workItemId: openWorkItem.id, - taskId: pendingTask.id, - taskStatus: pendingTask.status, - workItemEndSeq: openWorkItem.end_seq ?? null, - freshHumanMessageCount: freshHumanMessages.length, - }, - 'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active', - ); - openWorkItem = undefined; - } - } - const openWorkItemOutcome = await processOpenWorkItemDelivery({ - chatJid, - runId, - openWorkItem, - pendingTask, - channel, - roleToChannel, - log, - attachmentBaseDirs: resolveRuntimeAttachmentBaseDirs(group), - isPairedRoom: hasReviewerLease(chatJid), - getMissingRoleChannelMessage, - isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, - openContinuation: (targetChatJid) => - continuationTracker.open(targetChatJid), - enqueueFollowUpAfterDeliveryRetry: ({ - deliveryRole, - pendingTask, - workItemId, - }) => - enqueueDeliveryRetryFollowUp({ - chatJid, - runId, - deliveryRole, - pendingTask, - workItemId, - log, - enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), - }), - }); - if (openWorkItemOutcome === 'failed') { - return false; - } - if (openWorkItemOutcome === 'delivered') { - return true; - } - - while (true) { - const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0'; - const rawMissedMessages = getMessagesSinceSeq( - chatJid, - sinceSeqCursor, - deps.assistantName, - ); - const missedMessages = filterLoopingPairedBotMessages( - chatJid, - getProcessableMessages(chatJid, rawMissedMessages, channel), - FAILURE_FINAL_TEXT, - ); - - if (missedMessages.length === 0) { - const pendingTurnOutcome = await runPendingPairedTurnIfNeeded({ - chatJid, - group, - runId, - log, - timezone: deps.timezone, - task: hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null, - rawMissedMessages, - saveState: deps.saveState, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - executeTurn, - getFixedRoleChannelName, - roleToChannel, - labelPairedSenders: labelPairedRuntimeSenders, - mode: 'idle', - }); - if (pendingTurnOutcome !== null) { - return pendingTurnOutcome; - } - - const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; - if (lastIgnored) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.timestamp, - ); - } - return true; - } - - const botOnlyPendingTurnOutcome = await runPendingPairedTurnIfNeeded({ - chatJid, - group, - runId, - log, - timezone: deps.timezone, - task: hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null, - rawMissedMessages, - saveState: deps.saveState, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - executeTurn, - getFixedRoleChannelName, - roleToChannel, - labelPairedSenders: labelPairedRuntimeSenders, - mode: 'bot-only', - missedMessages, - }); - if (botOnlyPendingTurnOutcome !== null) { - return botOnlyPendingTurnOutcome; - } - - if (shouldSkipBotOnlyCollaboration(chatJid, missedMessages)) { - const lastMessage = missedMessages[missedMessages.length - 1]; - if (lastMessage?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastMessage.seq, - ); - } - log.info( - 'Skipping bot-only collaboration because no recent human message exists', - ); - return true; - } - - const gateResult = await handleQueuedRunGates({ - chatJid, - group, - runId, - missedMessages, triggerPattern: deps.triggerPattern, timezone: deps.timezone, - hasImplicitContinuationWindow: continuationTracker.has, - sessionCommandDeps: { - sendMessage: (text) => - deliverSessionCommandMessage(deps, chatJid, text), - setTyping: (typing) => - channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), - runAgent: (prompt, onOutput) => - runAgent(group, prompt, chatJid, runId, onOutput), - closeStdin: () => - deps.queue.closeStdin(chatJid, { - reason: 'session-command', - }), - clearSession: (opts) => deps.clearSession(group.folder, opts), - advanceCursor: (cursorOrTimestamp) => { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - cursorOrTimestamp, - ); - }, - formatMessages, - isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender), - canSenderInteract: () => true, - resetPairedTask: () => { - if (hasReviewerLease(chatJid)) { - const task = getLatestOpenPairedTaskForChat(chatJid); - if (task) { - const now = new Date().toISOString(); - transitionPairedTaskStatus({ - taskId: task.id, - currentStatus: task.status, - nextStatus: 'completed', - expectedUpdatedAt: task.updated_at, - updatedAt: now, - patch: { - completion_reason: 'stopped', - }, - }); - } - } - }, - killProcess: () => deps.queue.killProcess(chatJid), - }, - }); - if (gateResult.handled) { - return gateResult.success; - } - - return runQueuedGroupTurn({ - chatJid, - group, - runId, - log, - timezone: deps.timezone, - missedMessages, - task: hasReviewerLease(chatJid) - ? (getLatestOpenPairedTaskForChat(chatJid) ?? null) - : undefined, - roleToChannel, - ownerChannel: channel, + channels: deps.channels, + getRoomBindings: deps.getRoomBindings, + getLastTimestamp: deps.getLastTimestamp, + setLastTimestamp: deps.setLastTimestamp, lastAgentTimestamps: deps.getLastAgentTimestamps(), saveState: deps.saveState, + hasImplicitContinuationWindow: continuationTracker.has, executeTurn, - getFixedRoleChannelName, + enqueuePendingHandoffs, + schedulePairedFollowUpWithMessageCheck, + enqueueScopedGroupMessageCheck, + sendQueuedMessage: deps.queue.sendMessage.bind(deps.queue), + closeStdin: (chatJid, reason) => + deps.queue.closeStdin(chatJid, { + reason, + }), + isRunningMessageTurn: (chatJid) => + deps.queue.getStatuses([chatJid])[0]?.runPhase === 'running_messages', labelPairedSenders: labelPairedRuntimeSenders, - formatMessages, }); - } - }; - - const startMessageLoop = async (): Promise => { - if (messageLoopRunning) { - logger.debug('Message loop already running, skipping duplicate start'); - return; - } - messageLoopRunning = true; - - logger.info('EJClaw running'); - - while (true) { - try { - await processMessageLoopTick({ - assistantName: deps.assistantName, - failureFinalText: FAILURE_FINAL_TEXT, - triggerPattern: deps.triggerPattern, - timezone: deps.timezone, - channels: deps.channels, - getRoomBindings: deps.getRoomBindings, - getLastTimestamp: deps.getLastTimestamp, - setLastTimestamp: deps.setLastTimestamp, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - saveState: deps.saveState, - hasImplicitContinuationWindow: continuationTracker.has, - executeTurn, - enqueuePendingHandoffs, - schedulePairedFollowUpWithMessageCheck, - enqueueScopedGroupMessageCheck, - sendQueuedMessage: deps.queue.sendMessage.bind(deps.queue), - closeStdin: (chatJid, reason) => - deps.queue.closeStdin(chatJid, { - reason, - }), - isRunningMessageTurn: (chatJid) => - deps.queue.getStatuses([chatJid])[0]?.runPhase === - 'running_messages', - labelPairedSenders: labelPairedRuntimeSenders, - }); - } catch (err) { - logger.error({ err }, 'Error in message loop'); - } - await new Promise((resolve) => setTimeout(resolve, deps.pollInterval)); - } - }; + }, + }); const recoverPendingMessages = (): void => { recoverRuntimePendingMessages({