feat(feishu): add streaming card support via Card Kit API (openclaw#10379) thanks @xzq-xu

Verified:
- pnpm build
- pnpm check
- pnpm test

Co-authored-by: xzq-xu <53989315+xzq-xu@users.noreply.github.com>
Co-authored-by: Tak Hoffman <781889+Takhoffman@users.noreply.github.com>

extensions/feishu/src/reply-dispatcher.ts

@@ -3,29 +3,22 @@ import {
   createTypingCallbacks,
   logTypingFailure,
   type ClawdbotConfig,
-  type RuntimeEnv,
   type ReplyPayload,
+  type RuntimeEnv,
 } from "openclaw/plugin-sdk";
 import type { MentionTarget } from "./mention.js";
 import { resolveFeishuAccount } from "./accounts.js";
+import { createFeishuClient } from "./client.js";
+import { buildMentionedCardContent } from "./mention.js";
 import { getFeishuRuntime } from "./runtime.js";
-import { sendMessageFeishu, sendMarkdownCardFeishu } from "./send.js";
+import { sendMarkdownCardFeishu, sendMessageFeishu } from "./send.js";
+import { FeishuStreamingSession } from "./streaming-card.js";
+import { resolveReceiveIdType } from "./targets.js";
 import { addTypingIndicator, removeTypingIndicator, type TypingIndicatorState } from "./typing.js";
 
-/**
- * Detect if text contains markdown elements that benefit from card rendering.
- * Used by auto render mode.
- */
+/** Detect if text contains markdown elements that benefit from card rendering */
 function shouldUseCard(text: string): boolean {
-  // Code blocks (fenced)
-  if (/```[\s\S]*?```/.test(text)) {
-    return true;
-  }
-  // Tables (at least header + separator row with |)
-  if (/\|.+\|[\r\n]+\|[-:| ]+\|/.test(text)) {
-    return true;
-  }
-  return false;
+  return /```[\s\S]*?```/.test(text) || /\|.+\|[\r\n]+\|[-:| ]+\|/.test(text);
 }
 
 export type CreateFeishuReplyDispatcherParams = {
@@ -34,35 +27,23 @@ export type CreateFeishuReplyDispatcherParams = {
   runtime: RuntimeEnv;
   chatId: string;
   replyToMessageId?: string;
-  /** Mention targets, will be auto-included in replies */
   mentionTargets?: MentionTarget[];
-  /** Account ID for multi-account support */
   accountId?: string;
 };
 
 export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherParams) {
   const core = getFeishuRuntime();
   const { cfg, agentId, chatId, replyToMessageId, mentionTargets, accountId } = params;
-
-  // Resolve account for config access
   const account = resolveFeishuAccount({ cfg, accountId });
+  const prefixContext = createReplyPrefixContext({ cfg, agentId });
 
-  const prefixContext = createReplyPrefixContext({
-    cfg,
-    agentId,
-  });
-
-  // Feishu doesn't have a native typing indicator API.
-  // We use message reactions as a typing indicator substitute.
   let typingState: TypingIndicatorState | null = null;
-
   const typingCallbacks = createTypingCallbacks({
     start: async () => {
       if (!replyToMessageId) {
         return;
       }
       typingState = await addTypingIndicator({ cfg, messageId: replyToMessageId, accountId });
-      params.runtime.log?.(`feishu[${account.accountId}]: added typing indicator reaction`);
     },
     stop: async () => {
       if (!typingState) {
@@ -70,24 +51,21 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP
       }
       await removeTypingIndicator({ cfg, state: typingState, accountId });
       typingState = null;
-      params.runtime.log?.(`feishu[${account.accountId}]: removed typing indicator reaction`);
     },
-    onStartError: (err) => {
+    onStartError: (err) =>
       logTypingFailure({
         log: (message) => params.runtime.log?.(message),
         channel: "feishu",
         action: "start",
         error: err,
-      });
-    },
-    onStopError: (err) => {
+      }),
+    onStopError: (err) =>
       logTypingFailure({
         log: (message) => params.runtime.log?.(message),
         channel: "feishu",
         action: "stop",
         error: err,
-      });
-    },
+      }),
   });
 
   const textChunkLimit = core.channel.text.resolveTextChunkLimit(cfg, "feishu", accountId, {
@@ -95,77 +73,139 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP
   });
   const chunkMode = core.channel.text.resolveChunkMode(cfg, "feishu");
   const tableMode = core.channel.text.resolveMarkdownTableMode({ cfg, channel: "feishu" });
