/** * 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) => Promise; readonly patch: (messageId: string, card: Record) => Promise; } 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 = 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 { 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 { 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 { 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 { // 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 { 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}`; }