/** * 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 */ import type { FeishuRuntime, SendMessageOptions } from "../client.js"; import { sendCard, patchCard, sendText } 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"; 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; } 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; 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; 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 updated = true; if (this.text.length > 0) { this.text = appendFooter(this.text, footerText); updated = await this.flushCard("complete", this.text); } else if (this.currentMessageId === null && fallbackWithFooter.length > 0) { // No streaming text was sent. If we never created a card, send one now. this.text = fallbackWithFooter; updated = await this.flushCard("complete", this.text); } else if (this.currentMessageId !== null) { // Patch the existing card with the final text. updated = await this.flushCard("complete", fallbackWithFooter); } 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): Promise { const chunks = splitAtBoundary(text, this.maxMessageLength); const firstChunk = chunks[0]; if (firstChunk === undefined) return true; const toolUseSteps = getToolUseTraceSteps(this.runId); const card = buildAgentCard({ phase, text: firstChunk, 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) 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; } else { let updated = await patchCard(this.rt, this.currentMessageId, card); // For overflow, create new messages 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 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.trimEnd()}\n\n${footerText}`; }