diff --git a/CLAUDE.md b/CLAUDE.md index 2d7410d..4ec9638 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -97,7 +97,7 @@ How each signal from the OpenClaw pipeline translates into Stream Chat API calls | Pre-dispatch (before agent runs) | `channel.sendMessage({ text: "", ai_generated: true })` | Creates the bot's placeholder message | | Pre-dispatch (before agent runs) | `channel.sendEvent({ type: "ai_indicator.update", ai_state: "AI_STATE_THINKING" })` | Sent immediately with placeholder | | `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 | +| `onPartialReply` per token — throttled | `client.partialUpdateMessage(msgId, { set: { text, generating: true } })` | Delta-computed from cumulative text. Chunks 1 and 5, then every N (default 35). 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 | | Dispatcher resolves (run complete) | `channel.sendEvent({ type: "ai_indicator.clear" })` | Clears the indicator bubble | @@ -114,7 +114,7 @@ The `ai_indicator` events are sent via `safeSendEvent`, which retries up to 5 ti 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) +2. `onTextChunk` — accumulates text, switches indicator to `AI_STATE_GENERATING` on first chunk, calls `client.partialUpdateMessage({ set: { text, generating: true } })` throttled (early burst on chunks 1 and 5; then every Nth chunk, default N=35) 3. `onRunCompleted` — waits for in-flight partial updates, sends final `partialUpdateMessage({ generating: false })`, sends `ai_indicator.clear` Force-stop (`ai_indicator.stop` from client) calls `onForceStop`, which clears `generating` without overwriting the accumulated text. diff --git a/README.md b/README.md index 7486ef0..9e8ad1e 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ If you used Option B, add the channel config and plugin entry to `~/.openclaw/op // Optional: "ackReaction": "eyes", // reaction added when message is received (default: "eyes") "doneReaction": "white_check_mark", // reaction swapped in when response is done (default: "white_check_mark") - "streamingThrottle": 15, // partial-update every Nth chunk (default: 15) + "streamingThrottle": 35, // partial-update every Nth chunk after chunks 1 and 5 (default: 35) "mockResponse": "hello" // if set, reply with this string and skip agent dispatch (for testing) } }, diff --git a/scripts/test-channel-streaming-delivery.ts b/scripts/test-channel-streaming-delivery.ts new file mode 100644 index 0000000..51746a0 --- /dev/null +++ b/scripts/test-channel-streaming-delivery.ts @@ -0,0 +1,256 @@ +#!/usr/bin/env npx tsx +import assert from "node:assert/strict"; +import { handleStreamChatMessage } from "../src/channel.js"; +import { setStreamChatRuntime } from "../src/runtime.js"; +import { RunContextMap } from "../src/run-context.js"; +import { StreamingHandler } from "../src/streaming.js"; +import type { ResolvedAccount } from "../src/types.js"; + +const responseChannel = { + sendMessage: async () => ({ message: { id: "response-1" } }), + sendEvent: async () => undefined, +}; + +const chatRuntime = { + getOrQueryChannel: async () => responseChannel, +}; + +const cfg = { + channels: { + streamchat: { + apiKey: "key", + botUserId: "bot-1", + botUserToken: "token", + }, + }, +}; + +const account: ResolvedAccount = { + accountId: "default", + enabled: true, + configured: true, + apiKey: "key", + botUserId: "bot-1", + botUserToken: "token", + dmPolicy: "open", + ackReaction: "", + doneReaction: "", + streamingThrottle: 35, +}; + +async function dispatchTestMessage( + streamingHandler: Parameters[0]["streamingHandler"], + inboundId: string, +): Promise { + await handleStreamChatMessage({ + cfg: cfg as never, + accountId: "default", + account, + event: { + user: { id: "user-1", name: "OpenClaw User" }, + channel_type: "messaging", + channel_id: "user-1-main", + channel: { cid: "messaging:user-1-main" }, + message: { + id: inboundId, + text: "stream a long answer", + created_at: new Date().toISOString(), + }, + } as never, + chatRuntime: chatRuntime as never, + streamingHandler, + runContexts: new RunContextMap(), + }); +} + +async function testCumulativePartialDelivery(): Promise { + const deliveredChunks: string[] = []; + let completed = false; + let capturedReplyOptions: Record | undefined; + + const streamingHandler = { + onRunStarted: async () => "response-1", + onTextChunk: async (_runId: string, chunk: string) => { + deliveredChunks.push(chunk); + }, + onRunCompleted: async () => { + completed = true; + }, + onRunProgress: async () => undefined, + onRunError: async () => undefined, + }; + + setStreamChatRuntime({ + channel: { + routing: { + resolveAgentRoute: () => ({ + agentId: "default", + sessionKey: "agent:default:streamchat:channel:messaging:user-1-main", + mainSessionKey: "agent:default:streamchat:channel:messaging:user-1-main", + }), + }, + session: { + resolveStorePath: () => "/tmp/openclaw-session-store", + recordInboundSession: async () => undefined, + }, + reply: { + finalizeInboundContext: (ctx: Record) => ctx, + dispatchReplyWithBufferedBlockDispatcher: async (params: { + replyOptions?: Record; + dispatcherOptions: { + deliver: ( + payload: { text?: string; isError?: boolean }, + info: { kind: string }, + ) => Promise; + }; + }) => { + capturedReplyOptions = params.replyOptions; + const onPartialReply = params.replyOptions?.onPartialReply as + | ((payload: { text?: string }) => void) + | undefined; + + onPartialReply?.({ text: "first partial" }); + onPartialReply?.({ text: "first partial plus more" }); + await params.dispatcherOptions.deliver( + { text: "first partial plus more" }, + { kind: "block" }, + ); + await params.dispatcherOptions.deliver( + { text: "first partial plus more" }, + { kind: "final" }, + ); + + return { queuedFinal: true, counts: { tool: 0, block: 0, final: 1 } }; + }, + }, + }, + } as never); + + await dispatchTestMessage(streamingHandler as never, "inbound-1"); + + assert.equal( + capturedReplyOptions?.sourceReplyDeliveryMode, + "automatic", + "Stream Chat should use automatic source replies so text stays in the streaming placeholder", + ); + assert.deepEqual( + deliveredChunks, + ["first partial", " plus more"], + "final dispatcher text must not duplicate text already delivered by partial streaming", + ); + assert.equal(completed, true, "run should complete and finalize the placeholder"); +} + +async function testNonPrefixFinalReplacesStreamedText(): Promise { + const updates: Array<{ messageId: string; payload: unknown }> = []; + let responseId = 0; + const responseChannelWithEvents = { + sendMessage: async () => ({ message: { id: `replacement-response-${++responseId}` } }), + sendEvent: async () => undefined, + }; + const chatRuntimeWithReplacementChannel = { + getOrQueryChannel: async () => responseChannelWithEvents, + }; + const runContexts = new RunContextMap(); + const streamingHandler = new StreamingHandler({ + client: { + partialUpdateMessage: async (messageId: string, payload: unknown) => { + updates.push({ messageId, payload }); + }, + deleteMessage: async () => undefined, + } as never, + runContexts, + }); + + setStreamChatRuntime({ + channel: { + routing: { + resolveAgentRoute: () => ({ + agentId: "default", + sessionKey: "agent:default:streamchat:channel:messaging:user-1-main", + mainSessionKey: "agent:default:streamchat:channel:messaging:user-1-main", + }), + }, + session: { + resolveStorePath: () => "/tmp/openclaw-session-store", + recordInboundSession: async () => undefined, + }, + reply: { + finalizeInboundContext: (ctx: Record) => ctx, + dispatchReplyWithBufferedBlockDispatcher: async (params: { + replyOptions?: Record; + dispatcherOptions: { + deliver: ( + payload: { text?: string; isError?: boolean }, + info: { kind: string }, + ) => Promise; + }; + }) => { + const onPartialReply = params.replyOptions?.onPartialReply as + | ((payload: { text?: string }) => void) + | undefined; + + onPartialReply?.({ text: "draft partial" }); + await params.dispatcherOptions.deliver( + { text: "Authoritative final answer." }, + { kind: "final" }, + ); + + return { queuedFinal: true, counts: { tool: 0, block: 0, final: 1 } }; + }, + }, + }, + } as never); + + await handleStreamChatMessage({ + cfg: cfg as never, + accountId: "default", + account, + event: { + user: { id: "user-1", name: "OpenClaw User" }, + channel_type: "messaging", + channel_id: "user-1-main", + channel: { cid: "messaging:user-1-main" }, + message: { + id: "inbound-2", + text: "stream a normalized answer", + created_at: new Date().toISOString(), + }, + } as never, + chatRuntime: chatRuntimeWithReplacementChannel as never, + streamingHandler, + runContexts, + }); + + const finalUpdate = updates.findLast(({ payload }) => { + const set = (payload as { set?: { generating?: boolean } }).set; + return set?.generating === false; + }); + + assert.deepEqual( + finalUpdate, + { + messageId: "replacement-response-1", + payload: { + set: { + text: "Authoritative final answer.", + generating: false, + }, + }, + }, + "non-prefix final text should replace stale streamed partial text", + ); + assert.equal( + updates.some(({ payload }) => { + const set = (payload as { set?: { text?: string } }).set; + return set?.text === "draft partialAuthoritative final answer."; + }), + false, + "non-prefix final text must not be appended after the stale partial", + ); +} + +await testCumulativePartialDelivery(); +await testNonPrefixFinalReplacesStreamedText(); + +console.log("channel streaming delivery test passed"); diff --git a/scripts/test-streaming-final-retry.ts b/scripts/test-streaming-final-retry.ts new file mode 100644 index 0000000..1ffb305 --- /dev/null +++ b/scripts/test-streaming-final-retry.ts @@ -0,0 +1,166 @@ +#!/usr/bin/env npx tsx +import assert from "node:assert/strict"; +import { StreamingHandler } from "../src/streaming.js"; +import { RunContextMap } from "../src/run-context.js"; +import type { RunContext } from "../src/types.js"; + +function deferred(): { + promise: Promise; + resolve: (value: T | PromiseLike) => void; + reject: (reason?: unknown) => void; +} { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function rateLimitError(): Error & { status: number } { + const err = new Error("Too many requests") as Error & { status: number }; + err.status = 429; + return err; +} + +function runContext(runId: string): RunContext { + return { + runId, + channelType: "messaging", + channelId: "user-1-main", + threadParentId: null, + inboundMessageId: `inbound-${runId}`, + senderId: "user-1", + responseMessageId: null, + }; +} + +async function testFinalUpdateRetries(): Promise { + const events: Array> = []; + const updates: Array<{ messageId: string; payload: unknown }> = []; + + const channel = { + sendMessage: async () => ({ message: { id: "response-1" } }), + sendEvent: async (event: Record) => { + events.push(event); + }, + }; + + let finalAttempts = 0; + const client = { + partialUpdateMessage: async (messageId: string, payload: unknown) => { + updates.push({ messageId, payload }); + const set = (payload as { set?: { generating?: boolean } }).set; + if (set?.generating === false && finalAttempts < 2) { + finalAttempts++; + throw rateLimitError(); + } + }, + }; + + const runContexts = new RunContextMap(); + const runCtx = runContext("run-1"); + runContexts.set(runCtx.runId, runCtx); + + const logMessages: string[] = []; + const handler = new StreamingHandler({ + client: client as never, + runContexts, + log: { + warn: (message: string) => logMessages.push(message), + error: (message: string) => logMessages.push(message), + }, + }); + + await handler.onRunStarted(runCtx.runId, channel as never, runCtx); + await handler.onTextChunk(runCtx.runId, "alpha ", 1); + await handler.onTextChunk(runCtx.runId, "beta", 1); + await handler.onRunCompleted(runCtx.runId); + + const finalUpdates = updates.filter(({ payload }) => { + const set = (payload as { set?: { generating?: boolean } }).set; + return set?.generating === false; + }); + + assert.equal(finalUpdates.length, 3, "final update should retry retryable failures"); + assert.deepEqual(finalUpdates.at(-1), { + messageId: "response-1", + payload: { set: { text: "alpha beta", generating: false } }, + }); + assert.equal( + events.at(-1)?.type, + "ai_indicator.clear", + "indicator should clear after final text update succeeds", + ); + assert.equal(logMessages.length, 0, "successful retry should not log a final failure"); + + runContexts.delete(runCtx.runId); +} + +async function testLateGeneratingIndicatorCannotAppendAfterFinal(): Promise { + const events: Array> = []; + const updates: Array<{ messageId: string; payload: unknown }> = []; + const generatingIndicator = deferred(); + + const channel = { + sendMessage: async () => ({ message: { id: "response-2" } }), + sendEvent: async (event: Record) => { + events.push(event); + if (event.ai_state === "AI_STATE_GENERATING") { + return generatingIndicator.promise; + } + }, + }; + + const client = { + partialUpdateMessage: async (messageId: string, payload: unknown) => { + updates.push({ messageId, payload }); + }, + deleteMessage: async () => undefined, + }; + + const runContexts = new RunContextMap(); + const runCtx = runContext("run-2"); + runContexts.set(runCtx.runId, runCtx); + + const handler = new StreamingHandler({ + client: client as never, + runContexts, + }); + + await handler.onRunStarted(runCtx.runId, channel as never, runCtx); + + const chunkPromise = handler.onTextChunk(runCtx.runId, "late text", 1); + assert.equal( + events.at(-1)?.ai_state, + "AI_STATE_GENERATING", + "test setup should block the generating indicator", + ); + + await handler.onRunCompleted(runCtx.runId); + assert.deepEqual(updates.at(-1), { + messageId: "response-2", + payload: { set: { text: "late text", generating: false } }, + }); + + generatingIndicator.resolve(); + await chunkPromise; + await Promise.resolve(); + + assert.deepEqual( + updates.at(-1), + { + messageId: "response-2", + payload: { set: { text: "late text", generating: false } }, + }, + "late onTextChunk continuation must not enqueue generating:true after final update", + ); + + runContexts.delete(runCtx.runId); +} + +await testFinalUpdateRetries(); +await testLateGeneratingIndicatorCannotAppendAfterFinal(); + +console.log("streaming final retry test passed"); diff --git a/src/channel.ts b/src/channel.ts index fd93184..5a213be 100644 --- a/src/channel.ts +++ b/src/channel.ts @@ -32,6 +32,65 @@ const seenThreads = new Set(); // startAccount again without having called stop() first (e.g. in-process reloads). const activeGatewayCleanup = new Map void>(); +// --------------------------------------------------------------------------- +// Stream Chat target parsing +// --------------------------------------------------------------------------- + +interface ParsedStreamChatTarget { + channelType: string; + channelId: string; + cid: string; +} + +const rejectedStreamChatTargetPrefixes = new Set([ + "streamchat", + "channel", + "group", + "conversation", + "room", + "dm", + "user", + "thread", +]); + +function getEventChannelCid(event: Event): string | undefined { + const channel = event.channel as { cid?: unknown } | undefined; + const cid = typeof channel?.cid === "string" ? channel.cid.trim() : ""; + return cid || undefined; +} + +function buildStreamChatCid(channelType: string, channelId: string): string { + return `${channelType}:${channelId}`; +} + +function parseStreamChatTarget( + raw?: string | null, +): ParsedStreamChatTarget | null { + const cid = raw?.trim(); + if (!cid) return null; + + const separator = cid.indexOf(":"); + if (separator <= 0 || separator === cid.length - 1) return null; + if (separator !== cid.lastIndexOf(":")) return null; + + const channelType = cid.slice(0, separator).trim(); + const channelId = cid.slice(separator + 1).trim(); + if (!channelType || !channelId) return null; + if (rejectedStreamChatTargetPrefixes.has(channelType.toLowerCase())) { + return null; + } + + return { + channelType, + channelId, + cid: buildStreamChatCid(channelType, channelId), + }; +} + +function looksLikeStreamChatTarget(raw: string): boolean { + return parseStreamChatTarget(raw) !== null; +} + // --------------------------------------------------------------------------- // Reactions helper // --------------------------------------------------------------------------- @@ -87,7 +146,7 @@ interface HandleMessageParams { log?: ChannelLogSink; } -async function handleStreamChatMessage(params: HandleMessageParams): Promise { +export async function handleStreamChatMessage(params: HandleMessageParams): Promise { const { cfg, accountId, @@ -110,8 +169,14 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise:streamchat:channel: + // agent::streamchat:channel:: // This ensures each Stream Chat channel gets its own session (per action plan). const route = rt.channel.routing.resolveAgentRoute({ cfg, channel: "streamchat", accountId, - peer: { kind: "channel", id: channelId }, + peer: { kind: "channel", id: streamTarget }, }); const storePath = rt.channel.session.resolveStorePath( @@ -205,7 +270,7 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise { + if (!delta) return; + streamedText += delta; + void streamingHandler.onTextChunk(runId, delta, account.streamingThrottle); + }; + + const appendPartialText = (text: string) => { + if (!text) return; + + if (text.startsWith(streamedText)) { + appendTextDelta(text.slice(streamedText.length)); + return; + } + + appendTextDelta(text); + }; + + const appendFinalText = (full: string) => { + if (!full || full === streamedText) return; + + if (!streamedText || full.startsWith(streamedText)) { + appendTextDelta(full.slice(streamedText.length)); + return; + } + + log?.warn?.( + "[StreamChat] Final text diverged from accumulated stream text; replacing streamed text", + ); + streamedText = full; + streamingHandler.replaceText(runId, full); + }; + + const replyOptions = { + sourceReplyDeliveryMode: "automatic", + onPartialReply: (payload: { text?: string }) => { + if (payload.text) sawPartialText = true; + appendPartialText(payload.text ?? ""); + }, + } as unknown as NonNullable< + Parameters< + typeof rt.channel.reply.dispatchReplyWithBufferedBlockDispatcher + >[0]["replyOptions"] + >; // Dispatch reply via the buffered block dispatcher. // onPartialReply fires for every streaming token (preview streaming). @@ -301,16 +410,7 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise { - const full = payload.text ?? ""; - const delta = full.slice(lastPartialText.length); - lastPartialText = full; - if (delta) { - void streamingHandler.onTextChunk(runId, delta, account.streamingThrottle); - } - }, - }, + replyOptions, dispatcherOptions: { responsePrefix: "", deliver: async ( @@ -334,7 +434,13 @@ async function handleStreamChatMessage(params: HandleMessageParams): Promise target.trim() || undefined, + inferTargetChatType: ({ to }: { to: string }) => + looksLikeStreamChatTarget(to) ? "channel" : undefined, + targetResolver: { + hint: "Use a Stream Chat CID, e.g. messaging:ai-test-channel", + looksLikeId: looksLikeStreamChatTarget, + resolveTarget: async ({ + input, + normalized, + }: { + input: string; + normalized: string; + }) => { + const target = + parseStreamChatTarget(normalized) ?? parseStreamChatTarget(input); + if (!target) return null; + return { + to: target.cid, + kind: "channel" as const, + display: target.cid, + source: "normalized" as const, + }; + }, + }, + formatTargetDisplay: ({ target, display }: { target: string; display?: string }) => + display ?? `#${target}`, + } as unknown as NonNullable, + reload: { configPrefixes: ["channels.streamchat"] }, configSchema: buildChannelConfigSchema(StreamChatConfigSchema), @@ -437,6 +572,19 @@ export const streamchatPlugin: StreamChatChannelPlugin = { outbound: { deliveryMode: "direct", + resolveTarget: ({ to }) => { + const target = parseStreamChatTarget(to); + if (!target) { + return { + ok: false, + error: new Error( + "Stream Chat target must be a CID like messaging:ai-test-channel.", + ), + }; + } + return { ok: true, to: target.cid }; + }, + sendText: async (ctx: ChannelOutboundContext) => { const account = resolveStreamChatAccount(ctx.cfg, ctx.accountId); if (!account.configured) { @@ -449,9 +597,15 @@ export const streamchatPlugin: StreamChatChannelPlugin = { const tempRuntime = new StreamChatClientRuntime(account); try { await tempRuntime.start(); + const target = parseStreamChatTarget(ctx.to); + if (!target) { + throw new Error( + "Stream Chat target must be a CID like messaging:ai-test-channel.", + ); + } const channel = await tempRuntime.getOrQueryChannel( - "messaging", - ctx.to, + target.channelType, + target.channelId, ); const msgPayload: Record = { text: ctx.text }; diff --git a/src/config-schema.ts b/src/config-schema.ts index 294c14b..7117cde 100644 --- a/src/config-schema.ts +++ b/src/config-schema.ts @@ -9,7 +9,7 @@ export const StreamChatConfigSchema: z.ZodTypeAny = z.object({ dmPolicy: z.enum(["open", "pairing"]).optional().default("open"), ackReaction: z.string().optional().default("eyes"), doneReaction: z.string().optional().default("white_check_mark"), - streamingThrottle: z.number().int().min(1).optional().default(15), + streamingThrottle: z.number().int().min(1).optional().default(35), mockResponse: z.string().optional(), accounts: z .record(z.string(), z.lazy(() => StreamChatConfigSchema)) diff --git a/src/streaming.ts b/src/streaming.ts index cf6e925..66957b9 100644 --- a/src/streaming.ts +++ b/src/streaming.ts @@ -3,6 +3,64 @@ import type { ChannelLogSink } from "openclaw/plugin-sdk"; import type { RunContext } from "./types.js"; import type { RunContextMap } from "./run-context.js"; +type PartialUpdatePayload = Parameters[1]; + +function getErrorStatus(err: unknown): number | undefined { + return ( + (err as { status?: number })?.status ?? + (err as { response?: { status?: number } })?.response?.status + ); +} + +function getErrorCode(err: unknown): number | string | undefined { + return (err as { code?: number | string })?.code; +} + +function isRetryableStreamChatError(err: unknown): boolean { + const status = getErrorStatus(err); + if (status === 429 || (status != null && status >= 500 && status < 600)) { + return true; + } + + const code = getErrorCode(err); + if (code === 9 || code === "9") { + return true; + } + + return String(err).includes("Too many requests"); +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function partialUpdateMessageWithRetry( + client: StreamChat, + messageId: string, + payload: PartialUpdatePayload, + options: { + maxAttempts?: number; + baseDelayMs?: number; + } = {}, +): Promise { + const maxAttempts = options.maxAttempts ?? 7; + let delayMs = options.baseDelayMs ?? 500; + + for (let attempt = 1; attempt <= maxAttempts; attempt++) { + try { + await client.partialUpdateMessage(messageId, payload); + return; + } catch (err) { + if (!isRetryableStreamChatError(err) || attempt === maxAttempts) { + throw err; + } + + await sleep(delayMs + Math.floor(Math.random() * 100)); + delayMs = Math.min(delayMs * 2, 5_000); + } + } +} + // --------------------------------------------------------------------------- // safeSendEvent — exponential backoff retry for indicator events // --------------------------------------------------------------------------- @@ -19,9 +77,7 @@ async function safeSendEvent( await channel.sendEvent(event as unknown as Parameters[0]); return; } catch (err: unknown) { - const status = - (err as { status?: number })?.status ?? - (err as { response?: { status?: number } })?.response?.status; + const status = getErrorStatus(err); const retryable = status === 429 || (status != null && status >= 500 && status < 600); if (!retryable || attempt === maxAttempts) { // Swallow in streaming context to avoid breaking generation flow @@ -123,13 +179,13 @@ export class StreamingHandler { /** * Called for each text chunk from the agent. Accumulates text and * periodically flushes via partialUpdateMessage with a throttle pattern: - * - Early bursts: update on odd chunks < 8 (chunks 1, 3, 5, 7) - * - Then every Nth chunk (configurable via streamingThrottle, default 15) + * - Early burst: update on chunks 1 and 5 + * - Then every Nth chunk (configurable via streamingThrottle, default 35) */ async onTextChunk( runId: string, chunk: string, - streamingThrottle: number = 15, + streamingThrottle: number = 35, ): Promise { const stream = this.streams.get(runId); if (!stream || stream.finalized) return; @@ -143,7 +199,7 @@ export class StreamingHandler { // Switch to GENERATING on first text chunk if (stream.indicatorState !== "AI_STATE_GENERATING") { stream.indicatorState = "AI_STATE_GENERATING"; - await safeSendEvent( + void safeSendEvent( stream.channel, { type: "ai_indicator.update", @@ -154,9 +210,9 @@ export class StreamingHandler { ); } - // Throttle: early bursts (odd chunks < 8) then every Nth + // Throttle: early burst (chunks 1 and 5) then every Nth const shouldUpdate = - (n < 8 && n % 2 !== 0) || n % streamingThrottle === 0; + n === 1 || n === 5 || n % streamingThrottle === 0; if (shouldUpdate) { const text = stream.accumulatedText; @@ -176,6 +232,17 @@ export class StreamingHandler { } } + /** + * Replaces the accumulated text for authoritative final payloads that do + * not extend the streamed partial text. + */ + replaceText(runId: string, text: string): void { + const stream = this.streams.get(runId); + if (!stream || stream.finalized) return; + + stream.accumulatedText = text; + } + /** * Called when the agent invokes a tool. Updates the indicator to * EXTERNAL_SOURCES so the UI shows tool activity. @@ -212,14 +279,51 @@ export class StreamingHandler { // Wait for any in-flight partial updates await stream.lastUpdatePromise.catch(() => {}); + const finalText = stream.accumulatedText.trim(); + if (!finalText) { + try { + await client.deleteMessage(stream.messageId, { hardDelete: true }); + } catch (err) { + log?.warn?.( + `[StreamChat][streaming] Empty placeholder delete failed: ${String(err)}`, + ); + try { + await partialUpdateMessageWithRetry( + client, + stream.messageId, + { + set: { text: "", generating: false }, + }, + ); + } catch (fallbackErr) { + log?.error?.( + `[StreamChat][streaming] Empty placeholder fallback update failed: ${String(fallbackErr)}`, + ); + } + } + + await safeSendEvent( + stream.channel, + { type: "ai_indicator.clear", message_id: stream.messageId }, + log, + ); + + this.streams.delete(runId); + return; + } + // Final update with complete text try { - await client.partialUpdateMessage(stream.messageId, { - set: { - text: stream.accumulatedText || "(No response)", - generating: false, + await partialUpdateMessageWithRetry( + client, + stream.messageId, + { + set: { + text: stream.accumulatedText || "(No response)", + generating: false, + }, }, - }); + ); } catch (err) { log?.error?.( `[StreamChat][streaming] Final update failed: ${String(err)}`, @@ -254,9 +358,13 @@ export class StreamingHandler { : `Error: ${error}`; try { - await client.partialUpdateMessage(stream.messageId, { - set: { text: errorText, generating: false }, - }); + await partialUpdateMessageWithRetry( + client, + stream.messageId, + { + set: { text: errorText, generating: false }, + }, + ); } catch (err) { log?.error?.( `[StreamChat][streaming] Error update failed: ${String(err)}`, @@ -293,9 +401,13 @@ export class StreamingHandler { // Just clear generating flag, don't touch text try { - await client.partialUpdateMessage(stream.messageId, { - set: { generating: false }, - }); + await partialUpdateMessageWithRetry( + client, + stream.messageId, + { + set: { generating: false }, + }, + ); } catch (err) { log?.warn?.( `[StreamChat][streaming] Force stop update failed: ${String(err)}`, diff --git a/src/types.ts b/src/types.ts index cf002b9..6b4cd92 100644 --- a/src/types.ts +++ b/src/types.ts @@ -106,7 +106,7 @@ export function resolveStreamChatAccount( dmPolicy: base.dmPolicy ?? "open", ackReaction: base.ackReaction ?? "eyes", doneReaction: base.doneReaction ?? "white_check_mark", - streamingThrottle: base.streamingThrottle ?? 15, + streamingThrottle: base.streamingThrottle ?? 35, mockResponse: base.mockResponse, }; }