From d6724e5a834e6ef689df6603a9c0b36e9a65e729 Mon Sep 17 00:00:00 2001 From: Hong Jiarong Date: Thu, 9 Jul 2026 17:18:40 +0800 Subject: [PATCH] feat: streaming agent card with tool-use trace, reasoning panel, and tool results MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace text-only PatchableTextStream with a unified StreamingAgentCard that renders the full agent run lifecycle in a single Feishu interactive card: - Tool-use panel: collapsible, shows each tool call with status (running/ success/error), input summary, and result/error in code blocks - Reasoning panel: collapsible, streams thinking text inline then collapses on completion - Answer text: rendered via native Feishu markdown component (no post-row round-trip) Runner now captures previously-discarded SDK events: - thinking_delta → reasoning stream - tool_use id + input on content_block_start / assistant message - tool_result from user messages (was entirely unhandled) New files: - card/trace-store.ts: per-run in-memory tool-use trace with secret redaction - card/builder.ts: Feishu card JSON builder (tool panel + reasoning + text) - card/streaming-card.ts: StreamingAgentCard controller (400ms throttled) client.ts: add sendCard/patchCard raw JSON primitives --- hub/src/agent/runner.ts | 74 ++++- hub/src/feishu/card/builder.ts | 376 ++++++++++++++++++++++++++ hub/src/feishu/card/streaming-card.ts | 208 ++++++++++++++ hub/src/feishu/card/trace-store.ts | 225 +++++++++++++++ hub/src/feishu/client.ts | 32 +++ hub/src/feishu/trigger.ts | 66 +++-- hub/test/integration/helpers.ts | 15 +- 7 files changed, 964 insertions(+), 32 deletions(-) create mode 100644 hub/src/feishu/card/builder.ts create mode 100644 hub/src/feishu/card/streaming-card.ts create mode 100644 hub/src/feishu/card/trace-store.ts diff --git a/hub/src/agent/runner.ts b/hub/src/agent/runner.ts index 6ece973..e08cb58 100644 --- a/hub/src/agent/runner.ts +++ b/hub/src/agent/runner.ts @@ -28,7 +28,7 @@ * is the mechanism, the contract pins the invariant. */ import { homedir } from "node:os"; -import { query, type McpServerConfig, type SDKMessage, type SDKAssistantMessage, type SDKResultMessage, type SDKPartialAssistantMessage } from "@anthropic-ai/claude-agent-sdk"; +import { query, type McpServerConfig, type SDKMessage, type SDKAssistantMessage, type SDKUserMessage, type SDKResultMessage, type SDKPartialAssistantMessage } from "@anthropic-ai/claude-agent-sdk"; import type { PrismaClient } from "@prisma/client"; export interface ProjectContext { @@ -39,8 +39,10 @@ export interface ProjectContext { export type StreamEvent = | { readonly type: "text-delta"; readonly text: string } - | { readonly type: "tool-start"; readonly toolName: string } - | { readonly type: "tool-end"; readonly toolName: string } + | { readonly type: "thinking-delta"; readonly text: string } + | { readonly type: "tool-start"; readonly toolName: string; readonly toolUseId: string } + | { readonly type: "tool-end"; readonly toolName: string; readonly toolUseId: string; readonly input: unknown; readonly durationMs?: number } + | { readonly type: "tool-result"; readonly toolUseId: string; readonly toolName: string; readonly result: string; readonly isError: boolean; readonly durationMs?: number } | { readonly type: "finish" }; export type StreamCallback = (event: StreamEvent) => void; @@ -156,25 +158,30 @@ export async function runAgent(req: RunRequest): Promise { options, }); + // Track tool start timestamps for duration calculation + const toolStartTimestamps = new Map(); + for await (const message of conversation) { switch (message.type) { case "stream_event": { - // Partial messages — real-time text deltas + tool call chunks. const evt = (message as SDKPartialAssistantMessage).event; if (evt.type === "content_block_delta" && evt.delta.type === "text_delta") { onStream?.({ type: "text-delta", text: evt.delta.text }); } + if (evt.type === "content_block_delta" && evt.delta.type === "thinking_delta") { + onStream?.({ type: "thinking-delta", text: evt.delta.thinking }); + } if (evt.type === "content_block_start" && evt.content_block.type === "tool_use") { - onStream?.({ type: "tool-start", toolName: evt.content_block.name }); + const toolUseId = evt.content_block.id; + toolStartTimestamps.set(toolUseId, Date.now()); + onStream?.({ type: "tool-start", toolName: evt.content_block.name, toolUseId }); } if (evt.type === "content_block_stop") { - // Tool end is hard to attribute from stream events alone; - // the assistant message below carries tool_use blocks. + // Tool end is attributed from the assistant message below. } break; } case "assistant": { - // Complete assistant message (one turn done). Extract text + usage. const msg = (message as SDKAssistantMessage).message; await heartbeat(); numTurns++; @@ -185,7 +192,20 @@ export async function runAgent(req: RunRequest): Promise { assistantText += block.text; } if (block.type === "tool_use") { - onStream?.({ type: "tool-end", toolName: block.name }); + const durationMs = toolStartTimestamps.has(block.id) + ? Date.now() - (toolStartTimestamps.get(block.id) ?? 0) + : undefined; + onStream?.({ + type: "tool-end", + toolName: block.name, + toolUseId: block.id, + input: block.input, + ...(durationMs !== undefined ? { durationMs } : {}), + }); + } + if (block.type === "thinking" && typeof block.thinking === "string") { + // Complete thinking block from the assistant message. + // Stream deltas already delivered this; no separate event needed. } } await persistAgentMessage(req, "assistant", assistantText); @@ -195,6 +215,30 @@ export async function runAgent(req: RunRequest): Promise { } break; } + case "user": { + // User messages carry tool_result blocks — what each tool returned. + const msg = (message as SDKUserMessage).message; + const content = msg.content; + if (!Array.isArray(content)) break; + for (const block of content) { + if (typeof block !== "object" || block === null || block.type !== "tool_result") continue; + // Narrowed to BetaToolResultBlockParam by discriminant + field presence + if (!("tool_use_id" in block)) continue; + const toolUseId = block.tool_use_id; + const isError = block.is_error === true; + const resultText = extractToolResultText(block.content); + const durationMs = toolStartTimestamps.get(toolUseId); + onStream?.({ + type: "tool-result", + toolUseId, + toolName: toolUseId, + result: resultText, + isError, + ...(durationMs !== undefined ? { durationMs: Date.now() - durationMs } : {}), + }); + } + break; + } case "result": { const result = message as SDKResultMessage; sdkSessionId = result.session_id; @@ -243,3 +287,15 @@ async function persistAgentMessage(req: RunRequest, role: string, content: strin // Best-effort projection: a history write failure must not break the run. } } + +function extractToolResultText(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + const parts: string[] = []; + for (const part of content) { + if (typeof part === "object" && part !== null && "type" in part && part.type === "text" && "text" in part && typeof part.text === "string") { + parts.push(part.text); + } + } + return parts.join("\n"); +} diff --git a/hub/src/feishu/card/builder.ts b/hub/src/feishu/card/builder.ts new file mode 100644 index 0000000..d3e44de --- /dev/null +++ b/hub/src/feishu/card/builder.ts @@ -0,0 +1,376 @@ +/** + * Feishu interactive card builder for agent run output. + * + * Produces card JSON with: + * 1. A collapsible tool-use panel (tool steps with status, params, results) + * 2. A collapsible reasoning panel (thinking text) + * 3. The streaming/final answer text (markdown) + * + * Adapted from openclaw-lark's builder.ts, simplified for our + * message.patch-based approach (no CardKit 2.0 streaming_mode). + * Uses JSON 1.0 card format with `collapsible_panel` and `markdown` tags. + */ + +import type { ToolUseTraceStep } from "./trace-store.js"; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +export interface FeishuCard { + readonly config: Record; + readonly elements: ReadonlyArray; +} + +export type CardPhase = "thinking" | "streaming" | "complete"; + +const STEP_INDENT = "0px 0px 0px 22px"; +const MAX_TEXT_LENGTH = 7500; // Feishu card content limit ~30k; leave room for panels + +// --------------------------------------------------------------------------- +// Tool name → icon mapping +// --------------------------------------------------------------------------- + +const TOOL_ICONS: Record = { + read: "doc-filled", + write: "edit-filled", + bash: "terminal-filled", + glob: "search-filled", + grep: "search-filled", + edit: "edit-filled", + send_file: "send-filled", + request_approval: "thumb-up-filled", + feishu_read_context: "search-filled", +}; + +function toolIcon(toolName: string): string { + const normalized = toolName.toLowerCase().replace(/^mcp_/, ""); + return TOOL_ICONS[normalized] ?? "tool-filled"; +} + +// --------------------------------------------------------------------------- +// Card builder +// --------------------------------------------------------------------------- + +export function buildAgentCard(params: { + phase: CardPhase; + text: string; + reasoningText: string | undefined; + toolUseSteps: ToolUseTraceStep[]; + toolUseElapsedMs: number | undefined; + isError: boolean | undefined; + runId: string | undefined; +}): Record { + const { phase, text, reasoningText, toolUseSteps, toolUseElapsedMs, isError } = params; + const elements: unknown[] = []; + + // Tool-use panel (always present if there are steps) + if (toolUseSteps.length > 0) { + elements.push(buildToolUsePanel(toolUseSteps, toolUseElapsedMs, phase !== "complete")); + } else if (phase === "thinking" || (phase === "streaming" && text === "")) { + elements.push(buildPendingToolUsePanel()); + } + + // Reasoning panel + if (reasoningText !== undefined && reasoningText !== "") { + if (phase === "streaming" && text === "") { + // Still thinking: show reasoning inline + elements.push({ + tag: "markdown", + content: `\u{1F4AD} \u601d\u8003\u4E2D...\n\n${reasoningText}`, + text_size: "notation", + }); + } else if (phase === "complete") { + // Collapsed reasoning panel in complete state + elements.push(buildReasoningPanel(reasoningText)); + } + } + + // Main text content + if (text !== "") { + elements.push({ + tag: "markdown", + content: truncateText(text, MAX_TEXT_LENGTH), + }); + } else if (phase === "thinking" && toolUseSteps.length === 0 && (reasoningText === undefined || reasoningText === "")) { + elements.push({ + tag: "markdown", + content: "\u{1F4AD} \u601d\u8003\u4E2D...", + text_size: "notation", + }); + } + + // Footer for complete state + if (phase === "complete") { + const footerParts: string[] = []; + if (isError === true) footerParts.push("\u274C \u5931\u8D25"); + else footerParts.push("\u2705 \u5B8C\u6210"); + if (toolUseElapsedMs !== undefined) { + footerParts.push(formatElapsed(toolUseElapsedMs)); + } + elements.push({ + tag: "div", + text: { + tag: "plain_text", + content: footerParts.join(" \u00B7 "), + text_size: "notation", + text_color: "grey", + }, + }); + } + + return { + config: { wide_screen_mode: true, update_multi: true }, + elements, + }; +} + +// --------------------------------------------------------------------------- +// Tool-use panel +// --------------------------------------------------------------------------- + +function buildPendingToolUsePanel(): unknown { + return { + tag: "collapsible_panel", + expanded: false, + header: { + title: { + tag: "plain_text", + content: "\u{1F6E0}\uFE0F \u7B49\u5F85\u5DE5\u5177\u6267\u884C", + text_color: "grey", + text_size: "notation", + }, + vertical_align: "center", + icon: { + tag: "standard_icon", + token: "down-small-ccm_outlined", + color: "grey", + size: "16px 16px", + }, + icon_position: "right", + icon_expanded_angle: -180, + }, + border: { color: "grey", corner_radius: "5px" }, + vertical_spacing: "4px", + padding: "8px 8px 8px 8px", + elements: [], + }; +} + +function buildToolUsePanel(steps: ToolUseTraceStep[], elapsedMs?: number, expanded = false): unknown { + const parts: string[] = ["\u{1F6E0}\uFE0F \u5DE5\u5177\u6267\u884C"]; + if (steps.length > 0) { + parts.push(`${steps.length} \u6B65`); + } + if (elapsedMs !== undefined && elapsedMs > 0) { + parts.push(`(${formatElapsed(elapsedMs)})`); + } + + const stepElements = steps.flatMap((step) => buildToolStepElements(step)); + + return { + tag: "collapsible_panel", + expanded, + header: { + title: { + tag: "plain_text", + content: parts.join(" \u00B7 "), + text_color: "grey", + text_size: "notation", + }, + vertical_align: "center", + icon: { + tag: "standard_icon", + token: "down-small-ccm_outlined", + color: "grey", + size: "16px 16px", + }, + icon_position: "right", + icon_expanded_angle: -180, + }, + border: { color: "grey", corner_radius: "5px" }, + vertical_spacing: "4px", + padding: "8px 8px 8px 8px", + elements: stepElements, + }; +} + +function buildToolStepElements(step: ToolUseTraceStep): unknown[] { + const elements: unknown[] = [buildToolStepTitle(step)]; + const detailElement = buildToolStepDetail(step); + if (detailElement !== undefined) { + elements.push(detailElement); + } + const outputElement = buildToolStepOutput(step); + if (outputElement !== undefined) { + elements.push(outputElement); + } + return elements; +} + +function buildToolStepTitle(step: ToolUseTraceStep): unknown { + const status = formatStepStatus(step.status); + return { + tag: "div", + icon: { + tag: "standard_icon", + token: toolIcon(step.toolName), + color: "grey", + }, + text: { + tag: "lark_md", + content: `**${escapeMarkdown(step.toolName)}** \u00B7 ${status.label}`, + text_size: "notation", + }, + }; +} + +function buildToolStepDetail(step: ToolUseTraceStep): unknown | undefined { + const detail = extractStepDetail(step); + if (detail === undefined) return undefined; + return { + tag: "div", + margin: STEP_INDENT, + text: { + tag: "plain_text", + content: detail, + text_color: "grey", + text_size: "notation", + }, + }; +} + +function buildToolStepOutput(step: ToolUseTraceStep): unknown | undefined { + const content = buildStepOutputMarkdown(step); + if (content === undefined) return undefined; + return { + tag: "div", + margin: STEP_INDENT, + text: { + tag: "lark_md", + content, + text_size: "notation", + }, + }; +} + +function buildStepOutputMarkdown(step: ToolUseTraceStep): string | undefined { + const lines: string[] = []; + if (step.error !== undefined) { + lines.push("**\u9519\u8BEF**"); + lines.push(formatCodeBlock(step.error, "text")); + } else if (step.result !== undefined && step.result !== "") { + const resultStr = typeof step.result === "string" ? step.result : JSON.stringify(step.result, null, 2); + lines.push("**\u7ED3\u679C**"); + lines.push(formatCodeBlock(resultStr, typeof step.result === "string" ? "text" : "json")); + } + if (lines.length === 0) return undefined; + return lines.join("\n"); +} + +function extractStepDetail(step: ToolUseTraceStep): string | undefined { + if (step.input === undefined || step.input === null) return undefined; + if (typeof step.input === "string") return truncateText(step.input, 200); + const input = step.input as Record; + // Extract the most relevant field based on tool name + const toolLower = step.toolName.toLowerCase().replace(/^mcp_/, ""); + if (toolLower === "bash" && typeof input.command === "string") { + return truncateText(input.command, 200); + } + if (toolLower === "read" && typeof input.file_path === "string") { + return input.file_path; + } + if (toolLower === "write" && typeof input.file_path === "string") { + return input.file_path; + } + if (toolLower === "edit" && typeof input.file_path === "string") { + return input.file_path; + } + if (toolLower === "glob" && typeof input.pattern === "string") { + return input.pattern; + } + if (toolLower === "grep" && typeof input.pattern === "string") { + return input.pattern; + } + // Fallback: show keys + const keys = Object.keys(input); + if (keys.length === 0) return undefined; + return keys.slice(0, 4).join(", "); +} + +// --------------------------------------------------------------------------- +// Reasoning panel +// --------------------------------------------------------------------------- + +function buildReasoningPanel(reasoningText: string): unknown { + return { + tag: "collapsible_panel", + expanded: false, + header: { + title: { + tag: "markdown", + content: "\u{1F4AD} \u601D\u8003", + text_color: "grey", + }, + vertical_align: "center", + icon: { + tag: "standard_icon", + token: "down-small-ccm_outlined", + size: "16px 16px", + }, + icon_position: "follow_text", + icon_expanded_angle: -180, + }, + border: { color: "grey", corner_radius: "5px" }, + vertical_spacing: "8px", + padding: "8px 8px 8px 8px", + elements: [ + { + tag: "markdown", + content: truncateText(reasoningText, MAX_TEXT_LENGTH), + text_size: "notation", + }, + ], + }; +} + +// --------------------------------------------------------------------------- +// Formatting helpers +// --------------------------------------------------------------------------- + +function formatStepStatus(status: ToolUseTraceStep["status"]): { label: string; color: string } { + switch (status) { + case "running": + return { label: "\u8FD0\u884C\u4E2D", color: "turquoise" }; + case "error": + return { label: "\u5931\u8D25", color: "red" }; + case "success": + default: + return { label: "\u6210\u529F", color: "green" }; + } +} + +function formatCodeBlock(content: string, language: "json" | "text"): string { + const normalized = content.replace(/\r\n/g, "\n").trim(); + const fence = "`".repeat(Math.max(3, longestBacktickRun(normalized) + 1)); + return `${fence}${language}\n${normalized}\n${fence}`; +} + +function longestBacktickRun(value: string): number { + const matches = value.match(/`+/g); + if (matches === null) return 0; + return matches.reduce((max, run) => Math.max(max, run.length), 0); +} + +function escapeMarkdown(value: string): string { + return value.replace(/\\/g, "\\\\").replace(/([`*_{}[\]<>])/g, "\\$1"); +} + +function truncateText(value: string, maxLength: number): string { + return value.length <= maxLength ? value : `${value.slice(0, maxLength - 3)}...`; +} + +function formatElapsed(ms: number): string { + if (ms < 1000) return `${ms} ms`; + return `${(ms / 1000).toFixed(1)} s`; +} diff --git a/hub/src/feishu/card/streaming-card.ts b/hub/src/feishu/card/streaming-card.ts new file mode 100644 index 0000000..8b07d58 --- /dev/null +++ b/hub/src/feishu/card/streaming-card.ts @@ -0,0 +1,208 @@ +/** + * 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 } from "../client.js"; +import { sendCard, patchCard } 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 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 readonly runId: string; + private readonly rt: FeishuRuntime; + private readonly chatId: string; + private readonly patchIntervalMs: number; + private readonly maxMessageLength: number; + + constructor(options: StreamingCardOptions) { + this.runId = options.runId; + this.rt = options.rt; + this.chatId = options.chatId; + 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): Promise { + await this.flushChain; + if (this.text.length > 0) { + await this.flushCard("complete", this.text); + return; + } + // No streaming text was sent. If we never created a card, send one now. + if (this.currentMessageId === null && fallbackText.length > 0) { + this.text = fallbackText; + await this.flushCard("complete", this.text); + } else if (this.currentMessageId !== null) { + // Patch the existing card with the final text + await this.flushCard("complete", fallbackText); + } + clearToolUseTraceRun(this.runId); + } + + async fail(errorText: string): Promise { + await this.flushChain; + this.text = errorText; + await this.flushCard("complete", errorText, true); + 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) { + await this.flushCard(this.currentPhase(), this.text); + this.lastPatchAt = Date.now(); + return; + } + + const now = Date.now(); + if (now - this.lastPatchAt < this.patchIntervalMs) return; + this.lastPatchAt = now; + await 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; + + const toolUseSteps = getToolUseTraceSteps(this.runId); + const card = buildAgentCard({ + phase, + text: firstChunk, + reasoningText: this.reasoningText || undefined, + toolUseSteps, + toolUseElapsedMs: this.toolUseElapsedMs, + isError, + runId: this.runId, + }); + + if (this.currentMessageId === null) { + this.currentMessageId = await sendCard(this.rt, this.chatId, card); + // 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, + runId: this.runId, + }); + this.currentMessageId = await sendCard(this.rt, this.chatId, overflowCard); + } + } else { + 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, + runId: this.runId, + }); + this.currentMessageId = await sendCard(this.rt, this.chatId, overflowCard); + } + } + } + + private currentPhase(): CardPhase { + if (this.text.length > 0) return "streaming"; + if (this.reasoningText.length > 0) return "streaming"; + return "thinking"; + } +} diff --git a/hub/src/feishu/card/trace-store.ts b/hub/src/feishu/card/trace-store.ts new file mode 100644 index 0000000..9058a60 --- /dev/null +++ b/hub/src/feishu/card/trace-store.ts @@ -0,0 +1,225 @@ +/** + * In-memory tool-use trace store, adapted from openclaw-lark's + * tool-use-trace-store. One trace per agent run; the card builder + * reads steps from here to render the tool-use panel. + * + * The store is per-run (not per-session) because each run has its + * own card lifecycle. Steps are capped to prevent unbounded memory. + */ + +export interface ToolUseTraceStep { + readonly id: string; + readonly seq: number; + readonly toolName: string; + readonly toolUseId: string | undefined; + readonly input: unknown; + readonly result: unknown; + readonly error: string | undefined; + readonly durationMs: number | undefined; + readonly status: "running" | "success" | "error"; + readonly startedAt: number; + readonly finishedAt: number | undefined; +} + +interface RunTraceState { + nextSeq: number; + updatedAt: number; + steps: ToolUseTraceStep[]; +} + +const MAX_STEPS = 64; +const STEP_RUNNING_TIMEOUT_MS = 5 * 60 * 1000; +const RUN_TTL_MS = 30 * 60 * 1000; +const RESULT_LIMIT = 2048; +const GENERIC_LIMIT = 512; + +const runs = new Map(); + +export function startToolUseTraceRun(runId: string): void { + pruneStaleRuns(); + runs.set(runId, { nextSeq: 1, updatedAt: Date.now(), steps: [] }); +} + +export function clearToolUseTraceRun(runId: string): void { + runs.delete(runId); +} + +export function recordToolUseStart(params: { + runId: string; + toolName: string; + toolUseId: string | undefined; + input: unknown; +}): void { + const state = runs.get(params.runId); + if (state === undefined) return; + + if (state.steps.length >= MAX_STEPS) { + state.steps.splice(0, state.steps.length - MAX_STEPS + 1); + } + + const now = Date.now(); + state.steps.push({ + id: `${state.nextSeq}`, + seq: state.nextSeq, + toolName: params.toolName, + toolUseId: params.toolUseId, + input: sanitizeTraceValue(params.input), + result: undefined, + error: undefined, + durationMs: undefined, + status: "running", + startedAt: now, + finishedAt: undefined, + }); + state.nextSeq += 1; + state.updatedAt = now; +} + +export function recordToolUseEnd(params: { + runId: string; + toolName: string; + toolUseId: string | undefined; + result: unknown; + error: string | undefined; + durationMs: number | undefined; +}): void { + const state = runs.get(params.runId); + if (state === undefined) return; + + const now = Date.now(); + const pendingIndex = findPendingStepIndex(state.steps, params.toolUseId, params.toolName); + + if (pendingIndex >= 0) { + const step = state.steps[pendingIndex]; + if (step !== undefined) { + (state.steps as Array)[pendingIndex] = { + ...step, + status: params.error !== undefined ? "error" : "success", + result: sanitizeTraceValue(params.result), + error: params.error !== undefined ? truncate(params.error, 160) : undefined, + durationMs: params.durationMs, + finishedAt: now, + }; + state.updatedAt = now; + return; + } + } + + // No matching pending step — record as already-completed + state.steps.push({ + id: `${state.nextSeq}`, + seq: state.nextSeq, + toolName: params.toolName, + toolUseId: params.toolUseId, + input: undefined, + result: sanitizeTraceValue(params.result), + error: params.error !== undefined ? truncate(params.error, 160) : undefined, + durationMs: params.durationMs, + status: params.error !== undefined ? "error" : "success", + startedAt: now, + finishedAt: now, + }); + state.nextSeq += 1; + state.updatedAt = now; +} + +export function getToolUseTraceSteps(runId: string): ToolUseTraceStep[] { + const state = runs.get(runId); + if (state === undefined) return []; + if (Date.now() - state.updatedAt > RUN_TTL_MS) { + runs.delete(runId); + return []; + } + const now = Date.now(); + return state.steps.map((step) => { + if (step.status === "running" && now - step.startedAt > STEP_RUNNING_TIMEOUT_MS) { + return { ...step, status: "error" as const, error: "timed out", finishedAt: now }; + } + return { ...step }; + }); +} + +// --------------------------------------------------------------------------- +// Internal helpers +// --------------------------------------------------------------------------- + +function findPendingStepIndex( + steps: ToolUseTraceStep[], + toolUseId: string | undefined, + toolName: string | undefined, +): number { + if (toolUseId !== undefined) { + for (let i = steps.length - 1; i >= 0; i -= 1) { + const step = steps[i]; + if (step !== undefined && step.status === "running" && step.toolUseId === toolUseId) { + return i; + } + } + } + if (toolName !== undefined) { + for (let i = steps.length - 1; i >= 0; i -= 1) { + const step = steps[i]; + if (step !== undefined && step.status === "running" && step.toolName === toolName) { + return i; + } + } + } + return -1; +} + +function pruneStaleRuns(): void { + const now = Date.now(); + for (const [runId, state] of runs) { + if (now - state.updatedAt > RUN_TTL_MS) { + runs.delete(runId); + } + } +} + +function truncate(value: string, maxLength: number): string { + return value.length <= maxLength ? value : `${value.slice(0, maxLength - 3)}...`; +} + +/** + * Sanitize a tool input or result value for display. + * Truncates strings, redacts sensitive keys, limits depth. + */ +function sanitizeTraceValue( + value: unknown, + depth = 0, + context: { source: "input" | "result" | undefined; key: string | undefined } = { source: undefined, key: undefined }, +): unknown { + if (value === null || value === undefined) return undefined; + + if (typeof value === "string") { + const limit = context.source === "result" ? RESULT_LIMIT : GENERIC_LIMIT; + return truncate(value, limit); + } + + if (typeof value === "number" || typeof value === "boolean") return value; + if (depth >= 2) return "[truncated]"; + + if (Array.isArray(value)) { + return value.slice(0, 8).map((item) => sanitizeTraceValue(item, depth + 1, { source: context.source, key: undefined })); + } + + if (typeof value === "object") { + const input = value as Record; + const output: Record = {}; + for (const [key, entryValue] of Object.entries(input).slice(0, 12)) { + output[key] = isSensitiveKey(key) + ? "[redacted]" + : sanitizeTraceValue(entryValue, depth + 1, { source: context.source, key }); + } + return output; + } + + return truncate(String(value), 180); +} + +const SENSITIVE_KEY_RE = + /token|secret|password|api[_-]?key|authorization|cookie|credential|bearer|session[_-]?id|client[_-]?secret|access[_-]?key/i; + +function isSensitiveKey(key: string): boolean { + return SENSITIVE_KEY_RE.test(key); +} diff --git a/hub/src/feishu/client.ts b/hub/src/feishu/client.ts index 86240d1..f285778 100644 --- a/hub/src/feishu/client.ts +++ b/hub/src/feishu/client.ts @@ -277,6 +277,38 @@ export async function patchTextMessage(rt: FeishuRuntime, messageId: string, tex } } +/** Send a raw interactive card (arbitrary JSON). Returns message_id on success. */ +export async function sendCard(rt: FeishuRuntime, chatId: string, card: Record): Promise { + const client = rt.client as unknown as { + im: { v1: { message: { create: (p: unknown) => Promise } } }; + }; + try { + const res = await client.im.v1.message.create({ + params: { receive_id_type: "chat_id" }, + data: { receive_id: chatId, msg_type: "interactive", content: JSON.stringify(card) }, + }); + return messageIdFromResponse(res); + } catch { + return null; + } +} + +/** Patch an interactive card in-place with raw JSON. */ +export async function patchCard(rt: FeishuRuntime, messageId: string, card: Record): Promise { + const client = rt.client as unknown as { + im: { v1: { message: { patch: (p: unknown) => Promise } } }; + }; + try { + await client.im.v1.message.patch({ + path: { message_id: messageId }, + params: { msg_type: "interactive" }, + data: { content: JSON.stringify(card) }, + }); + } catch (e) { + rt.logger.warn({ messageId, err: e instanceof Error ? e.message : String(e) }, "patchCard failed"); + } +} + /** Send an approval card with buttons. Returns message_id on success. */ export async function sendApprovalCard( rt: FeishuRuntime, diff --git a/hub/src/feishu/trigger.ts b/hub/src/feishu/trigger.ts index f7b2cec..558becb 100644 --- a/hub/src/feishu/trigger.ts +++ b/hub/src/feishu/trigger.ts @@ -1,9 +1,10 @@ /** - * Feishu @bot trigger → AgentRun lifecycle (streaming). - * - * Sends a "processing" card immediately, patches it with text deltas + tool - * status as the agent streams, then replaces it with a final status card on - * completion. Throttled to ~1 patch/sec to avoid spamming the Feishu API. + * Sends a "processing" reaction immediately, then streams a single + * interactive card through the full agent run lifecycle: thinking → tool + * calls (with trace panel) → streaming answer text → final card. The card + * shows a collapsible tool-use panel, a collapsible reasoning panel, and + * the markdown answer text. Throttled to ~2.5 patches/sec to avoid + * spamming the Feishu API. * * ADR-0001..0003, 0017: chat→project binding, permission gate, lock, * provider-bound session, SDK resume continuity, ADR-0003-authorized reads. @@ -14,8 +15,6 @@ import type { FastifyBaseLogger } from "fastify"; import { sendText, sendTextMessage, - sendInteractiveCardMessage, - patchTextMessage, reactToMessage, addReaction, removeReaction, @@ -32,7 +31,7 @@ import type { RuntimeSettings } from "../settings/runtime.js"; import { acquireLock, currentLockRunId, releaseLock } from "../lock.js"; import { createPermissionAuthorizer, type AuthorizationDecision, type PermissionAuthorizer } from "../permission.js"; import { writeAudit } from "../audit.js"; -import { PatchableTextStream } from "./textStream.js"; +import { StreamingAgentCard } from "./card/streaming-card.js"; import { createFileDeliveryMcpServer } from "./fileDeliveryTool.js"; import { readFeishuContext } from "./read.js"; import { MessageBatcher, messageBatchKey, type MessageBatcherOptions } from "./messageBatcher.js"; @@ -265,12 +264,10 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { return removeReaction(rt, msg.message_id, processingReactionId); }; - // Stream agent response as text-as-card messages. Lazy-init: don't send - // anything until the first text-delta arrives (avoids empty placeholder). - const textStream = new PatchableTextStream({ - create: (text) => sendInteractiveCardMessage(rt, chatId, text), - patch: (messageId, text) => patchTextMessage(rt, messageId, text), - }); + // Streaming agent card: single interactive card through the full run + // lifecycle (thinking → tool calls → streaming text → complete). + // Shows tool-use trace panel + reasoning panel + answer text. + const card = new StreamingAgentCard({ runId: run.id, rt, chatId, patchIntervalMs: undefined, maxMessageLength: undefined }); const deliveredFiles: string[] = []; const fileDeliveryMcpServer = createFileDeliveryMcpServer({ rt, @@ -297,21 +294,49 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { sessionId: session.id, prisma: deps.prisma, onStream: (event) => { - if (event.type === "text-delta") { - textStream.append(event.text); + switch (event.type) { + case "text-delta": + card.appendText(event.text); + break; + case "thinking-delta": + card.appendReasoning(event.text); + break; + case "tool-start": + card.onToolStart(event.toolName, event.toolUseId); + break; + case "tool-end": + card.onToolEnd({ + toolName: event.toolName, + toolUseId: event.toolUseId, + input: event.input, + result: undefined, + error: undefined, + durationMs: event.durationMs, + }); + break; + case "tool-result": + card.onToolEnd({ + toolName: event.toolName, + toolUseId: event.toolUseId, + input: undefined, + result: event.result, + error: event.isError ? event.result : undefined, + durationMs: event.durationMs, + }); + break; + case "finish": + break; } }, }) .then(async (result) => { - // Final flush: patch the last message with complete text, or send - // the full response if no streaming message was created yet. const finalText = result.text !== "" ? result.text : result.status === "failed" && result.error !== undefined - ? `处理失败: ${result.error}` + ? `\u5904\u7406\u5931\u8D25: ${result.error}` : result.text; - await textStream.finish(finalText); + await card.finish(finalText); const metadataPatch = sessionMetadataPatch(result.sdkSessionId); if (metadataPatch !== null) { await deps.prisma.agentSession.update({ @@ -342,6 +367,7 @@ export function makeTriggerHandler(deps: TriggerDeps): TriggerHandler { if (removedProcessingReaction) { await addReaction(rt, msg.message_id, "CrossMark"); } + await card.fail(e instanceof Error ? e.message : String(e)); try { await deps.prisma.agentRun.update({ where: { id: run.id }, diff --git a/hub/test/integration/helpers.ts b/hub/test/integration/helpers.ts index 7079d10..73df524 100644 --- a/hub/test/integration/helpers.ts +++ b/hub/test/integration/helpers.ts @@ -167,10 +167,19 @@ function textFromCard(card: unknown): string | null { const parts: string[] = []; for (const element of elements) { if (typeof element !== "object" || element === null) continue; + // New format: { tag: "markdown", content: string } + const tag = (element as { tag?: unknown }).tag; + const directContent = (element as { content?: unknown }).content; + if (tag === "markdown" && typeof directContent === "string") { + parts.push(directContent); + continue; + } + // Old format: { text: { content: string } } const text = (element as { text?: unknown }).text; - if (typeof text !== "object" || text === null) continue; - const content = (text as { content?: unknown }).content; - if (typeof content === "string") parts.push(content); + if (typeof text === "object" && text !== null) { + const content = (text as { content?: unknown }).content; + if (typeof content === "string") parts.push(content); + } } return parts.length === 0 ? null : parts.join("\n"); }