curriculum-project-hub/hub/src/feishu/card/streaming-card.ts

/**
 * Streaming card controller for agent runs.
 *
 * Replaces PatchableTextStream for the agent-run use case. Manages a single
 * Feishu interactive card through the full run lifecycle:
 *
 *   thinking → tool calls → streaming text → complete
 *
 * The card is rebuilt from scratch on each flush (throttled at 400ms) using
 * the trace store for tool steps + accumulated reasoning/text. This is
 * simpler than CardKit 2.0 element-level streaming and works with the
 * existing message.patch API.
 *
 * Lifecycle:
 *   1. appendText(delta)    — streaming answer text
 *   2. appendReasoning(delta) — streaming thinking text
 *   3. onToolStart(name, id)  — register a tool step in the trace store
 *   4. onToolEnd(name, id, input, result?, error?) — complete a tool step
 *   5. finish(finalText)    — flush + transition to complete card
 *   6. fail(errorText)      — flush + transition to error card
 *
 * On finish, markdown image references (`![](url|path)`) are downloaded /
 * read, uploaded to Feishu as message images, and embedded as native card
 * `img` elements so external URLs never hit Feishu content-security checks.
 */
import type { FeishuRuntime, SendMessageOptions } from "../client.js";
import { sendCard, patchCard, sendText, sendLongText } from "../client.js";
import { DEFAULT_MAX_MESSAGE_LENGTH, splitAtBoundary } from "../textStream.js";
import {
  startToolUseTraceRun,
  clearToolUseTraceRun,
  recordToolUseStart,
  recordToolUseEnd,
  getToolUseTraceSteps,
} from "./trace-store.js";
import { buildAgentCard, type CardPhase } from "./builder.js";
import {
  type CardContentSegment,
  maskMarkdownImagesForStreaming,
  materializeAnswerSegments,
  sendImageMessage,
} from "../outboundImages.js";

export interface StreamingCardSink {
  readonly create: (card: Record<string, unknown>) => Promise<string | null>;
  readonly patch: (messageId: string, card: Record<string, unknown>) => Promise<void>;
}

export interface StreamingCardOptions {
  readonly runId: string;
  readonly rt: FeishuRuntime;
  readonly chatId: string;
  readonly sendOptions?: SendMessageOptions | undefined;
  readonly patchIntervalMs: number | undefined;
  readonly maxMessageLength: number | undefined;
  /** Project workspace root; required to resolve local image paths. */
  readonly workspaceRoot?: string | undefined;
  /** Project workspace directory; required to resolve local image paths. */
  readonly workspaceDir?: string | undefined;
  readonly maxImageBytes?: number | undefined;
}

const DEFAULT_PATCH_INTERVAL_MS = 400;

export class StreamingAgentCard {
  private currentMessageId: string | null = null;
  private text = "";
  private reasoningText = "";
  private runStartedAt = Date.now();
  private toolUseElapsedMs: number | undefined;
  private flushChain: Promise<void> = Promise.resolve();
  private flushScheduled = false;
  private lastPatchAt = 0;
  private interrupted = false;

  private readonly runId: string;
  private readonly rt: FeishuRuntime;
  private readonly chatId: string;
  private readonly sendOptions: SendMessageOptions | undefined;
  private readonly patchIntervalMs: number;
  private readonly maxMessageLength: number;
  private readonly workspaceRoot: string | undefined;
  private readonly workspaceDir: string | undefined;
  private readonly maxImageBytes: number | undefined;

  constructor(options: StreamingCardOptions) {
    this.runId = options.runId;
    this.rt = options.rt;
    this.chatId = options.chatId;
    this.sendOptions = options.sendOptions;
    this.patchIntervalMs = options.patchIntervalMs ?? DEFAULT_PATCH_INTERVAL_MS;
    this.maxMessageLength = options.maxMessageLength ?? DEFAULT_MAX_MESSAGE_LENGTH;
    this.workspaceRoot = options.workspaceRoot;
    this.workspaceDir = options.workspaceDir;
    this.maxImageBytes = options.maxImageBytes;
    startToolUseTraceRun(this.runId);
  }

