EJClaw/src/message-runtime-loop.ts

import {
  getMessagesSinceSeq,
  getNewMessagesBySeq,
  getOpenWorkItemForChat,
} from './db.js';
import { resolveGroupIpcPath } from './group-folder.js';
import { logger } from './logger.js';
import { processLoopGroupMessages } from './message-runtime-dispatch.js';
import {
  advanceLastAgentCursor,
  getProcessableMessages,
} from './message-runtime-rules.js';
import { SERVICE_SESSION_SCOPE } from './config.js';
import type { ExecuteTurnFn } from './message-runtime-types.js';
import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js';
import { findChannel, formatMessages } from './router.js';
import type {
  Channel,
  NewMessage,
  PairedTask,
  RegisteredGroup,
} from './types.js';

export async function processMessageLoopTick(args: {
  assistantName: string;
  failureFinalText: string;
  triggerPattern: RegExp;
  timezone: string;
  channels: Channel[];
  getRegisteredGroups: () => Record<string, RegisteredGroup>;
  getLastTimestamp: () => string;
  setLastTimestamp: (timestamp: string) => void;
  lastAgentTimestamps: Record<string, string>;
  saveState: () => void;
  hasImplicitContinuationWindow: (
    chatJid: string,
    messages: NewMessage[],
  ) => boolean;
  executeTurn: ExecuteTurnFn;
  enqueuePendingHandoffs: () => void;
  scheduleQueuedPairedFollowUp: (args: {
    chatJid: string;
    runId: string;
    task: PairedTask | null | undefined;
    intentKind: ScheduledPairedFollowUpIntentKind;
    enqueue: () => void;
  }) => boolean;
  enqueueScopedGroupMessageCheck: (
    chatJid: string,
    groupFolder: string,
  ) => void;
  sendQueuedMessage: (chatJid: string, text: string) => boolean;
  closeStdin: (chatJid: string, reason: string) => void;
  labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[];
}): Promise<void> {
  args.enqueuePendingHandoffs();
  const registeredGroups = args.getRegisteredGroups();
  const jids = Object.keys(registeredGroups);
  const { messages, newSeqCursor } = getNewMessagesBySeq(
    jids,
    args.getLastTimestamp(),
    args.assistantName,
  );

  if (messages.length === 0) {
    return;
  }

  logger.info({ count: messages.length }, 'New messages');
  args.setLastTimestamp(newSeqCursor);
  args.saveState();

  const messagesByGroup = new Map<string, NewMessage[]>();
  for (const msg of messages) {
    const existing = messagesByGroup.get(msg.chat_jid);
    if (existing) {
      existing.push(msg);
    } else {
      messagesByGroup.set(msg.chat_jid, [msg]);
    }
  }

  for (const [chatJid, groupMessages] of messagesByGroup) {
    const group = registeredGroups[chatJid];
    if (!group) continue;

    const channel = findChannel(args.channels, chatJid);
    if (!channel) {
      logger.warn({ chatJid }, 'No channel owns JID, skipping messages');
      continue;
    }

    await processLoopGroupMessages({
      chatJid,
      group,
      groupMessages,
      channel,
      assistantName: args.assistantName,
      failureFinalText: args.failureFinalText,
      triggerPattern: args.triggerPattern,
      hasImplicitContinuationWindow: args.hasImplicitContinuationWindow,
      lastAgentTimestamps: args.lastAgentTimestamps,
      saveState: args.saveState,
      timezone: args.timezone,
      executeTurn: args.executeTurn,
      schedulePairedFollowUp: (task, intentKind, followUpRunId) =>
        args.scheduleQueuedPairedFollowUp({
          chatJid,
          runId: followUpRunId,
          task,
          intentKind,
          enqueue: () =>
            args.enqueueScopedGroupMessageCheck(chatJid, group.folder),
        }),
      enqueueMessageCheck: () =>
        args.enqueueScopedGroupMessageCheck(chatJid, group.folder),
      sendQueuedMessage: args.sendQueuedMessage,
      closeStdin: (reason) => args.closeStdin(chatJid, reason),
      labelPairedSenders: args.labelPairedSenders,
      formatMessages,
    });
  }
}

export function recoverPendingMessages(args: {
  assistantName: string;
  channels: Channel[];
  getRegisteredGroups: () => Record<string, RegisteredGroup>;
  lastAgentTimestamps: Record<string, string>;
  saveState: () => void;
  enqueueScopedGroupMessageCheck: (
    chatJid: string,
    groupFolder: string,
  ) => void;
}): void {
  const registeredGroups = args.getRegisteredGroups();
  for (const [chatJid, group] of Object.entries(registeredGroups)) {
    const openWorkItem = getOpenWorkItemForChat(
      chatJid,
      SERVICE_SESSION_SCOPE,
    );
    if (openWorkItem) {
      logger.info(
        { chatJid, group: group.name, workItemId: openWorkItem.id },
        'Recovery: found open work item awaiting delivery',
      );
      args.enqueueScopedGroupMessageCheck(chatJid, group.folder);
      continue;
    }

    const sinceSeqCursor = args.lastAgentTimestamps[chatJid] || '';
    const rawPending = getMessagesSinceSeq(
      chatJid,
      sinceSeqCursor,
      args.assistantName,
    );
    const recoveryChannel = findChannel(args.channels, chatJid);
    const pending = getProcessableMessages(
      chatJid,
      rawPending,
      recoveryChannel ?? undefined,
    );
    if (pending.length > 0) {
      logger.info(
        { group: group.name, pendingCount: pending.length },
        'Recovery: found unprocessed messages',
      );
      args.enqueueScopedGroupMessageCheck(chatJid, group.folder);
      continue;
    }

    if (rawPending.length > 0) {
      const endSeq = rawPending[rawPending.length - 1].seq;
      if (endSeq != null) {
        advanceLastAgentCursor(
          args.lastAgentTimestamps,
          args.saveState,
          chatJid,
          endSeq,
        );
      }
    }
  }
}

export function buildScopedMessageCheckEnqueuer(queue: {
  enqueueMessageCheck: (chatJid: string, ipcDir?: string) => void;
}): (chatJid: string, groupFolder: string) => void {
  return (chatJid: string, groupFolder: string): void => {
    queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(groupFolder));
  };
}