openclaw/extensions/qa-lab/src/bus-queries.ts

// Qa Lab plugin module implements bus queries behavior.
import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime";
import { parseQaTarget } from "./qa-bus-protocol.js";
import type {
  QaBusAttachment,
  QaBusConversation,
  QaBusEvent,
  QaBusMessage,
  QaBusPollInput,
  QaBusPollResult,
  QaBusReadMessageInput,
  QaBusSearchMessagesInput,
  QaBusSnapshotConversation,
  QaBusStateSnapshot,
  QaBusThread,
  QaBusToolCall,
} from "./runtime-api.js";

export const DEFAULT_ACCOUNT_ID = "default";

export function normalizeAccountId(raw?: string): string {
  const trimmed = raw?.trim();
  return trimmed || DEFAULT_ACCOUNT_ID;
}

export function normalizeConversationFromTarget(target: string): {
  conversation: QaBusConversation;
  threadId?: string;
} {
  const parsed = parseQaTarget(target);
  return {
    conversation: { id: parsed.conversationId, kind: parsed.chatType },
    ...(parsed.threadId !== undefined ? { threadId: parsed.threadId } : {}),
  };
}

export function cloneMessage(message: QaBusMessage): QaBusMessage {
  return {
    ...message,
    conversation: { ...message.conversation },
    attachments: (message.attachments ?? []).map((attachment) => cloneAttachment(attachment)),
    ...(message.nativeCommand ? { nativeCommand: { ...message.nativeCommand } } : {}),
    toolCalls: message.toolCalls?.map((toolCall) => cloneToolCall(toolCall)),
    reactions: message.reactions.map((reaction) => ({ ...reaction })),
  };
}

function cloneAttachment(attachment: QaBusAttachment): QaBusAttachment {
  return { ...attachment };
}

function cloneToolCall(toolCall: QaBusToolCall): QaBusToolCall {
  return {
    name: toolCall.name,
    ...(toolCall.arguments ? { arguments: structuredClone(toolCall.arguments) } : {}),
  };
}

export function cloneEvent(event: QaBusEvent): QaBusEvent {
  switch (event.kind) {
    case "inbound-message":
    case "outbound-message":
    case "message-edited":
    case "message-deleted":
    case "reaction-added":
      return { ...event, message: cloneMessage(event.message) };
    case "thread-created":
      return { ...event, thread: { ...event.thread } };
  }
  throw new Error("Unsupported QA bus event kind");
}

export function buildQaBusSnapshot(params: {
  cursor: number;
  conversations: Map<string, QaBusSnapshotConversation>;
  threads: Map<string, QaBusThread>;
  messages: Map<string, QaBusMessage>;
  events: QaBusEvent[];
}): QaBusStateSnapshot {
  return {
    cursor: params.cursor,
    conversations: Array.from(params.conversations.values()).map((conversation) =>
      Object.assign({}, conversation),
    ),
    threads: Array.from(params.threads.values()).map((thread) => Object.assign({}, thread)),
    messages: Array.from(params.messages.values()).map((message) => cloneMessage(message)),
    events: params.events.map((event) => cloneEvent(event)),
  };
}

export function requireQaBusMessageForAccount(params: {
  messages: Map<string, QaBusMessage>;
  input: Pick<QaBusReadMessageInput, "accountId" | "messageId">;
}): QaBusMessage {
  const accountId = normalizeAccountId(params.input.accountId);
  let match: QaBusMessage | undefined;
  for (const message of params.messages.values()) {
    if (message.id !== params.input.messageId || message.accountId !== accountId) {
      continue;
    }
    // Reads have no conversation selector, so never guess which chat to mutate.
    if (match) {
      throw new Error(
        `qa-bus message id is ambiguous for selected account: ${params.input.messageId}`,
      );
    }
    match = message;
  }
  if (!match) {
    throw new Error(`qa-bus message not found: ${params.input.messageId}`);
  }
  return match;
}

export function readQaBusMessage(params: {
  messages: Map<string, QaBusMessage>;
  input: QaBusReadMessageInput;
}) {
  const message = requireQaBusMessageForAccount(params);
  return cloneMessage(message);
}

export function searchQaBusMessages(params: {
  messages: Map<string, QaBusMessage>;
  input: QaBusSearchMessagesInput;
}) {
  const accountId = normalizeAccountId(params.input.accountId);
  const limit = Math.max(1, Math.min(params.input.limit ?? 20, 100));
  const query = normalizeOptionalLowercaseString(params.input.query);
  return Array.from(params.messages.values())
    .filter((message) => message.accountId === accountId && !message.deleted)
    .filter((message) =>
      params.input.conversationId !== undefined
        ? message.conversation.id === params.input.conversationId
        : true,
    )
    .filter((message) =>
      params.input.conversationKind
        ? message.conversation.kind === params.input.conversationKind
        : true,
    )
    .filter((message) =>
      params.input.threadId !== undefined
        ? (message.threadId ?? null) === params.input.threadId
        : true,
    )
    .filter((message) => {
      if (!query) {
        return true;
      }
      const attachmentHaystack = message.attachments ?? [];
      const searchableAttachmentText = attachmentHaystack
        .flatMap((attachment) => [
          attachment.fileName,
          attachment.altText,
          attachment.transcript,
          attachment.mimeType,
        ])
        .filter((value): value is string => Boolean(value))
        .join(" ")
        .toLowerCase();
      const messageText = normalizeOptionalLowercaseString(message.text) ?? "";
      const searchableToolText = (message.toolCalls ?? [])
        .map((toolCall) => toolCall.name)
        .join(" ")
        .toLowerCase();
      return `${messageText} ${searchableAttachmentText} ${searchableToolText}`.includes(query);
    })
    .slice(-limit)
    .map((message) => cloneMessage(message));
}

export function resolveQaBusPollStartCursor(params: {
  currentCursor: number;
  requestedCursor?: number;
}): number {
  const requestedCursor = params.requestedCursor ?? 0;
  return params.currentCursor < requestedCursor ? 0 : requestedCursor;
}

export function pollQaBusEvents(params: {
  events: QaBusEvent[];
  cursor: number;
  input?: QaBusPollInput;
}): QaBusPollResult {
  const accountId = normalizeAccountId(params.input?.accountId);
  const effectiveStartCursor = resolveQaBusPollStartCursor({
    currentCursor: params.cursor,
    requestedCursor: params.input?.cursor,
  });
  const limit = Math.max(1, Math.min(params.input?.limit ?? 100, 500));
  const matchingEvents = params.events.filter(
    (event) => event.accountId === accountId && event.cursor > effectiveStartCursor,
  );
  const page = matchingEvents.slice(0, limit);
  // A limited page may advance only through events it returns. Jumping to the
  // bus-wide cursor would make every remaining account event unreachable.
  const nextCursor = matchingEvents.length > page.length ? page.at(-1)?.cursor : params.cursor;
  return {
    cursor: nextCursor ?? params.cursor,
    events: page.map((event) => cloneEvent(event)),
  };
}