+  const renderMode = account.config?.renderMode ?? "auto";
+  const streamingEnabled = account.config?.streaming !== false && renderMode !== "raw";
+
+  let streaming: FeishuStreamingSession | null = null;
+  let streamText = "";
+  let lastPartial = "";
+  let partialUpdateQueue: Promise<void> = Promise.resolve();
+  let streamingStartPromise: Promise<void> | null = null;
+
+  const startStreaming = () => {
+    if (!streamingEnabled || streamingStartPromise || streaming) {
+      return;
+    }
+    streamingStartPromise = (async () => {
+      const creds =
+        account.appId && account.appSecret
+          ? { appId: account.appId, appSecret: account.appSecret, domain: account.domain }
+          : null;
+      if (!creds) {
+        return;
+      }
+
+      streaming = new FeishuStreamingSession(createFeishuClient(account), creds, (message) =>
+        params.runtime.log?.(`feishu[${account.accountId}] ${message}`),
+      );
+      try {
+        await streaming.start(chatId, resolveReceiveIdType(chatId));
+      } catch (error) {
+        params.runtime.error?.(`feishu: streaming start failed: ${String(error)}`);
+        streaming = null;
+      }
+    })();
+  };
+
+  const closeStreaming = async () => {
+    if (streamingStartPromise) {
+      await streamingStartPromise;
+    }
+    await partialUpdateQueue;
+    if (streaming?.isActive()) {
+      let text = streamText;
+      if (mentionTargets?.length) {
+        text = buildMentionedCardContent(mentionTargets, text);
+      }
+      await streaming.close(text);
+    }
+    streaming = null;
+    streamingStartPromise = null;
+    streamText = "";
+    lastPartial = "";
+  };
 
   const { dispatcher, replyOptions, markDispatchIdle } =
     core.channel.reply.createReplyDispatcherWithTyping({
       responsePrefix: prefixContext.responsePrefix,
       responsePrefixContextProvider: prefixContext.responsePrefixContextProvider,
       humanDelay: core.channel.reply.resolveHumanDelayConfig(cfg, agentId),
-      onReplyStart: typingCallbacks.onReplyStart,
-      deliver: async (payload: ReplyPayload) => {
-        params.runtime.log?.(
-          `feishu[${account.accountId}] deliver called: text=${payload.text?.slice(0, 100)}`,
-        );
+      onReplyStart: () => {
+        if (streamingEnabled && renderMode === "card") {
+          startStreaming();
+        }
+        void typingCallbacks.onReplyStart?.();
+      },
+      deliver: async (payload: ReplyPayload, info) => {
         const text = payload.text ?? "";
         if (!text.trim()) {
-          params.runtime.log?.(`feishu[${account.accountId}] deliver: empty text, skipping`);
           return;
         }
 
-        // Check render mode: auto (default), raw, or card
-        const feishuCfg = account.config;
-        const renderMode = feishuCfg?.renderMode ?? "auto";
-
-        // Determine if we should use card for this message
         const useCard = renderMode === "card" || (renderMode === "auto" && shouldUseCard(text));
 
-        // Only include @mentions in the first chunk (avoid duplicate @s)
-        let isFirstChunk = true;
+        if ((info?.kind === "block" || info?.kind === "final") && streamingEnabled && useCard) {
+          startStreaming();
+          if (streamingStartPromise) {
+            await streamingStartPromise;
+          }
+        }
 
+        if (streaming?.isActive()) {
+          if (info?.kind === "final") {
+            streamText = text;
+            await closeStreaming();
+          }
+          return;
+        }
+
+        let first = true;
         if (useCard) {
-          // Card mode: send as interactive card with markdown rendering
-          const chunks = core.channel.text.chunkTextWithMode(text, textChunkLimit, chunkMode);
-          params.runtime.log?.(
-            `feishu[${account.accountId}] deliver: sending ${chunks.length} card chunks to ${chatId}`,
-          );
-          for (const chunk of chunks) {
+          for (const chunk of core.channel.text.chunkTextWithMode(
+            text,
+            textChunkLimit,
+            chunkMode,
+          )) {
             await sendMarkdownCardFeishu({
               cfg,
               to: chatId,
               text: chunk,
               replyToMessageId,
-              mentions: isFirstChunk ? mentionTargets : undefined,
+              mentions: first ? mentionTargets : undefined,
               accountId,
             });
-            isFirstChunk = false;
+            first = false;
           }
         } else {
-          // Raw mode: send as plain text with table conversion
           const converted = core.channel.text.convertMarkdownTables(text, tableMode);
-          const chunks = core.channel.text.chunkTextWithMode(converted, textChunkLimit, chunkMode);
-          params.runtime.log?.(
-            `feishu[${account.accountId}] deliver: sending ${chunks.length} text chunks to ${chatId}`,
-          );
-          for (const chunk of chunks) {
+          for (const chunk of core.channel.text.chunkTextWithMode(
+            converted,
+            textChunkLimit,
+            chunkMode,
+          )) {
             await sendMessageFeishu({
               cfg,
               to: chatId,
               text: chunk,
               replyToMessageId,
-              mentions: isFirstChunk ? mentionTargets : undefined,
+              mentions: first ? mentionTargets : undefined,
               accountId,
             });
-            isFirstChunk = false;
+            first = false;
           }
         }
       },
-      onError: (err, info) => {
+      onError: async (error, info) => {
         params.runtime.error?.(
-          `feishu[${account.accountId}] ${info.kind} reply failed: ${String(err)}`,
+          `feishu[${account.accountId}] ${info.kind} reply failed: ${String(error)}`,
         );
+        await closeStreaming();
+        typingCallbacks.onIdle?.();
+      },
+      onIdle: async () => {
+        await closeStreaming();
         typingCallbacks.onIdle?.();
       },
-      onIdle: typingCallbacks.onIdle,
     });
 
   return {
@@ -173,6 +213,23 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP
     replyOptions: {
       ...replyOptions,
       onModelSelected: prefixContext.onModelSelected,
+      onPartialReply: streamingEnabled
+        ? (payload: ReplyPayload) => {
+            if (!payload.text || payload.text === lastPartial) {
+              return;
+            }
+            lastPartial = payload.text;
+            streamText = payload.text;
+            partialUpdateQueue = partialUpdateQueue.then(async () => {
+              if (streamingStartPromise) {
+                await streamingStartPromise;
+              }
+              if (streaming?.isActive()) {
+                await streaming.update(streamText);
+              }
+            });
+          }
+        : undefined,
     },
     markDispatchIdle,
   };