  // --- Public API ---

  appendText(delta: string): void {
    if (delta === "") return;
    this.text += delta;
    this.scheduleFlush();
  }

  appendReasoning(delta: string): void {
    if (delta === "") return;
    this.reasoningText += delta;
    this.scheduleFlush();
  }

  onToolStart(toolName: string, toolUseId: string | undefined): void {
    recordToolUseStart({ runId: this.runId, toolName, toolUseId, input: undefined });
    this.scheduleFlush();
  }

  onToolEnd(params: {
    toolName: string;
    toolUseId: string | undefined;
    input: unknown;
    result: unknown;
    error: string | undefined;
    durationMs: number | undefined;
  }): void {
    recordToolUseEnd({ runId: this.runId, ...params });
    this.scheduleFlush();
  }

  async finish(
    fallbackText: string,
    options: { readonly interrupted?: boolean; readonly footerText?: string | undefined } = {},
  ): Promise<void> {
    await this.flushChain;
    this.interrupted = options.interrupted === true;
    const footerText = options.footerText ?? "";
    const fallbackWithFooter = appendFooter(fallbackText, footerText);
    try {
      let answerText =
        this.text.length > 0 ? appendFooter(this.text, footerText) : fallbackWithFooter;
      this.text = answerText;

      const { segments, unresolved } = await materializeAnswerSegments(answerText, {
        rt: this.rt,
        workspaceRoot: this.workspaceRoot,
        workspaceDir: this.workspaceDir,
        maxImageBytes: this.maxImageBytes,
      });
      if (unresolved.length > 0) {
        this.rt.logger.warn(
          { runId: this.runId, unresolvedCount: unresolved.length, unresolved: unresolved.slice(0, 5) },
          "some answer images could not be uploaded to Feishu",
        );
      }

      let updated = true;
      if (answerText.length > 0 || segments.length > 0) {
        updated = await this.flushCard("complete", answerText, false, segments);
      } else if (this.currentMessageId !== null) {
        updated = await this.flushCard("complete", "", false, []);
      }

      if (!updated) {
        // Card path failed (e.g. residual content policy). Deliver text + standalone images.
        updated = await this.deliverPlainFallback(segments, answerText);
      }
      if (!updated && this.interrupted) {
        await sendText(this.rt, this.chatId, "\u5DF2\u4E2D\u65AD\u5F53\u524D\u8FD0\u884C\u3002", this.sendOptions);
      }
    } finally {
      clearToolUseTraceRun(this.runId);
    }
  }

  async fail(errorText: string): Promise<void> {
    await this.flushChain;
    try {
      this.text = errorText;
      await this.flushCard("complete", errorText, true);
    } finally {
      clearToolUseTraceRun(this.runId);
    }
  }

  // --- Internal flush logic ---

  private scheduleFlush(): void {
    if (this.flushScheduled) return;
    this.flushScheduled = true;
    this.flushChain = this.flushChain.then(async () => {
      this.flushScheduled = false;
      await this.flush();
    });
  }

  private async flush(): Promise<boolean> {
    if (this.currentMessageId === null) {
      const updated = await this.flushCard(this.currentPhase(), this.text);
      this.lastPatchAt = Date.now();
      return updated;
    }

    const now = Date.now();
    if (now - this.lastPatchAt < this.patchIntervalMs) return true;
    this.lastPatchAt = now;
    return this.flushCard(this.currentPhase(), this.text);
  }

