adolf/extensions/qa-channel/src/inbound.ts

// Qa Channel plugin module implements inbound behavior.
import { resolveStableChannelMessageIngress } from "openclaw/plugin-sdk/channel-ingress-runtime";
import { resolveNativeCommandSessionTargets } from "openclaw/plugin-sdk/command-auth-native";
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
import { resolveInboundRouteEnvelopeBuilderWithRuntime } from "openclaw/plugin-sdk/inbound-envelope";
import {
  buildAgentMediaPayload,
  saveMediaBuffer,
  saveMediaSource,
} from "openclaw/plugin-sdk/media-runtime";
import {
  sanitizeQaBusToolCallArguments,
  type QaBusToolCall,
} from "openclaw/plugin-sdk/qa-channel-protocol";
import {
  buildQaTarget,
  deleteQaBusMessage,
  editQaBusMessage,
  sendQaBusMessage,
  type QaBusMessage,
} from "./bus-client.js";
import { getQaChannelRuntime } from "./runtime.js";
import type { CoreConfig, ResolvedQaChannelAccount } from "./types.js";

export function isHttpMediaUrl(value: string): boolean {
  try {
    const parsed = new URL(value);
    return parsed.protocol === "http:" || parsed.protocol === "https:";
  } catch {
    return false;
  }
}

function normalizeBase64ForCompare(value: string): string {
  return value.replace(/=+$/u, "").replace(/-/gu, "+").replace(/_/gu, "/");
}

function decodeAttachmentBase64(value: string): Buffer | null {
  const buffer = Buffer.from(value, "base64");
  if (normalizeBase64ForCompare(buffer.toString("base64")) !== normalizeBase64ForCompare(value)) {
    return null;
  }
  return buffer;
}

async function resolveQaInboundMediaPayload(attachments: QaBusMessage["attachments"]) {
  if (!Array.isArray(attachments) || attachments.length === 0) {
    return {};
  }
  const mediaList: Array<{ path: string; contentType?: string | null }> = [];
  for (const attachment of attachments) {
    if (!attachment?.mimeType) {
      continue;
    }
    if (typeof attachment.contentBase64 === "string" && attachment.contentBase64.trim()) {
      const buffer = decodeAttachmentBase64(attachment.contentBase64);
      if (!buffer) {
        console.warn("[qa-channel] inbound attachment contentBase64 rejected (invalid base64)");
        continue;
      }
      const saved = await saveMediaBuffer(
        buffer,
        attachment.mimeType,
        "inbound",
        undefined,
        attachment.fileName,
      );
      mediaList.push({
        path: saved.path,
        contentType: saved.contentType,
      });
      continue;
    }
    if (typeof attachment.url === "string" && attachment.url.trim()) {
      if (!isHttpMediaUrl(attachment.url)) {
        console.warn(
          `[qa-channel] inbound attachment URL rejected (non-http scheme): ${attachment.url}`,
        );
        continue;
      }
      const saved = await saveMediaSource(attachment.url, undefined, "inbound");
      mediaList.push({
        path: saved.path,
        contentType: saved.contentType,
      });
    }
  }
  return mediaList.length > 0 ? buildAgentMediaPayload(mediaList) : {};
}

function resolveQaGroupConfig(params: {
  account: ResolvedQaChannelAccount;
  conversationId: string;
  target: string;
}) {
  const groups = params.account.config.groups;
  return groups?.[params.conversationId] ?? groups?.[params.target] ?? groups?.["*"];
}

function formatQaErrorForLog(error: unknown): string {
  let escaped = "";
  const message = formatErrorMessage(error) || Object.prototype.toString.call(error);
  for (const character of message) {
    const codePoint = character.codePointAt(0) ?? 0;
    const isControl = codePoint <= 0x1f || (codePoint >= 0x7f && codePoint <= 0x9f);
    const isLineSeparator = codePoint === 0x2028 || codePoint === 0x2029;
    escaped +=
      isControl || isLineSeparator ? `\\u${codePoint.toString(16).padStart(4, "0")}` : character;
  }
  return escaped;
}

