Dirigent/plugin/hooks/message-received.ts

import type { OpenClawPluginApi } from "openclaw/plugin-sdk/core";
import type { ChannelStore } from "../core/channel-store.js";
import type { IdentityRegistry } from "../core/identity-registry.js";
import { parseDiscordChannelId } from "./before-model-resolve.js";
import { isDormant, wakeFromDormant, isCurrentSpeaker, hasSpeakers, setSpeakerList, getInitializingChannels, type SpeakerEntry } from "../turn-manager.js";
import { sendScheduleTrigger, userIdFromBotToken } from "../core/moderator-discord.js";
import { fetchVisibleChannelBotAccountIds } from "../core/channel-members.js";
import type { InterruptFn } from "./agent-end.js";

type Deps = {
  api: OpenClawPluginApi;
  channelStore: ChannelStore;
  identityRegistry: IdentityRegistry;
  moderatorBotToken: string | undefined;
  scheduleIdentifier: string;
  interruptTailMatch: InterruptFn;
  debugMode: boolean;
  /**
   * When true, the moderator service handles wake-from-dormant and
   * interrupt-tail-match via HTTP callback. This hook only runs speaker-list
   * initialization in that case.
   */
  moderatorHandlesMessages?: boolean;
};

export function registerMessageReceivedHook(deps: Deps): void {
  const { api, channelStore, identityRegistry, moderatorBotToken, scheduleIdentifier, interruptTailMatch, debugMode, moderatorHandlesMessages } = deps;
  const moderatorBotUserId = moderatorBotToken ? userIdFromBotToken(moderatorBotToken) : undefined;

  api.on("message_received", async (event, ctx) => {
    try {
      const e = event as Record<string, unknown>;
      const c = ctx as Record<string, unknown>;

      // Extract Discord channel ID from ctx or event metadata
      let channelId: string | undefined;

      if (typeof c.channelId === "string") {
        const bare = c.channelId.match(/^(\d+)$/)?.[1] ?? c.channelId.match(/:(\d+)$/)?.[1];
        if (bare) channelId = bare;
      }
      if (!channelId && typeof c.sessionKey === "string") {
        channelId = parseDiscordChannelId(c.sessionKey);
      }
      if (!channelId) {
        const metadata = e.metadata as Record<string, unknown> | undefined;
        const to = String(metadata?.to ?? metadata?.originatingTo ?? "");
        const toMatch = to.match(/:(\d+)$/);
        if (toMatch) channelId = toMatch[1];
      }
      if (!channelId) {
        const metadata = e.metadata as Record<string, unknown> | undefined;
        const convInfo = metadata?.conversation_info as Record<string, unknown> | undefined;
        const raw = String(convInfo?.channel_id ?? metadata?.channelId ?? "");
        if (/^\d+$/.test(raw)) channelId = raw;
      }
      if (!channelId) return;

      const mode = channelStore.getMode(channelId);

      if (mode === "report") return;
      if (mode === "none" || mode === "work") return;

      // ── Speaker-list initialization (always runs, even with moderator service) ──
      const initializingChannels = getInitializingChannels();
      if (!hasSpeakers(channelId) && moderatorBotToken) {
        if (initializingChannels.has(channelId)) {
          api.logger.info(`dirigent: message_received init in progress, skipping channel=${channelId}`);
          return;
        }
        initializingChannels.add(channelId);
        try {
          const agentIds = await fetchVisibleChannelBotAccountIds(api, channelId, identityRegistry);
          const speakers: SpeakerEntry[] = agentIds
            .map((aid) => {
              const entry = identityRegistry.findByAgentId(aid);
              return entry ? { agentId: aid, discordUserId: entry.discordUserId } : null;
            })
            .filter((s): s is SpeakerEntry => s !== null);

          if (speakers.length > 0) {
            setSpeakerList(channelId, speakers);
            const first = speakers[0];
            api.logger.info(`dirigent: initialized speaker list channel=${channelId} speakers=${speakers.map(s => s.agentId).join(",")}`);
            await sendScheduleTrigger(moderatorBotToken, channelId, `<@${first.discordUserId}>${scheduleIdentifier}`, api.logger, debugMode);
            return;
          }
        } finally {
          initializingChannels.delete(channelId);
        }
      }

      // ── Wake / interrupt (skipped when moderator service handles it via HTTP callback) ──
      if (moderatorHandlesMessages) return;

      const senderId = String(
        (e.metadata as Record<string, unknown>)?.senderId ??
        (e.metadata as Record<string, unknown>)?.sender_id ??
        e.from ?? "",
      );

      const currentSpeakerIsThisSender = (() => {
        if (!senderId) return false;
        const entry = identityRegistry.findByDiscordUserId(senderId);
        if (!entry) return false;
        return isCurrentSpeaker(channelId!, entry.agentId);
      })();

      if (!currentSpeakerIsThisSender) {
        if (senderId !== moderatorBotUserId) {
          interruptTailMatch(channelId);
          api.logger.info(`dirigent: message_received interrupt tail-match channel=${channelId} senderId=${senderId}`);
        }

        if (isDormant(channelId) && moderatorBotToken && senderId !== moderatorBotUserId) {
          const first = wakeFromDormant(channelId);
          if (first) {
            const msg = `<@${first.discordUserId}>${scheduleIdentifier}`;
            await sendScheduleTrigger(moderatorBotToken, channelId, msg, api.logger, debugMode);
            api.logger.info(`dirigent: woke dormant channel=${channelId} first speaker=${first.agentId}`);
          }
        }
      }
    } catch (err) {
      api.logger.warn(`dirigent: message_received hook error: ${String(err)}`);
    }
  });
}