Fix Stream Chat scheduled delivery targets

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.

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)
     }
   },

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<typeof handleStreamChatMessage>[0]["streamingHandler"],
+  inboundId: string,
+): Promise<void> {
+  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<void> {
+  const deliveredChunks: string[] = [];
+  let completed = false;
+  let capturedReplyOptions: Record<string, unknown> | 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<string, unknown>) => ctx,
+        dispatchReplyWithBufferedBlockDispatcher: async (params: {
+          replyOptions?: Record<string, unknown>;
+          dispatcherOptions: {
+            deliver: (
+              payload: { text?: string; isError?: boolean },
+              info: { kind: string },
+            ) => Promise<void>;
+          };
+        }) => {
+          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<void> {
+  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<string, unknown>) => ctx,
+        dispatchReplyWithBufferedBlockDispatcher: async (params: {
+          replyOptions?: Record<string, unknown>;
+          dispatcherOptions: {
+            deliver: (
+              payload: { text?: string; isError?: boolean },
+              info: { kind: string },
+            ) => Promise<void>;
+          };
+        }) => {
+          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");