import type { Client } from "@larksuiteoapi/node-sdk";
import { formatInboundEnvelope, resolveEnvelopeFormatOptions } from "../auto-reply/envelope.js";
import { dispatchReplyWithBufferedBlockDispatcher } from "../auto-reply/reply/provider-dispatcher.js";
import { resolveSessionAgentId } from "../agents/agent-scope.js";
import { parseTextCommand, getHelpMenuText, toSlashCommand } from "../channels/text-commands.js";
import { createReplyPrefixOptions } from "../channels/reply-prefix.js";
import type { ClawdbotConfig } from "../config/config.js";
import { loadConfig } from "../config/config.js";
import { readSessionUpdatedAt, resolveStorePath } from "../config/sessions.js";
import { logVerbose } from "../globals.js";
import { formatErrorMessage } from "../infra/errors.js";
import { getChildLogger } from "../logging.js";
import { isSenderAllowed, normalizeAllowFromWithStore, resolveSenderAllowMatch } from "./access.js";
import { resolveAgentRoute } from "../routing/resolve-route.js";
import {
  resolveFeishuConfig,
  resolveFeishuGroupConfig,
  resolveFeishuGroupEnabled,
  type ResolvedFeishuConfig,
} from "./config.js";
import { resolveFeishuDocsFromMessage } from "./docs.js";
import {
  resolveFeishuMedia,
  downloadPostImages,
  extractPostImageKeys,
  type FeishuMediaRef,
} from "./download.js";
import { readFeishuAllowFromStore, upsertFeishuPairingRequest } from "./pairing-store.js";
import { sendMessageFeishu } from "./send.js";
import { FeishuStreamingSession } from "./streaming-card.js";
import { createTypingIndicatorCallbacks } from "./typing.js";
import { EmbeddedBlockChunker } from "../agents/pi-embedded-block-chunker.js";

const logger = getChildLogger({ module: "feishu-message" });

// Supported message types for processing
// - post: rich text (may contain document links)
const SUPPORTED_MSG_TYPES = new Set(["text", "post", "image", "file", "audio", "media", "sticker"]);

/** Feishu mention structure from SDK */
export type FeishuMention = {
  key?: string;
  id?: {
    union_id?: string;
    user_id?: string;
    open_id?: string;
  };
  name?: string;
  tenant_key?: string;
};

export type ProcessFeishuMessageOptions = {
  cfg?: ClawdbotConfig;
  accountId?: string;
  resolvedConfig?: ResolvedFeishuConfig;
  /** Feishu app credentials for streaming card API */
  credentials?: { appId: string; appSecret: string; domain?: string };
  /** Bot name for streaming card title (optional, defaults to no title) */
  botName?: string;
  /** Bot's own open_id for @mention detection in groups */
  botOpenId?: string;
  /**
   * Bot's app name fetched from /bot/v3/info (Feishu-authoritative).
   * Used as a fallback for @mention detection when botOpenId (application-level
   * open_id) differs from the user-level open_id stored in group @mention events.
   */
  botAppName?: string;
};

