refactor: unify message hook mapping and async dispatch

src/auto-reply/reply/dispatch-from-config.ts

@@ -2,7 +2,14 @@ import { resolveSessionAgentId } from "../../agents/agent-scope.js";
 import type { OpenClawConfig } from "../../config/config.js";
 import { loadSessionStore, resolveStorePath, type SessionEntry } from "../../config/sessions.js";
 import { logVerbose } from "../../globals.js";
+import { fireAndForgetHook } from "../../hooks/fire-and-forget.js";
 import { createInternalHookEvent, triggerInternalHook } from "../../hooks/internal-hooks.js";
+import {
+  deriveInboundMessageHookContext,
+  toInternalMessageReceivedContext,
+  toPluginMessageContext,
+  toPluginMessageReceivedEvent,
+} from "../../hooks/message-hook-mappers.js";
 import { isDiagnosticsEnabled } from "../../infra/diagnostic-events.js";
 import {
   logMessageProcessed,
@@ -167,81 +174,31 @@ export async function dispatchReplyFromConfig(params: {
     typeof ctx.Timestamp === "number" && Number.isFinite(ctx.Timestamp) ? ctx.Timestamp : undefined;
   const messageIdForHook =
     ctx.MessageSidFull ?? ctx.MessageSid ?? ctx.MessageSidFirst ?? ctx.MessageSidLast;
-  const content =
-    typeof ctx.BodyForCommands === "string"
-      ? ctx.BodyForCommands
-      : typeof ctx.RawBody === "string"
-        ? ctx.RawBody
-        : typeof ctx.Body === "string"
-          ? ctx.Body
-          : "";
-  const channelId = (ctx.OriginatingChannel ?? ctx.Surface ?? ctx.Provider ?? "").toLowerCase();
-  const conversationId = ctx.OriginatingTo ?? ctx.To ?? ctx.From ?? undefined;
-  const isGroup = Boolean(ctx.GroupSubject || ctx.GroupChannel);
-  const groupId = isGroup ? conversationId : undefined;
+  const hookContext = deriveInboundMessageHookContext(ctx, { messageId: messageIdForHook });
+  const { isGroup, groupId } = hookContext;
 
   // Trigger plugin hooks (fire-and-forget)
   if (hookRunner?.hasHooks("message_received")) {
-    void hookRunner
-      .runMessageReceived(
-        {
-          from: ctx.From ?? "",
-          content,
-          timestamp,
-          metadata: {
-            to: ctx.To,
-            provider: ctx.Provider,
-            surface: ctx.Surface,
-            threadId: ctx.MessageThreadId,
-            originatingChannel: ctx.OriginatingChannel,
-            originatingTo: ctx.OriginatingTo,
-            messageId: messageIdForHook,
-            senderId: ctx.SenderId,
-            senderName: ctx.SenderName,
-            senderUsername: ctx.SenderUsername,
-            senderE164: ctx.SenderE164,
-            guildId: ctx.GroupSpace,
-            channelName: ctx.GroupChannel,
-          },
-        },
-        {
-          channelId,
-          accountId: ctx.AccountId,
-          conversationId,
-        },
-      )
-      .catch((err) => {
-        logVerbose(`dispatch-from-config: message_received plugin hook failed: ${String(err)}`);
-      });
+    fireAndForgetHook(
+      hookRunner.runMessageReceived(
+        toPluginMessageReceivedEvent(hookContext),
+        toPluginMessageContext(hookContext),
+      ),
+      "dispatch-from-config: message_received plugin hook failed",
+    );
   }
 
   // Bridge to internal hooks (HOOK.md discovery system) - refs #8807
   if (sessionKey) {
-    void triggerInternalHook(
-      createInternalHookEvent("message", "received", sessionKey, {
-        from: ctx.From ?? "",
-        content,
-        timestamp,
-        channelId,
-        accountId: ctx.AccountId,
-        conversationId,
-        messageId: messageIdForHook,
-        metadata: {
-          to: ctx.To,
-          provider: ctx.Provider,
-          surface: ctx.Surface,
-          threadId: ctx.MessageThreadId,
-          senderId: ctx.SenderId,
-          senderName: ctx.SenderName,
-          senderUsername: ctx.SenderUsername,
-          senderE164: ctx.SenderE164,
-          guildId: ctx.GroupSpace,
-          channelName: ctx.GroupChannel,
-        },
-      }),
-    ).catch((err) => {
-      logVerbose(`dispatch-from-config: message_received internal hook failed: ${String(err)}`);
-    });
+    fireAndForgetHook(
+      triggerInternalHook(
+        createInternalHookEvent("message", "received", sessionKey, {
+          ...toInternalMessageReceivedContext(hookContext),
+          timestamp,
+        }),
+      ),
+      "dispatch-from-config: message_received internal hook failed",
+    );
   }
 
   // Check if we should route replies to originating channel instead of dispatcher.

src/auto-reply/reply/get-reply.ts

@@ -9,8 +9,6 @@ import { resolveAgentTimeoutMs } from "../../agents/timeout.js";
 import { DEFAULT_AGENT_WORKSPACE_DIR, ensureAgentWorkspace } from "../../agents/workspace.js";
 import { resolveChannelModelOverride } from "../../channels/model-overrides.js";
 import { type OpenClawConfig, loadConfig } from "../../config/config.js";
-import { logVerbose } from "../../globals.js";
-import { createInternalHookEvent, triggerInternalHook } from "../../hooks/internal-hooks.js";
 import { applyLinkUnderstanding } from "../../link-understanding/apply.js";
 import { applyMediaUnderstanding } from "../../media-understanding/apply.js";
 import { defaultRuntime } from "../../runtime.js";
@@ -24,6 +22,7 @@ import { resolveReplyDirectives } from "./get-reply-directives.js";
 import { handleInlineActions } from "./get-reply-inline-actions.js";
 import { runPreparedReply } from "./get-reply-run.js";
 import { finalizeInboundContext } from "./inbound-context.js";
+import { emitPreAgentMessageHooks } from "./message-preprocess-hooks.js";
 import { applyResetModelOverride } from "./session-reset-model.js";
 import { initSessionState } from "./session.js";
 import { stageSandboxMedia } from "./stage-sandbox-media.js";
@@ -137,76 +136,11 @@ export async function getReplyFromConfig(
       cfg,
     });
   }
-
-  const channelId = (
-    finalized.OriginatingChannel ??
-    finalized.Surface ??
-    finalized.Provider ??
-    ""
-  ).toLowerCase();
-  const hookSessionKey = finalized.SessionKey?.trim();
-  const conversationId = finalized.OriginatingTo ?? finalized.To ?? finalized.From ?? undefined;
-  const isGroupConversation = Boolean(finalized.GroupSubject || finalized.GroupChannel);
-  const groupId = isGroupConversation ? conversationId : undefined;
-
-  // Trigger message:transcribed hook after media understanding completes
-  // Only fire if transcription actually occurred (skip in fast test mode or non-audio)
-  if (!isFastTestEnv && hookSessionKey && finalized.Transcript) {
-    void triggerInternalHook(
-      createInternalHookEvent("message", "transcribed", hookSessionKey, {
-        from: finalized.From,
-        to: finalized.To,
-        body: finalized.Body,
-        bodyForAgent: finalized.BodyForAgent,
-        transcript: finalized.Transcript,
-        timestamp: finalized.Timestamp,
-        channelId,
-        conversationId,
-        messageId: finalized.MessageSid,
-        senderId: finalized.SenderId,
-        senderName: finalized.SenderName,
-        senderUsername: finalized.SenderUsername,
-        provider: finalized.Provider,
-        surface: finalized.Surface,
-        mediaPath: finalized.MediaPath,
-        mediaType: finalized.MediaType,
-        cfg,
-      }),
-    ).catch((err) => {
-      logVerbose(`get-reply: message:transcribed internal hook failed: ${String(err)}`);
-    });
-  }
-
-  // Trigger message:preprocessed hook after all media + link understanding.
-  // Fires for every message, giving hooks access to the fully enriched body
-  // (transcripts, image descriptions, link summaries) before the agent sees it.
-  if (!isFastTestEnv && hookSessionKey) {
-    void triggerInternalHook(
-      createInternalHookEvent("message", "preprocessed", hookSessionKey, {
-        from: finalized.From,
-        to: finalized.To,
-        body: finalized.Body,
-        bodyForAgent: finalized.BodyForAgent,
-        transcript: finalized.Transcript,
-        timestamp: finalized.Timestamp,
-        channelId,
-        conversationId,
-        messageId: finalized.MessageSid,
-        senderId: finalized.SenderId,
-        senderName: finalized.SenderName,
-        senderUsername: finalized.SenderUsername,
-        provider: finalized.Provider,
-        surface: finalized.Surface,
-        mediaPath: finalized.MediaPath,
-        mediaType: finalized.MediaType,
-        isGroup: isGroupConversation,
-        groupId,
-        cfg,
-      }),
-    ).catch((err) => {
-      logVerbose(`get-reply: message:preprocessed internal hook failed: ${String(err)}`);
-    });
-  }
+  emitPreAgentMessageHooks({
+    ctx: finalized,
+    cfg,
+    isFastTestEnv,
+  });
 
   const commandAuthorized = finalized.CommandAuthorized;
   resolveCommandAuthorization({

src/auto-reply/reply/message-preprocess-hooks.test.ts

@@ -0,0 +1,93 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { OpenClawConfig } from "../../config/config.js";
+import { clearInternalHooks, registerInternalHook } from "../../hooks/internal-hooks.js";
+import type { FinalizedMsgContext } from "../templating.js";
+import { emitPreAgentMessageHooks } from "./message-preprocess-hooks.js";
+
+function makeCtx(overrides: Partial<FinalizedMsgContext> = {}): FinalizedMsgContext {
+  return {
+    SessionKey: "agent:main:telegram:chat-1",
+    From: "telegram:user:1",
+    To: "telegram:chat-1",
+    Body: "<media:audio>",
+    BodyForAgent: "[Audio] Transcript: hello",
+    BodyForCommands: "<media:audio>",
+    Transcript: "hello",
+    Provider: "telegram",
+    Surface: "telegram",
+    OriginatingChannel: "telegram",
+    OriginatingTo: "telegram:chat-1",
+    Timestamp: 1710000000,
+    MessageSid: "msg-1",
+    GroupChannel: "ops",
+    ...overrides,
+  } as FinalizedMsgContext;
+}
+
+describe("emitPreAgentMessageHooks", () => {
+  beforeEach(() => {
+    clearInternalHooks();
+  });
+
+  it("emits transcribed and preprocessed events when transcript exists", async () => {
+    const actions: string[] = [];
+    registerInternalHook("message", (event) => {
+      actions.push(event.action);
+    });
+
+    emitPreAgentMessageHooks({
+      ctx: makeCtx(),
+      cfg: {} as OpenClawConfig,
+      isFastTestEnv: false,
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(actions).toEqual(["transcribed", "preprocessed"]);
+  });
+
+  it("emits only preprocessed when transcript is missing", async () => {
+    const actions: string[] = [];
+    registerInternalHook("message", (event) => {
+      actions.push(event.action);
+    });
+
+    emitPreAgentMessageHooks({
+      ctx: makeCtx({ Transcript: undefined }),
+      cfg: {} as OpenClawConfig,
+      isFastTestEnv: false,
+    });
+    await Promise.resolve();
+    await Promise.resolve();
+
+    expect(actions).toEqual(["preprocessed"]);
+  });
+
+  it("skips hook emission in fast-test mode", async () => {
+    const handler = vi.fn();
+    registerInternalHook("message", handler);
+
+    emitPreAgentMessageHooks({
+      ctx: makeCtx(),
+      cfg: {} as OpenClawConfig,
+      isFastTestEnv: true,
+    });
+    await Promise.resolve();
+
+    expect(handler).not.toHaveBeenCalled();
+  });
+
+  it("skips hook emission without session key", async () => {
+    const handler = vi.fn();
+    registerInternalHook("message", handler);
+
+    emitPreAgentMessageHooks({
+      ctx: makeCtx({ SessionKey: " " }),
+      cfg: {} as OpenClawConfig,
+      isFastTestEnv: false,
+    });
+    await Promise.resolve();
+
+    expect(handler).not.toHaveBeenCalled();
+  });
+});

src/auto-reply/reply/message-preprocess-hooks.ts

@@ -0,0 +1,50 @@
+import type { OpenClawConfig } from "../../config/config.js";
+import { fireAndForgetHook } from "../../hooks/fire-and-forget.js";
+import { createInternalHookEvent, triggerInternalHook } from "../../hooks/internal-hooks.js";
+import {
+  deriveInboundMessageHookContext,
+  toInternalMessagePreprocessedContext,
+  toInternalMessageTranscribedContext,
+} from "../../hooks/message-hook-mappers.js";
+import type { FinalizedMsgContext } from "../templating.js";
+
+export function emitPreAgentMessageHooks(params: {
+  ctx: FinalizedMsgContext;
+  cfg: OpenClawConfig;
+  isFastTestEnv: boolean;
+}): void {
+  if (params.isFastTestEnv) {
+    return;
+  }
+  const sessionKey = params.ctx.SessionKey?.trim();
+  if (!sessionKey) {
+    return;
+  }
+
+  const canonical = deriveInboundMessageHookContext(params.ctx);
+  if (canonical.transcript) {
+    fireAndForgetHook(
+      triggerInternalHook(
+        createInternalHookEvent(
+          "message",
+          "transcribed",
+          sessionKey,
+          toInternalMessageTranscribedContext(canonical, params.cfg),
+        ),
+      ),
+      "get-reply: message:transcribed internal hook failed",
+    );
+  }
+
+  fireAndForgetHook(
+    triggerInternalHook(
+      createInternalHookEvent(
+        "message",
+        "preprocessed",
+        sessionKey,
+        toInternalMessagePreprocessedContext(canonical, params.cfg),
+      ),
+    ),
+    "get-reply: message:preprocessed internal hook failed",
+  );
+}