function createQaReplyPreview(params: {
  account: ResolvedQaChannelAccount;
  inbound: QaBusMessage;
  target: string;
  toolCalls: QaBusToolCall[];
}) {
  let messageId: string | null = null;
  let currentText = "";
  let pending = Promise.resolve();

  const write = (text: string) => {
    if (!text.trim() || text === currentText) {
      return pending;
    }
    pending = pending.then(async () => {
      if (messageId) {
        await editQaBusMessage({
          baseUrl: params.account.baseUrl,
          accountId: params.account.accountId,
          messageId,
          text,
        });
      } else {
        const response = await sendQaBusMessage({
          baseUrl: params.account.baseUrl,
          accountId: params.account.accountId,
          to: params.target,
          text,
          senderId: params.account.botUserId,
          senderName: params.account.botDisplayName,
          threadId: params.inbound.threadId,
          replyToId: params.inbound.id,
          toolCalls: params.toolCalls,
        });
        messageId = response.message.id;
      }
      currentText = text;
    });
    return pending;
  };

  const clear = async () => {
    await pending.catch(() => undefined);
    if (!messageId) {
      return;
    }
    await deleteQaBusMessage({
      baseUrl: params.account.baseUrl,
      accountId: params.account.accountId,
      messageId,
    });
    messageId = null;
    currentText = "";
  };

  const sendDurable = async (text: string) => {
    if (!text.trim()) {
      return;
    }
    await sendQaBusMessage({
      baseUrl: params.account.baseUrl,
      accountId: params.account.accountId,
      to: params.target,
      text,
      senderId: params.account.botUserId,
      senderName: params.account.botDisplayName,
      threadId: params.inbound.threadId,
      replyToId: params.inbound.id,
      toolCalls: params.toolCalls,
    });
  };

  return {
    clear,
    async deliver(text: string, kind: string) {
      await pending;
      if (kind === "final" && messageId && params.toolCalls.length === 0) {
        await write(text);
        return;
      }
      await clear();
      await sendDurable(text);
    },
    update: write,
  };
}