  private async flushCard(
    phase: CardPhase,
    text: string,
    isError = false,
    contentSegments?: readonly CardContentSegment[],
  ): Promise<boolean> {
    // During live streaming, strip image URLs so Feishu never fetches remote
    // ranks mid-run. Materialized segments are only used on the complete pass.
    const displayText =
      phase === "complete" && contentSegments !== undefined
        ? text
        : maskMarkdownImagesForStreaming(text);

    const chunks = splitAtBoundary(displayText, this.maxMessageLength);
    const firstChunk = chunks[0] ?? "";
    // When we have segments (complete+images), keep first-card complete content
    // on segments only; overflow text (rare) falls back to plain chunked cards.
    const toolUseSteps = getToolUseTraceSteps(this.runId);
    const card = buildAgentCard({
      phase,
      text: contentSegments !== undefined && contentSegments.length > 0 ? "" : firstChunk,
      contentSegments: contentSegments !== undefined && contentSegments.length > 0
        ? contentSegments
        : undefined,
      reasoningText: this.reasoningText || undefined,
      toolUseSteps,
      toolUseElapsedMs: this.toolUseElapsedMs,
      isError,
      interrupted: this.interrupted,
      runId: this.runId,
    });

    if (this.currentMessageId === null) {
      this.currentMessageId = await sendCard(this.rt, this.chatId, card, this.sendOptions);
      let updated = this.currentMessageId !== null;
      // Send overflow chunks as new messages (rare for agent output). Segments
      // already include the whole answer; only plain text overflows.
      if (contentSegments === undefined || contentSegments.length === 0) {
        for (const chunk of chunks.slice(1)) {
          const overflowCard = buildAgentCard({
            phase,
            text: chunk,
            reasoningText: undefined,
            toolUseSteps: [],
            toolUseElapsedMs: undefined,
            isError,
            interrupted: this.interrupted,
            runId: undefined,
          });
          const overflowMessageId = await sendCard(this.rt, this.chatId, overflowCard, this.sendOptions);
          updated = updated && overflowMessageId !== null;
          this.currentMessageId = overflowMessageId;
        }
      }
      return updated;
    }

    let updated = await patchCard(this.rt, this.currentMessageId, card);
    if (contentSegments === undefined || contentSegments.length === 0) {
      for (const chunk of chunks.slice(1)) {
        const overflowCard = buildAgentCard({
          phase,
          text: chunk,
          reasoningText: undefined,
          toolUseSteps: [],
          toolUseElapsedMs: undefined,
          isError,
          interrupted: this.interrupted,
          runId: undefined,
        });
        const overflowMessageId = await sendCard(this.rt, this.chatId, overflowCard, this.sendOptions);
        updated = updated && overflowMessageId !== null;
        this.currentMessageId = overflowMessageId;
      }
    }
    return updated;
  }

  private async deliverPlainFallback(
    segments: readonly CardContentSegment[],
    answerText: string,
  ): Promise<boolean> {
    const textParts: string[] = [];
    const imageKeys: string[] = [];
    if (segments.length > 0) {
      for (const segment of segments) {
        if (segment.type === "markdown") {
          if (segment.content.trim() !== "") textParts.push(segment.content);
        } else {
          imageKeys.push(segment.imgKey);
        }
      }
    } else if (answerText.trim() !== "") {
      textParts.push(maskMarkdownImagesForStreaming(answerText));
    }

    let any = false;
    if (textParts.length > 0) {
      const messageId = await sendLongText(
        this.rt,
        this.chatId,
        textParts.join("\n\n"),
        this.sendOptions,
      );
      any = messageId !== null;
    }
    for (const imageKey of imageKeys) {
      try {
        const messageId = await sendImageMessage(this.rt, this.chatId, imageKey, this.sendOptions);
        any = any || messageId !== null;
      } catch (error) {
        this.rt.logger.warn(
          { runId: this.runId, err: error instanceof Error ? error.message : String(error) },
          "standalone image fallback failed",
        );
      }
    }
    return any;
  }

  private currentPhase(): CardPhase {
    if (this.text.length > 0) return "streaming";
    if (this.reasoningText.length > 0) return "streaming";
    return "thinking";
  }
}

function appendFooter(text: string, footerText: string): string {
  if (footerText === "") return text;
  if (text === "") return footerText;
  return `${text}\n\n${footerText}`;
}