diff --git a/.gitignore b/.gitignore index eae078b..7d78ac0 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ dist/ !scripts/*.ts scripts/.env !scripts/.env.example +scripts/assets/ diff --git a/CLAUDE.md b/CLAUDE.md index 45d3514..1f424a8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,6 +13,7 @@ npx tsx scripts/generate-bot-token.ts npx tsx scripts/discover-channels.ts npx tsx scripts/chat-client.ts npx tsx scripts/test-roundtrip.ts +npx tsx scripts/test-roundtrip-audio.ts npx tsx scripts/test-thread.ts # Restart the gateway after code changes @@ -45,9 +46,9 @@ The OpenClaw framework discovers the plugin via `plugins.load.paths` in `~/.open | File | Responsibility | |---|---| -| `src/channel.ts` | Main plugin export (`streamchatPlugin`). Contains `handleStreamChatMessage` (inbound dispatch) and the `ChannelPlugin` adapter implementations: `config`, `outbound`, `gateway`, `status`. | +| `src/channel.ts` | Main plugin export (`streamchatPlugin`). Contains `handleStreamChatMessage` (inbound dispatch), `resolveStreamChatMedia` (inbound media download + WAV→OGG conversion), `uploadOutboundMedia` (outbound media upload), and the `ChannelPlugin` adapter implementations: `config`, `outbound`, `gateway`, `status`. | | `src/stream-chat-runtime.ts` | `StreamChatClientRuntime` — wraps the `stream-chat` SDK. Connects as bot user (`allowServerSideConnect: true` is required for Node.js server contexts), queries + watches channels on startup, auto-watches channels added later via `notification.added_to_channel`. | -| `src/streaming.ts` | `StreamingHandler` — manages the AI streaming lifecycle per run: creates placeholder message → sends `ai_indicator` events → calls `partialUpdateMessage` on throttled chunks → finalizes on completion. | +| `src/streaming.ts` | `StreamingHandler` — manages the AI streaming lifecycle per run: creates placeholder message → sends `ai_indicator` events → calls `partialUpdateMessage` on throttled chunks → finalizes on completion with text and any queued attachments. | | `src/run-context.ts` | `RunContextMap` — binds an OpenClaw `runId` (UUID generated per inbound message) to delivery routing state: `channelId`, `threadParentId`, `responseMessageId`. TTL of 5 min. | | `src/envelope.ts` | `buildEnvelope` — wraps the raw message text in `[Thread]` / `[Replying to]` XML-like tags so the LLM receives thread and quote context in the single-session model. | | `src/types.ts` | `StreamChatChannelConfig`, `ResolvedAccount`, `RunContext`, `EnvelopeResult` interfaces, plus config helper functions (`getStreamChatConfig`, `resolveStreamChatAccount`). | @@ -72,11 +73,12 @@ message.new (WebSocket) replyOptions.onPartialReply fires per streaming token (cumulative text): delta = full.slice(lastPartialText.length) → onTextChunk (throttled partialUpdateMessage) deliver(payload, info) called once per complete block: - info.kind === "tool" → onRunProgress (EXTERNAL_SOURCES indicator) - payload.isError → onRunError (error text + ERROR indicator) - text block → no-op (already handled token-by-token above) + info.kind === "tool" → onRunProgress (EXTERNAL_SOURCES indicator) + payload.isError → onRunError (error text + ERROR indicator) + payload.mediaUrl/mediaUrls → uploadOutboundMedia → streamingHandler.addAttachment + text block → no-op (already handled token-by-token above) after dispatcher returns: - → onRunCompleted (final partialUpdateMessage + ai_indicator.clear) + → onRunCompleted (final partialUpdateMessage with text + attachments + ai_indicator.clear) ``` **Why pre-create the placeholder:** `onPartialReply` is called fire-and-forget (`void`) by OpenClaw, so it cannot safely do async work (like `channel.sendMessage`). The placeholder must exist before the first token arrives. @@ -95,7 +97,8 @@ How each signal from the OpenClaw pipeline translates into Stream Chat API calls | `onPartialReply` first token | `channel.sendEvent({ type: "ai_indicator.update", ai_state: "AI_STATE_GENERATING" })` | Transitions from THINKING on the very first token | | `onPartialReply` per token — throttled | `client.partialUpdateMessage(msgId, { set: { text, generating: true } })` | Delta-computed from cumulative text. Odd chunks 1,3,5,7; then every N (default 15). Chained via `lastUpdatePromise` to avoid out-of-order updates | | `deliver` with `info.kind === "tool"` | `channel.sendEvent({ type: "ai_indicator.update", ai_state: "AI_STATE_EXTERNAL_SOURCES" })` | Only emitted once per run (de-duplicated by `indicatorState`) | -| Dispatcher resolves (run complete) | `client.partialUpdateMessage(msgId, { set: { text, generating: false } })` | Final flush, waits for any in-flight partial updates first | +| `deliver` with `payload.mediaUrl` / `payload.mediaUrls` | `channel.sendFile(buffer, name, mime)` → queued as `Attachment` on the stream | Outbound audio (TTS); attachment type is `"audio"` or `"voiceRecording"` if `payload.audioAsVoice` is true | +| Dispatcher resolves (run complete) | `client.partialUpdateMessage(msgId, { set: { text, generating: false, attachments? } })` | Final flush; includes accumulated attachments if any were uploaded during the run | | Dispatcher resolves (run complete) | `channel.sendEvent({ type: "ai_indicator.clear" })` | Clears the indicator bubble | | Dispatcher resolves (run complete) | `channel.deleteReaction(inboundMsgId, "eyes")` → `channel.sendReaction(inboundMsgId, { type: "white_check_mark" })` | Reaction swap on the original user message | | `deliver` with `payload.isError` | `client.partialUpdateMessage(msgId, { set: { text: "…\n\nError: …", generating: false } })` | Appends error to any partial text already accumulated | @@ -111,7 +114,8 @@ Each agent run that produces text goes through these steps in `StreamingHandler` 1. `onRunStarted` — `channel.sendMessage({ text: "", ai_generated: true })` → `ai_indicator.update(AI_STATE_THINKING)` 2. `onTextChunk` — accumulates text, switches indicator to `AI_STATE_GENERATING` on first chunk, calls `client.partialUpdateMessage({ set: { text, generating: true } })` throttled (early burst: odd chunks < 8; then every Nth chunk, default N=15) -3. `onRunCompleted` — waits for in-flight partial updates, sends final `partialUpdateMessage({ generating: false })`, sends `ai_indicator.clear` +3. `addAttachment` — called from the `deliver` callback when the agent produces media (e.g. TTS audio); uploads to Stream Chat CDN and queues the `Attachment` object on the stream state +4. `onRunCompleted` — waits for in-flight partial updates, sends final `partialUpdateMessage({ text, generating: false, attachments? })`, sends `ai_indicator.clear` Force-stop (`ai_indicator.stop` from client) calls `onForceStop`, which clears `generating` without overwriting the accumulated text. @@ -147,10 +151,12 @@ Config supports a flat default account or named sub-accounts: ## Key design decisions - **Bot token in config, secret is not.** The API secret is only used in `scripts/generate-bot-token.ts` to mint a JWT. Only the resulting token is stored in `openclaw.json`. -- **`deliver` callback vs. completion signal.** `dispatchReplyWithBufferedBlockDispatcher` signals completion by resolving its promise, not by passing an `isComplete` flag. The `info.kind` parameter (`"tool" | "block" | "final"`) distinguishes delivery type. `onRunCompleted` is called after the dispatcher awaits. The `ReplyPayload` type has `text` and `isError` as the only relevant fields — there is no `markdown`, `isComplete`, or `toolName` field, despite what seems intuitive. +- **`deliver` callback vs. completion signal.** `dispatchReplyWithBufferedBlockDispatcher` signals completion by resolving its promise, not by passing an `isComplete` flag. The `info.kind` parameter (`"tool" | "block" | "final"`) distinguishes delivery type. `onRunCompleted` is called after the dispatcher awaits. The `ReplyPayload` type fields relevant to this plugin: `text`, `isError`, `mediaUrl`, `mediaUrls`, `audioAsVoice`. There is no `markdown`, `isComplete`, or `toolName` field, despite what seems intuitive. - **Partial updates are chained via `lastUpdatePromise`.** Each `partialUpdateMessage` is `.then()`-chained onto the previous one to avoid out-of-order message text. - **`safeSendEvent` swallows errors.** Indicator events are best-effort; a failed `ai_indicator` update must not abort message delivery. Retries: 5 attempts, exponential backoff starting at 100 ms, only on 429/5xx. - **`seenThreads` is process-scoped.** The `Set` tracking "first message in thread" lives at module level, so it persists across gateway reloads until the process restarts. This is intentional — it avoids re-sending parent context for active threads after a config reload. - **`onTextChunk` receives deltas despite the wire protocol using full text.** `onPartialReply` provides cumulative text; `channel.ts` extracts the delta before calling `onTextChunk`. Inside `StreamingHandler`, `onTextChunk` re-accumulates deltas into `accumulatedText` and passes that full string to `partialUpdateMessage`. The round-trip is: cumulative → delta → cumulative. The delta extraction exists because `StreamingHandler` was designed around the "streaming chunks" mental model — it owns the accumulation and the throttle counter, making that API feel natural. The redundancy is intentional for architectural clarity, not a bug. - **`activeGatewayCleanup` is a module-level registry for defence against connection accumulation.** `startAccount` stores a cleanup function per `accountId`. On the next `startAccount` call for the same account, any existing entry is force-invoked before creating the new connection. This self-heals the case where OpenClaw's in-process gateway reload calls `startAccount` without calling `stop()` first, which would otherwise leave orphaned WebSocket connections accumulating (each receiving every `message.new` event). `handleAbort` is idempotent via a `stopped` boolean guard, so it is safe to call from both the abort signal and `stop()`. - **All three SDK event listeners are removed on cleanup.** `handleAbort` explicitly calls `client.off()` for `message.new` and `ai_indicator.stop`; `chatRuntime.stop()` removes `notification.added_to_channel` (saved as `addedToChannelHandler` in `StreamChatClientRuntime`). Failing to remove any one of these would cause listener accumulation across restarts. +- **Inbound WAV is converted to OGG before saving.** OpenClaw's `EXT_BY_MIME` table does not map `audio/wav` → `.wav`, so WAV files would be saved without an extension and never transcribed. `resolveStreamChatMedia` detects WAV MIME types (`audio/wav`, `audio/wave`, `audio/x-wav`) and converts to OGG Opus via `ffmpeg` before calling `saveMediaBuffer`. Fallback: if ffmpeg is unavailable the original WAV buffer is passed as-is. +- **Outbound audio is uploaded to Stream Chat CDN.** When the agent produces an audio file (`payload.mediaUrl`/`payload.mediaUrls` in the `deliver` callback), `uploadOutboundMedia` reads the local file, determines its MIME type via `mime-types`, and uploads it via `channel.sendFile()`. The returned CDN URL is wrapped in a Stream Chat `Attachment` object (`type: "audio"` or `"voiceRecording"` when `audioAsVoice` is true) and queued on the stream state via `addAttachment`. The attachment is included in the final `partialUpdateMessage` call by `onRunCompleted`. diff --git a/scripts/test-roundtrip-audio.ts b/scripts/test-roundtrip-audio.ts new file mode 100644 index 0000000..43c4423 --- /dev/null +++ b/scripts/test-roundtrip-audio.ts @@ -0,0 +1,121 @@ +#!/usr/bin/env npx tsx +/** + * Automated audio round-trip test: uploads a WAV file, sends it as a voice + * recording, and observes the bot's streaming response. + * + * Requires a .env file at the project root (see .env.example). + */ + +import { config } from "dotenv"; +config({ path: new URL(".env", import.meta.url).pathname }); +import { StreamChat } from "stream-chat"; +import { readFileSync } from "fs"; + +const apiKey = process.env.STREAM_API_KEY; +const userId = process.env.TEST_USER_ID; +const userToken = process.env.TEST_USER_TOKEN; +const channelId = process.env.TEST_CHANNEL_ID || "ai-test-channel"; + +if (!apiKey || !userId || !userToken) { + console.error("Error: STREAM_API_KEY, TEST_USER_ID, and TEST_USER_TOKEN must be set in .env"); + process.exit(1); +} + +const client = new StreamChat(apiKey, { allowServerSideConnect: true }); +console.log(`Connecting as ${userId}...`); +await client.connectUser({ id: userId }, userToken); +console.log("Connected."); + +// Watch the test channel +const channel = client.channel("messaging", channelId); +await channel.watch(); +console.log(`Using channel: messaging:${channelId}`); + +let gotResponse = false; + +// Listen for new messages +client.on("message.new", (event) => { + if (event.user?.id === userId) return; + if (!event.message) return; + const from = event.user?.name || event.user?.id || "unknown"; + const text = event.message.text || "(no text)"; + const aiGenerated = event.message.ai_generated ? " [AI]" : ""; + const attachments = event.message.attachments || []; + const hasAudio = attachments.some( + (a) => a.type === "voiceRecording" || a.mime_type?.startsWith("audio/"), + ); + console.log(`[NEW MSG][${from}]${aiGenerated}: ${text}${hasAudio ? " [has audio]" : ""}`); +}); + +// Listen for AI indicators +for (const evType of ["ai_indicator.update", "ai_indicator.clear"] as const) { + client.on(evType as Parameters[0], (event) => { + const raw = event as Record; + if (String(raw.type).includes("clear")) { + console.log(`[AI INDICATOR] cleared`); + } else { + console.log(`[AI INDICATOR] ${raw.ai_state}`); + } + }); +} + +// Listen for message updates (streaming) +client.on("message.updated", (event) => { + if (!event.message) return; + const text = event.message.text || ""; + const generating = (event.message as Record).generating; + const attachments = event.message.attachments || []; + const hasAudio = attachments.some( + (a) => a.type === "voiceRecording" || a.mime_type?.startsWith("audio/"), + ); + if (generating) { + console.log(`[STREAMING] ${text.slice(-120)}${hasAudio ? " [has audio]" : ""}`); + } else { + console.log(`[FINAL] ${text}${hasAudio ? " [has audio]" : ""}`); + gotResponse = true; + } +}); + +// Read audio file +const audioPath = new URL("assets/test-audio.wav", import.meta.url).pathname; +const wavBuffer = readFileSync(audioPath); +console.log(`\nLoaded test audio: ${wavBuffer.length} bytes`); + +// Upload WAV via channel.sendFile +console.log("Uploading audio file..."); +const uploadResp = await channel.sendFile(wavBuffer as any, "test-audio.wav", "audio/wav"); +const assetUrl = uploadResp.file; +console.log(`Uploaded → ${assetUrl.slice(0, 80)}...`); + +// Send message with voiceRecording attachment +console.log("Sending audio message..."); +await channel.sendMessage({ + text: "", + attachments: [ + { + type: "voiceRecording", + asset_url: assetUrl, + mime_type: "audio/wav", + title: "test-audio.wav", + file_size: wavBuffer.length, + duration: 5.071, + waveform_data: new Array(100).fill(0.2), + }, + ], +}); +console.log("Audio message sent. Waiting for response...\n"); + +// Wait up to 60 seconds for response (audio processing is slower) +const start = Date.now(); +while (!gotResponse && Date.now() - start < 60000) { + await new Promise((r) => setTimeout(r, 500)); +} + +if (gotResponse) { + console.log("\n✓ Audio round-trip test PASSED — got bot response."); +} else { + console.log("\n✗ Audio round-trip test TIMED OUT — no response in 60s."); +} + +await client.disconnectUser(); +process.exit(gotResponse ? 0 : 1); diff --git a/src/channel.ts b/src/channel.ts index 549226a..57cbe44 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -1,12 +1,19 @@ import { randomUUID } from "node:crypto"; -import type { Event } from "stream-chat"; +import { execFile } from "node:child_process"; +import { tmpdir } from "node:os"; +import { basename, extname, join } from "node:path"; +import { writeFile, readFile, unlink, stat } from "node:fs/promises"; +import { promisify } from "node:util"; +import { lookup as mimeLookup } from "mime-types"; +import type { Attachment, Channel, Event } from "stream-chat"; import type { ChannelGatewayContext, ChannelLogSink, ChannelOutboundContext, OpenClawConfig, + ReplyPayload, } from "openclaw/plugin-sdk"; -import { buildChannelConfigSchema } from "openclaw/plugin-sdk"; +import { buildAgentMediaPayload, buildChannelConfigSchema } from "openclaw/plugin-sdk"; import { StreamChatConfigSchema } from "./config-schema.js"; import { getStreamChatRuntime } from "./runtime.js"; import { StreamChatClientRuntime } from "./stream-chat-runtime.js"; @@ -72,6 +79,200 @@ async function removeReaction( } } +// --------------------------------------------------------------------------- +// Media helpers +// --------------------------------------------------------------------------- + +const MAX_MEDIA_FILES = 8; +const DEFAULT_MEDIA_MAX_BYTES = 25 * 1024 * 1024; // 25 MB + +/** Attachments that carry downloadable media files. */ +function isMediaAttachment(att: Attachment): boolean { + return Boolean(att.asset_url || att.image_url); +} + +/** Resolves the download URL for a Stream Chat attachment. */ +function resolveAttachmentUrl(att: Attachment): string | undefined { + return att.asset_url || att.image_url || undefined; +} + +/** Builds a human-readable label for an attachment (used in the envelope). */ +function attachmentLabel(att: Attachment): string { + const name = att.title || att.type || "file"; + return `[Attachment: ${name}]`; +} + +type ResolvedMedia = { + path: string; + contentType?: string; + label: string; +}; + +const execFileAsync = promisify(execFile); + +/** MIME types that need conversion to OGG before passing to OpenClaw. */ +const WAV_MIMES = new Set(["audio/wav", "audio/wave", "audio/x-wav"]); + +/** + * Converts a WAV buffer to OGG (Opus) via ffmpeg. + * Returns the converted buffer, or the original buffer if ffmpeg is unavailable + * or conversion fails. + */ +async function convertWavToOgg( + buffer: Buffer, + log?: ChannelLogSink, +): Promise<{ buffer: Buffer; contentType: string; ext: string }> { + const id = randomUUID(); + const wavPath = join(tmpdir(), `${id}.wav`); + const oggPath = join(tmpdir(), `${id}.ogg`); + try { + await writeFile(wavPath, buffer); + await execFileAsync("ffmpeg", [ + "-i", wavPath, + "-c:a", "libopus", + "-y", oggPath, + ], { timeout: 15_000 }); + const oggBuffer = await readFile(oggPath); + return { buffer: oggBuffer, contentType: "audio/ogg", ext: ".ogg" }; + } catch (err) { + log?.warn?.(`[StreamChat] WAV→OGG conversion failed, passing WAV as-is: ${String(err)}`); + return { buffer, contentType: "audio/wav", ext: ".wav" }; + } finally { + await unlink(wavPath).catch(() => {}); + await unlink(oggPath).catch(() => {}); + } +} + +/** + * Downloads Stream Chat attachments and saves them locally via the OpenClaw + * media pipeline. Returns an array of resolved media entries (empty if none + * could be downloaded). + */ +async function resolveStreamChatMedia( + attachments: Attachment[], + maxBytes: number, + log?: ChannelLogSink, +): Promise { + const rt = getStreamChatRuntime(); + const candidates = attachments.filter(isMediaAttachment).slice(0, MAX_MEDIA_FILES); + const results: ResolvedMedia[] = []; + + for (const att of candidates) { + const url = resolveAttachmentUrl(att); + if (!url) continue; + + try { + const fetched = await rt.channel.media.fetchRemoteMedia({ + url, + filePathHint: att.title ?? undefined, + maxBytes, + }); + if (fetched.buffer.byteLength > maxBytes) continue; + + let mediaBuffer = fetched.buffer; + let contentType = att.mime_type ?? fetched.contentType; + let fileName = att.title ?? fetched.fileName; + + // Convert WAV to OGG — OpenClaw's mime-to-extension map lacks audio/wav, + // so WAV files get saved without an extension and aren't transcribed. + if (contentType && WAV_MIMES.has(contentType.split(";")[0]?.trim().toLowerCase())) { + const converted = await convertWavToOgg(mediaBuffer, log); + mediaBuffer = converted.buffer; + contentType = converted.contentType; + if (fileName) { + fileName = fileName.replace(/\.wav$/i, converted.ext); + } + } + + if (mediaBuffer.byteLength > maxBytes) continue; + + const saved = await rt.channel.media.saveMediaBuffer( + mediaBuffer, + contentType, + "inbound", + maxBytes, + fileName, + ); + + results.push({ + path: saved.path, + contentType: contentType ?? saved.contentType, + label: attachmentLabel(att), + }); + } catch (err) { + log?.warn?.( + `[StreamChat] Failed to download attachment "${att.title ?? url}": ${String(err)}`, + ); + } + } + + return results; +} + +// --------------------------------------------------------------------------- +// Outbound media helpers +// --------------------------------------------------------------------------- + +const AUDIO_MIMES = new Set([ + "audio/mpeg", "audio/mp3", "audio/mp4", "audio/ogg", "audio/wav", + "audio/wave", "audio/x-wav", "audio/x-m4a", "audio/aac", "audio/flac", + "audio/webm", +]); + +function isAudioMime(mime?: string): boolean { + if (!mime) return false; + return AUDIO_MIMES.has(mime.split(";")[0]?.trim().toLowerCase()); +} + +/** + * Uploads a local media file to Stream Chat and returns a Stream Chat + * attachment object ready to be added to a message. + */ +async function uploadOutboundMedia( + ch: Channel, + localPath: string, + audioAsVoice: boolean, + log?: ChannelLogSink, +): Promise { + try { + const fileStat = await stat(localPath); + if (!fileStat.isFile()) return null; + + const fileName = basename(localPath); + const ext = extname(localPath).toLowerCase(); + const mimeType = mimeLookup(ext) || "application/octet-stream"; + const buffer = await readFile(localPath); + + const uploadResp = await ch.sendFile( + buffer as unknown as Parameters[0], + fileName, + mimeType, + ); + const assetUrl = uploadResp.file; + + const isAudio = isAudioMime(mimeType); + const attachmentType = isAudio && audioAsVoice ? "voiceRecording" : isAudio ? "audio" : "file"; + + const att: Attachment = { + type: attachmentType, + asset_url: assetUrl, + mime_type: mimeType, + title: fileName, + file_size: fileStat.size, + }; + + if (attachmentType === "voiceRecording") { + att.waveform_data = new Array(100).fill(0.5); + } + + log?.info?.(`[StreamChat] Uploaded outbound media: ${fileName} → ${attachmentType}`); + return att; + } catch (err) { + log?.warn?.(`[StreamChat] Failed to upload outbound media "${localPath}": ${String(err)}`); + return null; + } +} + // --------------------------------------------------------------------------- // Inbound message handler // --------------------------------------------------------------------------- @@ -107,8 +308,12 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise m.label).join("\n"); + const envelope = buildEnvelope({ - text, + text: effectiveText, senderId, senderName, messageId, @@ -195,13 +409,23 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise 0 + ? buildAgentMediaPayload( + media.map((m) => ({ + path: m.path, + contentType: m.contentType ?? null, + })), + ) + : {}; + // Finalize inbound context const to = channelId; const fromLabel = `${senderName} (${senderId})`; const ctx = rt.channel.reply.finalizeInboundContext({ Body: envelope.body, - RawBody: text, + RawBody: text || effectiveText, CommandBody: envelope.commandBody, From: to, To: to, @@ -219,6 +443,7 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise 0 ? ` media=${media.length}` : ""}`, ); // Send ack reaction @@ -305,7 +530,7 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise { try { @@ -325,6 +550,23 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise; finalized: boolean; + attachments: Attachment[]; } export class StreamingHandler { @@ -115,6 +116,7 @@ export class StreamingHandler { indicatorState: "AI_STATE_THINKING", lastUpdatePromise: Promise.resolve(), finalized: false, + attachments: [], }); return messageId; @@ -198,6 +200,16 @@ export class StreamingHandler { } } + /** + * Adds an attachment to the stream's pending list. Attachments are included + * in the final partialUpdateMessage when the run completes. + */ + addAttachment(runId: string, attachment: Attachment): void { + const stream = this.streams.get(runId); + if (!stream || stream.finalized) return; + stream.attachments.push(attachment); + } + /** * Called when the agent run completes successfully. * Finalizes the message text and clears the indicator. @@ -212,13 +224,17 @@ export class StreamingHandler { // Wait for any in-flight partial updates await stream.lastUpdatePromise.catch(() => {}); - // Final update with complete text + // Final update with complete text and any accumulated attachments try { + const setFields: Record = { + text: stream.accumulatedText || "(No response)", + generating: false, + }; + if (stream.attachments.length > 0) { + setFields.attachments = stream.attachments; + } await client.partialUpdateMessage(stream.messageId, { - set: { - text: stream.accumulatedText || "(No response)", - generating: false, - }, + set: setFields, }); } catch (err) { log?.error?.(