export async function handleQaInbound(params: {
  channelId: string;
  channelLabel: string;
  account: ResolvedQaChannelAccount;
  config: CoreConfig;
  message: QaBusMessage;
}) {
  const runtime = getQaChannelRuntime();
  const inbound = params.message;
  const target = buildQaTarget({
    chatType: inbound.conversation.kind,
    conversationId: inbound.conversation.id,
    threadId: inbound.threadId,
  });
  const toolCalls: QaBusToolCall[] = [];
  const preview = createQaReplyPreview({
    account: params.account,
    inbound,
    target,
    toolCalls,
  });
  const { route, buildEnvelope } = resolveInboundRouteEnvelopeBuilderWithRuntime({
    cfg: params.config as OpenClawConfig,
    channel: params.channelId,
    accountId: params.account.accountId,
    peer: {
      kind:
        inbound.conversation.kind === "direct"
          ? "direct"
          : inbound.conversation.kind === "group"
            ? "group"
            : "channel",
      id: target,
    },
    runtime: runtime.channel,
    sessionStore: params.config.session?.store,
  });
  const isGroup = inbound.conversation.kind !== "direct";
  const wasMentioned = isGroup
    ? runtime.channel.mentions.matchesMentionPatterns(
        inbound.text,
        runtime.channel.mentions.buildMentionRegexes(
          params.config as OpenClawConfig,
          route.agentId,
        ),
      )
    : undefined;
  const groupConfig = isGroup
    ? resolveQaGroupConfig({
        account: params.account,
        conversationId: inbound.conversation.id,
        target,
      })
    : undefined;
  const access = await resolveStableChannelMessageIngress({
    channelId: params.channelId,
    accountId: params.account.accountId,
    identity: { key: "sender", entryIdPrefix: "qa-entry" },
    groupAllowFromFallbackToAllowFrom: true,
    subject: { stableId: inbound.senderId },
    conversation: {
      kind: inbound.conversation.kind,
      id: inbound.conversation.id,
      threadId: inbound.threadId,
      title: inbound.conversation.title,
    },
    mentionFacts: isGroup
      ? {
          canDetectMention: true,
          wasMentioned: wasMentioned ?? false,
        }
      : undefined,
    dmPolicy: "open",
    groupPolicy: params.account.config.groupPolicy ?? "open",
    policy: {
      activation: isGroup
        ? {
            requireMention: groupConfig?.requireMention ?? false,
            allowTextCommands: true,
          }
        : undefined,
    },
    allowFrom: params.account.config.allowFrom,
    groupAllowFrom: params.account.config.groupAllowFrom,
  });
  if (access.ingress.admission !== "dispatch") {
    return;
  }
  const { storePath, body } = buildEnvelope({
    channel: params.channelLabel,
    from: inbound.senderName || inbound.senderId,
    timestamp: inbound.timestamp,
    body: inbound.text,
  });
  const mediaPayload = await resolveQaInboundMediaPayload(inbound.attachments);
  const nativeCommand = inbound.nativeCommand;
  const commandTargets = nativeCommand
    ? resolveNativeCommandSessionTargets({
        agentId: route.agentId,
        sessionPrefix: "qa-channel:slash",
        userId: inbound.senderId,
        targetSessionKey: route.sessionKey,
      })
    : undefined;
  const commandBody = nativeCommand ? `/${nativeCommand.name}` : inbound.text;

  const ctxPayload = runtime.channel.reply.finalizeInboundContext({
    Body: body,
    BodyForAgent: inbound.text,
    RawBody: inbound.text,
    CommandBody: commandBody,
    From: target,
    To: target,
    SessionKey: commandTargets?.sessionKey ?? route.sessionKey,
    CommandTargetSessionKey: commandTargets?.commandTargetSessionKey,
    AccountId: route.accountId ?? params.account.accountId,
    ChatType: inbound.conversation.kind === "direct" ? "direct" : "group",
    WasMentioned: wasMentioned,
    ConversationLabel:
      inbound.threadTitle ||
      inbound.conversation.title ||
      inbound.senderName ||
      inbound.conversation.id,
    GroupSubject: isGroup
      ? inbound.threadTitle || inbound.conversation.title || inbound.conversation.id
      : undefined,
    GroupChannel: inbound.conversation.kind === "channel" ? inbound.conversation.id : undefined,
    NativeChannelId: inbound.conversation.id,
    MessageThreadId: inbound.threadId,
    ThreadLabel: inbound.threadTitle,
    ThreadParentId: inbound.threadId ? inbound.conversation.id : undefined,
    SenderName: inbound.senderName,
    SenderId: inbound.senderId,
    Provider: params.channelId,
    Surface: params.channelId,
    MessageSid: inbound.id,
    MessageSidFull: inbound.id,
    ReplyToId: inbound.replyToId,
    Timestamp: inbound.timestamp,
    OriginatingChannel: params.channelId,
    OriginatingTo: target,
    CommandAuthorized: true,
    CommandSource: nativeCommand ? "native" : undefined,
    CommandTurn: nativeCommand
      ? {
          kind: "native",
          source: "native",
          authorized: true,
          body: commandBody,
        }
      : undefined,
    ...mediaPayload,
  });

  await runtime.channel.inbound.dispatchReply({
    cfg: params.config as OpenClawConfig,
    channel: params.channelId,
    accountId: params.account.accountId,
    agentId: route.agentId,
    routeSessionKey: route.sessionKey,
    storePath,
    ctxPayload,
    recordInboundSession: runtime.channel.session.recordInboundSession,
    dispatchReplyWithBufferedBlockDispatcher:
      runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher,
    delivery: {
      deliver: async (payload, info) => {
        const text =
          payload && typeof payload === "object" && "text" in payload
            ? ((payload as { text?: string }).text ?? "")
            : "";
        if (!text.trim()) {
          return;
        }
        await preview.deliver(text, info?.kind ?? "final");
      },
      onError: (error) => {
        void preview.clear().catch((clearError: unknown) => {
          console.warn(
            `[qa-channel] failed to clear reply preview after dispatch error: ${formatQaErrorForLog(clearError)}`,
          );
        });
        console.warn(`[qa-channel] reply dispatch failed: ${formatQaErrorForLog(error)}`);
      },
    },
    replyOptions: {
      onPartialReply: async (payload) => {
        await preview.update(payload.text ?? "");
      },
      onToolStart: (payload) => {
        if (payload.phase && payload.phase !== "start") {
          return;
        }
        const name = payload.name?.trim();
        if (!name) {
          return;
        }
        const args = sanitizeQaBusToolCallArguments(payload.args);
        toolCalls.push({
          name,
          ...(args && Object.keys(args).length > 0 ? { arguments: args } : {}),
        });
      },
    },
    replyPipeline: {},
    record: {
      onRecordError: (error) => {
        throw error instanceof Error
          ? error
          : new Error(`qa-channel session record failed: ${String(error)}`);
      },
    },
  });
}