export async function processFeishuMessage(
  client: Client,
  data: unknown,
  appId: string,
  options: ProcessFeishuMessageOptions = {},
) {
  const cfg = options.cfg ?? loadConfig();
  const accountId = options.accountId ?? appId;
  const feishuCfg = options.resolvedConfig ?? resolveFeishuConfig({ cfg, accountId });

  // SDK 2.0 schema: data directly contains message, sender, etc.
  const payload = data as Record<string, unknown>;
  const message = (payload.message ?? (payload.event as Record<string, unknown>)?.message) as
    | Record<string, unknown>
    | undefined;
  const sender = (payload.sender ?? (payload.event as Record<string, unknown>)?.sender) as
    | Record<string, unknown>
    | undefined;

  if (!message) {
    logger.warn(`Received event without message field`);
    return;
  }

  const chatId = message.chat_id as string | undefined;
  if (!chatId) {
    logger.warn("Received message without chat_id");
    return;
  }
  const isGroup = message.chat_type === "group";
  const msgType = message.message_type as string | undefined;
  const senderIdObj = sender?.sender_id as Record<string, string> | undefined;
  const senderId = senderIdObj?.open_id || senderIdObj?.user_id || "unknown";
  const senderUnionId = senderIdObj?.union_id;
  const maxMediaBytes = feishuCfg.mediaMaxMb * 1024 * 1024;

  // Resolve agent route
  const route = resolveAgentRoute({
    cfg,
    channel: "feishu",
    accountId,
    peer: {
      kind: isGroup ? "group" : "dm",
      id: isGroup ? chatId : senderId,
    },
  });

  // Check if this is a supported message type
  if (!msgType || !SUPPORTED_MSG_TYPES.has(msgType)) {
    logger.debug(`Skipping unsupported message type: ${msgType ?? "unknown"}`);
    return;
  }

  // Load allowlist from store — merge channel-level, "default"-scoped and accountId-scoped
  // stores. pairing approve writes to meta.accountId-scoped file ("default" by default),
  // but processFeishuMessage uses appId as accountId, so we must read all three.
  const [channelAllowFrom, defaultAllowFrom, accountAllowFrom] = await Promise.all([
    readFeishuAllowFromStore().catch(() => [] as string[]),
    readFeishuAllowFromStore(process.env, "default").catch(() => [] as string[]),
    accountId !== "default"
      ? readFeishuAllowFromStore(process.env, accountId).catch(() => [] as string[])
      : Promise.resolve([] as string[]),
  ]);
  const storeAllowFrom = Array.from(
    new Set([...channelAllowFrom, ...defaultAllowFrom, ...accountAllowFrom]),
  );

  // ===== Access Control =====

  // Group access control
  if (isGroup) {
    // Check if group is enabled
    if (!resolveFeishuGroupEnabled({ cfg, accountId, chatId })) {
      logVerbose(`Blocked feishu group ${chatId} (group disabled)`);
      return;
    }

    const { groupConfig } = resolveFeishuGroupConfig({ cfg, accountId, chatId });

    // Check group-level allowFrom override
    if (groupConfig?.allowFrom) {
      const groupAllow = normalizeAllowFromWithStore({
        allowFrom: groupConfig.allowFrom,
        storeAllowFrom,
      });
      if (!isSenderAllowed({ allow: groupAllow, senderId })) {
        logVerbose(`Blocked feishu group sender ${senderId} (group allowFrom override)`);
        return;
      }
    }

    // Apply groupPolicy
    const groupPolicy = feishuCfg.groupPolicy;
    if (groupPolicy === "disabled") {
      logVerbose(`Blocked feishu group message (groupPolicy: disabled)`);
      return;
    }

    if (groupPolicy === "allowlist") {
      const groupAllow = normalizeAllowFromWithStore({
        allowFrom:
          feishuCfg.groupAllowFrom.length > 0 ? feishuCfg.groupAllowFrom : feishuCfg.allowFrom,
        storeAllowFrom,
      });
      if (!groupAllow.hasEntries) {
        logVerbose(`Blocked feishu group message (groupPolicy: allowlist, no entries)`);
        return;
      }
      if (!isSenderAllowed({ allow: groupAllow, senderId })) {
        logVerbose(`Blocked feishu group sender ${senderId} (groupPolicy: allowlist)`);
        return;
      }
    }
  }

  // DM access control
  if (!isGroup) {
    const dmPolicy = feishuCfg.dmPolicy;

    if (dmPolicy === "disabled") {
      logVerbose(`Blocked feishu DM (dmPolicy: disabled)`);
      return;
    }

    if (dmPolicy !== "open") {
      const dmAllow = normalizeAllowFromWithStore({
        allowFrom: feishuCfg.allowFrom,
        storeAllowFrom,
      });
      const allowMatch = resolveSenderAllowMatch({ allow: dmAllow, senderId });
      const allowed = dmAllow.hasWildcard || (dmAllow.hasEntries && allowMatch.allowed);

      if (!allowed) {
        if (dmPolicy === "pairing") {
          // Generate pairing code for unknown sender
          try {
            const { code, created } = await upsertFeishuPairingRequest({
              openId: senderId,
              unionId: senderUnionId,
              name: senderIdObj?.user_id,
            });
            if (created) {
              logger.info({ openId: senderId, unionId: senderUnionId }, "feishu pairing request");
              await sendMessageFeishu(
                client,
                senderId,
                {
                  text: [
                    "OpenClaw 访问未配置。",
                    "",
                    `你的飞书 Open ID：${senderId}`,
                    "",
                    `配对码：${code}`,
                    "",
                    "请联系 OpenClaw 管理员执行以下命令以批准：",
                    `openclaw-cn pairing approve feishu ${code}`,
                  ].join("\n"),
                },
                { receiveIdType: "open_id" },
              );
            }
          } catch (err) {
            logger.error(`Failed to create pairing request: ${formatErrorMessage(err)}`);
          }
          return;
        }

        // allowlist policy: silently block
        logVerbose(`Blocked feishu DM from ${senderId} (dmPolicy: allowlist)`);
        return;
      }
    }
  }

  // Handle @mentions for group chats
  const mentions: FeishuMention[] =
    (message.mentions as FeishuMention[]) ?? (payload.mentions as FeishuMention[]) ?? [];
  const botOpenId = options.botOpenId;
  // Names to match against for fallback mention detection (when application-level
  // open_id from /bot/v3/info differs from the user-level open_id in mentions).
  const botNames: string[] = [options.botAppName, options.botName]
    .filter((n): n is string => Boolean(n?.trim()))
    .map((n) => n.trim().toLowerCase());
  // Check if the bot itself was mentioned, not just any @mention
  const wasMentioned =
    // Primary: match by open_id from /bot/v3/info
    (botOpenId ? mentions.some((m) => m.id?.open_id === botOpenId) : false) ||
    // Fallback: match by bot name — /bot/v3/info returns application-level open_id
    //   which may differ from the user-level open_id stored in group @mention events;
    //   using the bot name is a reliable secondary check in multi-bot groups.
    (botNames.length > 0
      ? mentions.some(
          (m) => Boolean(m.name?.trim()) && botNames.includes(m.name!.trim().toLowerCase()),
        )
      : false) ||
    // Last resort: any mention present (when neither open_id nor name is available)
    (!botOpenId && botNames.length === 0 && mentions.length > 0);

  // In group chat, check requireMention setting
  if (isGroup) {
    const { groupConfig } = resolveFeishuGroupConfig({ cfg, accountId, chatId });
    const requireMention = groupConfig?.requireMention ?? true;
    if (requireMention && !wasMentioned) {
      logger.debug(`Ignoring group message without @mention (requireMention: true)`);
      return;
    }
  }

  // Extract text content (for text messages or rich text)
  let text = "";
  if (msgType === "text") {
    try {
      const contentStr = message.content as string | undefined;
      if (contentStr) {
        const content = JSON.parse(contentStr);
        text = content.text || "";
      }
    } catch (err) {
      logger.error(`Failed to parse text message content: ${formatErrorMessage(err)}`);
    }
  } else if (msgType === "post") {
    // Extract text from rich text (post) message
    // Handles both direct format { title, content } and locale-wrapped format { post: { zh_cn: { title, content } } }
    try {
      const content = JSON.parse(message.content as string);
      const parts: string[] = [];

      // Try to find the actual post content
      // Format 1: { post: { zh_cn: { title, content } } }
      // Format 2: { title, content } (direct)
      let postData = content;
      if (content.post && typeof content.post === "object") {
        // Find the first locale key (zh_cn, en_us, etc.)
        const localeKey = Object.keys(content.post).find(
          (key: string) => content.post[key]?.content || content.post[key]?.title,
        );
        if (localeKey) {
          postData = content.post[localeKey];
        }
      }

      // Include title if present
      if (postData.title) {
        parts.push(postData.title);
      }

      // Extract text from content elements
      if (Array.isArray(postData.content)) {
        for (const line of postData.content) {
          if (!Array.isArray(line)) continue;
          const lineParts: string[] = [];
          for (const element of line) {
            if (element.tag === "text" && element.text) {
              lineParts.push(element.text);
            } else if (element.tag === "a" && element.text) {
              lineParts.push(element.text);
            } else if (element.tag === "at" && element.user_name) {
              lineParts.push(`@${element.user_name}`);
            }
          }
          if (lineParts.length > 0) {
            parts.push(lineParts.join(""));
          }
        }
      }

      text = parts.join("\n");
    } catch (err) {
      logger.error(`Failed to parse post message content: ${formatErrorMessage(err)}`);
    }
  }

  // Remove @mention placeholders from text
  for (const mention of mentions) {
    if (mention.key) {
      text = text.replace(mention.key, "").trim();
    }
  }

  // ===== Text Command Detection =====
  // Detect help triggers and Chinese command aliases
  const textCommandResult = parseTextCommand(text);

  if (textCommandResult.type === "help") {
    // Respond with help menu
    logger.debug(`Text command detected: help trigger`);
    await sendMessageFeishu(
      client,
      chatId,
      { text: getHelpMenuText() },
      {
        msgType: "text",
        receiveIdType: "chat_id",
      },
    );
    return;
  }

  // Convert Chinese command alias to slash command
  if (textCommandResult.type === "command") {
    const slashCommand = toSlashCommand(textCommandResult);
    if (slashCommand) {
      logger.debug(`Text command detected: ${text} -> ${slashCommand}`);
      text = slashCommand;
    }
  }

  // Resolve media if present (for image, file, audio, media, sticker types)
  let media: FeishuMediaRef | null = null;
  let postImages: FeishuMediaRef[] = [];
  if (!["text", "post"].includes(msgType)) {
    try {
      media = await resolveFeishuMedia(client, message, maxMediaBytes);
    } catch (err) {
      logger.error(`Failed to download media: ${formatErrorMessage(err)}`);
    }
  } else if (msgType === "post") {
    // Download embedded images from post (rich text) message
    try {
      const content = JSON.parse(message.content as string);
      const imageKeys = extractPostImageKeys(content);
      if (imageKeys.length > 0) {
        logger.debug(`Found ${imageKeys.length} embedded images in post message`);
        postImages = await downloadPostImages(
          client,
          message.message_id as string,
          imageKeys,
          maxMediaBytes,
          5, // max 5 images
        );
        logger.debug(`Downloaded ${postImages.length} embedded images`);
      }
    } catch (err) {
      logger.error(`Failed to download post images: ${formatErrorMessage(err)}`);
    }
  }

  // Resolve document content if message contains Feishu doc links
  let docContent: string | null = null;
  if (msgType === "text" || msgType === "post") {
    try {
      docContent = await resolveFeishuDocsFromMessage(client, message, {
        maxDocsPerMessage: 3,
        maxTotalLength: 100000,
      });
      if (docContent) {
        logger.debug(`Resolved ${docContent.length} chars of document content`);
      }
    } catch (err) {
      logger.error(`Failed to resolve document content: ${formatErrorMessage(err)}`);
    }
  }

  // Build body text
  let bodyText = text;
  if (!bodyText && media) {
    bodyText = media.placeholder;
  }
  // If we have embedded images from post message, add placeholders
  if (postImages.length > 0 && !media) {
    const imagePlaceholders = postImages.map(() => "<media:image>").join(" ");
    bodyText = bodyText ? `${bodyText}\n${imagePlaceholders}` : imagePlaceholders;
  }

  // Append document content if available
  if (docContent) {
    bodyText = bodyText ? `${bodyText}\n\n${docContent}` : docContent;
  }

  // Skip if no content
  if (!bodyText && !media && postImages.length === 0) {
    logger.debug(`Empty message after processing, skipping`);
    return;
  }

  // Build sender label (similar to Telegram format)
  const senderName = senderIdObj?.user_id || "unknown";
  const senderOpenId = senderIdObj?.open_id;
  // For DM: use sender info as conversation label
  // For group: use group title + id
  const chat = message.chat as Record<string, string> | undefined;
  const groupTitle = chat?.title || (message.chat_type === "group" ? "Group" : undefined);
  const conversationLabel = isGroup
    ? `${groupTitle} id:${chatId}`
    : senderOpenId
      ? `${senderName} id:${senderOpenId}`
      : senderName;

  // Resolve envelope options and previous timestamp for elapsed time
  const envelopeOptions = resolveEnvelopeFormatOptions(cfg);
  const storePath = resolveStorePath(cfg.session?.store, {
    agentId: route.agentId,
  });
  const previousTimestamp = readSessionUpdatedAt({
    storePath,
    sessionKey: route.sessionKey,
  });

  // Resolve reply-to mode for group chats
  // In group chats, we quote the original message to provide context
  const { groupConfig } = isGroup
    ? resolveFeishuGroupConfig({ cfg, accountId, chatId })
    : { groupConfig: undefined };
  const replyToMode = isGroup ? (groupConfig?.replyToMode ?? feishuCfg.replyToMode) : "off";
  const originalMessageId = message.message_id as string | undefined;
  let hasReplied = false; // Track if we've sent at least one reply (for "first" mode)

  // Streaming mode support
  const streamingEnabled = (feishuCfg.streaming ?? true) && Boolean(options.credentials);
  const streamingSession =
    streamingEnabled && options.credentials
      ? new FeishuStreamingSession(client, options.credentials)
      : null;
  let streamingStarted = false;
  let lastPartialText = "";
  // Chunker for throttling streaming updates (minimize API calls)
  const streamingChunker = streamingSession
    ? new EmbeddedBlockChunker({ minChars: 80, maxChars: 400, breakPreference: "sentence" })
    : null;

  // Typing indicator callbacks (for non-streaming mode)
  const typingCallbacks = createTypingIndicatorCallbacks(client, message.message_id as string);

  // Format body with standardized envelope (consistent with Telegram/WhatsApp)
  const formattedBody = formatInboundEnvelope({
    channel: "Feishu",
    from: conversationLabel,
    timestamp: message.create_time ? Number(message.create_time) * 1000 : undefined,
    body: bodyText,
    chatType: isGroup ? "group" : "direct",
    sender: {
      name: senderName,
      id: senderOpenId || senderId,
    },
    previousTimestamp,
    envelope: envelopeOptions,
  });

  // Context construction
  const ctx = {
    Body: formattedBody,
    RawBody: text || media?.placeholder || "",
    From: senderId,
    To: chatId,
    SessionKey: route.sessionKey,
    SenderId: senderId,
    SenderName: senderName,
    ChatType: isGroup ? "group" : "direct",
    Provider: "feishu",
    Surface: "feishu",
    Timestamp: Number(message.create_time),
    MessageSid: message.message_id as string | undefined,
    AccountId: route.accountId,
    OriginatingChannel: "feishu",
    OriginatingTo: chatId,
    // Media fields (similar to Telegram)
    MediaPath: media?.path ?? postImages[0]?.path,
    MediaType: media?.contentType ?? postImages[0]?.contentType,
    MediaUrl: media?.path ?? postImages[0]?.path,
    // Multiple media from post messages
    MediaPaths: postImages.length > 0 ? postImages.map((img) => img.path) : undefined,
    MediaUrls: postImages.length > 0 ? postImages.map((img) => img.path) : undefined,
    WasMentioned: isGroup ? wasMentioned : undefined,
    // Command authorization - if message passed access control, sender is authorized
    CommandAuthorized: true,
  };

  const agentId = resolveSessionAgentId({ config: cfg });
  const { onModelSelected, ...prefixOptions } = createReplyPrefixOptions({
    cfg,
    agentId,
    channel: "feishu",
    accountId,
  });

  await dispatchReplyWithBufferedBlockDispatcher({
    ctx,
    cfg,
    dispatcherOptions: {
      ...prefixOptions,
      deliver: async (payload, info) => {
        const hasMedia = payload.mediaUrl || (payload.mediaUrls && payload.mediaUrls.length > 0);
        if (!payload.text && !hasMedia) return;

        // Block replies are handled by onPartialReply with chunking/throttling.
        // Only skip when streaming is active to avoid duplicate updates;
        // when streaming is not active, block replies should still be delivered.
        if (info?.kind === "block" && streamingSession?.isActive()) return;

        // If streaming was active, close it with the final text.
        // If close() fails (permissions, API issues), fallback to regular message.
        if (streamingSession?.isActive() && info?.kind === "final") {
          const finalText = payload.text || lastPartialText;
          const closed = await streamingSession.close(finalText);
          streamingStarted = false;
          if (closed) return; // Card already contains the final text
          // Streaming close failed — fall through to sendMessageFeishu()
          logger.warn("Streaming close failed, falling back to regular message");
        }

        // Handle media URLs
        const mediaUrls = payload.mediaUrls?.length
          ? payload.mediaUrls
          : payload.mediaUrl
            ? [payload.mediaUrl]
            : [];

        // Determine if this reply should quote the original message
        const shouldQuote = replyToMode === "all" || (replyToMode === "first" && !hasReplied);
        const replyToMessageId = shouldQuote ? originalMessageId : undefined;

        if (mediaUrls.length > 0) {
          // Close streaming session before sending media
          if (streamingSession?.isActive()) {
            await streamingSession.close();
            streamingStarted = false;
          }
          // Remove typing indicator before sending media
          await typingCallbacks.onIdle();
          // Send each media item
          for (let i = 0; i < mediaUrls.length; i++) {
            const mediaUrl = mediaUrls[i];
            const caption = i === 0 ? payload.text || "" : "";
            // Only quote on the first media item
            const mediaReplyTo = i === 0 ? replyToMessageId : undefined;
            await sendMessageFeishu(
              client,
              chatId,
              { text: caption },
              {
                mediaUrl,
                receiveIdType: "chat_id",
                replyToMessageId: mediaReplyTo,
              },
            );
            if (i === 0) hasReplied = true;
          }
        } else if (payload.text) {
          // If streaming wasn't used, send as regular message
          if (!streamingSession?.isActive()) {
            // Remove typing indicator before sending final reply
            await typingCallbacks.onIdle();
            await sendMessageFeishu(
              client,
              chatId,
              { text: payload.text },
              {
                msgType: "text",
                receiveIdType: "chat_id",
                replyToMessageId,
              },
            );
            hasReplied = true;
          }
        }
      },
      onError: (err) => {
        const msg = formatErrorMessage(err);
        if (
          msg.includes("permission") ||
          msg.includes("forbidden") ||
          msg.includes("code: 99991660")
        ) {
          logger.error(
            `Reply error: ${msg} (Check if "im:message" or "im:resource" permissions are enabled in Feishu Console)`,
          );
        } else {
          logger.error(`Reply error: ${msg}`);
        }
        // Clean up streaming session on error
        if (streamingSession?.isActive()) {
          streamingSession.close().catch(() => {});
        }
        // Clean up typing indicator on error
        typingCallbacks.onIdle().catch(() => {});
      },
      onReplyStart: async () => {
        // Start streaming card when reply generation begins
        if (streamingSession && !streamingStarted) {
          try {
            await streamingSession.start(chatId, "chat_id", options.botName);
            streamingStarted = true;
            logger.debug(`Started streaming card for chat ${chatId}`);
          } catch (err) {
            const msg = formatErrorMessage(err);
            if (msg.includes("permission") || msg.includes("forbidden")) {
              logger.warn(
                `Failed to start streaming card: ${msg} (Check if "im:resource:msg:send" or card permissions are enabled)`,
              );
            } else {
              logger.warn(`Failed to start streaming card: ${msg}`);
            }
            // Continue without streaming
          }
        } else if (!streamingSession) {
          // Non-streaming mode: use typing indicator
          await typingCallbacks.onReplyStart();
        }
      },
    },
    replyOptions: {
      disableBlockStreaming: !feishuCfg.blockStreaming,
      onModelSelected,
      onPartialReply: streamingSession
        ? async (payload) => {
            if (!streamingSession.isActive() || !payload.text) return;
            if (payload.text === lastPartialText) return;
            // Calculate delta from cumulative text
            const delta = payload.text.slice(lastPartialText.length);
            lastPartialText = payload.text;
            if (!delta) return;
            // Capture current text for the emit callback
            const currentText = payload.text;
            // Use chunker to throttle updates
            streamingChunker?.append(delta);
            streamingChunker?.drain({
              force: false,
              emit: () => {
                // Update card with cumulative text (not just the chunk)
                streamingSession.update(currentText).catch((err) => {
                  logger.warn(`Streaming update failed: ${formatErrorMessage(err)}`);
                });
              },
            });
          }
        : undefined,
      onReasoningStream: streamingSession
        ? async (payload) => {
            // Also update on reasoning stream for extended thinking models
            if (!streamingSession.isActive() || !payload.text) return;
            if (payload.text === lastPartialText) return;
            const delta = payload.text.slice(lastPartialText.length);
            lastPartialText = payload.text;
            if (!delta) return;
            const currentText = payload.text;
            streamingChunker?.append(delta);
            streamingChunker?.drain({
              force: false,
              emit: () => {
                streamingSession.update(currentText).catch((err) => {
                  logger.warn(`Reasoning stream update failed: ${formatErrorMessage(err)}`);
                });
              },
            });
          }
        : undefined,
    },
  });

  // Flush any remaining buffered content and close streaming session
  if (streamingSession?.isActive()) {
    // Force drain remaining buffer before closing
    if (streamingChunker?.hasBuffered()) {
      streamingChunker.drain({
        force: true,
        emit: () => {
          streamingSession.update(lastPartialText).catch(() => {});
        },
      });
    }
    // Always close with the complete accumulated text
    await streamingSession.close(lastPartialText || undefined);
  }
}
