fix(telegram): materialize streaming progress placeholders (#95183)

snowzlmbot · obviyus · web-flow · commit 6441e5646594 · 2026-06-21T19:48:41.000+05:30
* fix(telegram): materialize streaming progress placeholders

* fix(telegram): cancel delayed progress drafts before final

* fix(telegram): satisfy progress placeholder lint

* fix(telegram): cancel delayed draft preview on clear

* refactor(telegram): simplify delayed preview flush

---------

Co-authored-by: snowzlmbot &lt;snowzlmbot@users.noreply.github.com&gt;
Co-authored-by: Ayaan Zaidi &lt;hi@obviy.us&gt;
diff --git a/docs/channels/telegram.md b/docs/channels/telegram.md
@@ -336,7 +336,8 @@ curl "https://api.telegram.org/bot<bot_token>/getUpdates"
     Requirement:
 
     - `channels.telegram.streaming` is `off | partial | block | progress` (default: `partial`)
-    - `progress` keeps one editable status draft for tool progress, clears it at completion, and sends the final answer as a normal message
+    - short initial answer previews are debounced, then materialized after a bounded delay if the run is still active
+    - `progress` keeps one editable status draft for tool progress, shows the stable status label when answer activity arrives before tool progress, clears it at completion, and sends the final answer as a normal message
     - `streaming.preview.toolProgress` controls whether tool/progress updates reuse the same edited preview message (default: `true` when preview streaming is active)
     - `streaming.preview.commandText` controls command/exec detail inside those tool-progress lines: `raw` (default, preserves released behavior) or `status` (tool label only)
     - `streaming.progress.commentary` (default: `false`) opts into assistant commentary/preamble text in the temporary progress draft
diff --git a/docs/concepts/streaming.md b/docs/concepts/streaming.md
@@ -160,9 +160,10 @@ Legacy key migration:
 Telegram:
 
 - Uses `sendMessage` + `editMessageText` preview updates across DMs and group/topics.
+- Short initial previews are still debounced for push-notification UX, but Telegram now materializes them after a bounded delay so active runs do not stay visually silent.
 - Final text edits the active preview in place; long finals reuse that message for the first chunk and send only the remaining chunks.
 - `block` mode rotates the preview into a new message at `streaming.preview.chunk.maxChars` (default 800, capped at Telegram's 4096 edit limit); other modes grow one preview up to 4096 characters.
-- `progress` mode keeps tool progress in an editable status draft, clears that draft at completion, and sends the final answer through normal delivery.
+- `progress` mode keeps tool progress in an editable status draft, materializes the status label when answer streaming is active but no tool line is available yet, clears that draft at completion, and sends the final answer through normal delivery.
 - If the final edit fails before the completed text is confirmed, OpenClaw uses normal final delivery and cleans up the stale preview.
 - Preview streaming is skipped when Telegram block streaming is explicitly enabled (to avoid double-streaming).
 - `/reasoning stream` can write reasoning to a transient preview that is deleted after final delivery.
diff --git a/extensions/telegram/src/bot-message-dispatch.test.ts b/extensions/telegram/src/bot-message-dispatch.test.ts
@@ -637,6 +637,7 @@ describe("dispatchTelegramMessage draft streaming", () => {
       chatId: 123,
       thread: { id: 777, scope: "dm" },
       minInitialChars: 30,
+      minInitialDelayMs: 5000,
     });
     expect(draftStream.update).toHaveBeenCalledWith("Hello");
     const delivery = expectDeliverRepliesParams({ thread: { id: 777, scope: "dm" } });
@@ -2600,6 +2601,27 @@ describe("dispatchTelegramMessage draft streaming", () => {
     expect(editMessageTelegram).not.toHaveBeenCalled();
   });
 
+  it("shows a stable progress placeholder for progress-mode answer activity", async () => {
+    const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
+    dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => {
+      await replyOptions?.onPartialReply?.({ text: "Short" });
+      await replyOptions?.onPartialReply?.({ text: "Short answer" });
+      return { queuedFinal: false };
+    });
+
+    await dispatchWithContext({
+      context: createContext(),
+      streamMode: "progress",
+      telegramCfg: { streaming: { mode: "progress", progress: { label: "Shelling" } } },
+    });
+
+    expect(answerDraftStream.update).not.toHaveBeenCalledWith("Short");
+    expect(answerDraftStream.update).not.toHaveBeenCalledWith("Short answer");
+    expect(answerDraftStream.updatePreview).toHaveBeenCalledWith(
+      telegramHtmlPreview("<b>Shelling</b>"),
+    );
+  });
+
   it("replaces Telegram command progress items with matching command output", async () => {
     const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
     dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => {
diff --git a/extensions/telegram/src/bot-message-dispatch.ts b/extensions/telegram/src/bot-message-dispatch.ts
@@ -152,6 +152,7 @@ const silentReplyDispatchLogger = createSubsystemLogger("telegram/silent-reply-d
 
 /** Minimum chars before sending first streaming message (improves push notification UX) */
 const DRAFT_MIN_INITIAL_CHARS = 30;
+const DRAFT_MIN_INITIAL_DELAY_MS = 5_000;
 
 type DraftPartialTextUpdate = {
   text: string;
@@ -1005,6 +1006,7 @@ export const dispatchTelegramMessage = async ({
           replyToMessageId: draftReplyToMessageId,
           richMessages: telegramCfg.richMessages,
           minInitialChars: draftMinInitialChars,
+          minInitialDelayMs: draftMinInitialChars > 0 ? DRAFT_MIN_INITIAL_DELAY_MS : undefined,
           renderText: renderStreamText,
           onSupersededPreview: (superseded) => {
             if (superseded.retain) {
@@ -1364,7 +1366,7 @@ export const dispatchTelegramMessage = async ({
       recomputeQueuedAnswerBlockRotations();
     }
   };
-  const updateDraftFromPartial = (lane: DraftLaneState, update: DraftPartialTextUpdate) => {
+  const updateDraftFromPartial = async (lane: DraftLaneState, update: DraftPartialTextUpdate) => {
     const laneStream = lane.stream;
     if (!laneStream || !update.text) {
       return;
@@ -1376,6 +1378,7 @@ export const dispatchTelegramMessage = async ({
     }
     if (lane === answerLane) {
       if (streamMode === "progress") {
+        await progressDraft.noteActivity();
         return;
       }
       resetAnswerToolProgressDraft();
@@ -1402,7 +1405,7 @@ export const dispatchTelegramMessage = async ({
         reasoningStepState.noteReasoningHint();
         reasoningStepState.noteReasoningDelivered();
       }
-      updateDraftFromPartial(lanes[segment.lane], segment.update);
+      await updateDraftFromPartial(lanes[segment.lane], segment.update);
     }
   };
   const flushDraftLane = async (lane: DraftLaneState) => {
diff --git a/extensions/telegram/src/draft-stream.test.ts b/extensions/telegram/src/draft-stream.test.ts
@@ -843,11 +843,16 @@ describe("createTelegramDraftStream", () => {
 describe("draft stream initial message debounce", () => {
   const createMockApi = () => createMockDraftApi(async () => ({ message_id: 42 }));
 
-  function createDebouncedStream(api: ReturnType<typeof createMockApi>, minInitialChars = 30) {
+  function createDebouncedStream(
+    api: ReturnType<typeof createMockApi>,
+    minInitialChars = 30,
+    minInitialDelayMs?: number,
+  ) {
     return createTelegramDraftStream({
       api: api as unknown as Bot["api"],
       chatId: 123,
       minInitialChars,
+      minInitialDelayMs,
     });
   }
 
@@ -916,6 +921,36 @@ describe("draft stream initial message debounce", () => {
       expect(api.sendMessage).toHaveBeenCalled();
     });
 
+    it("materializes a short first message after the initial delay", async () => {
+      const api = createMockApi();
+      const stream = createDebouncedStream(api, 30, 5000);
+
+      stream.update("Processing");
+      await stream.flush();
+      expect(api.sendMessage).not.toHaveBeenCalled();
+
+      await vi.advanceTimersByTimeAsync(5000);
+
+      expectPreviewSend(api, "Processing");
+    });
+
+    it("cancels a delayed first message when clear() removes the draft", async () => {
+      const api = createMockApi();
+      const stream = createDebouncedStream(api, 30, 5000);
+
+      stream.update("Processing");
+      await stream.flush();
+      expect(api.sendMessage).not.toHaveBeenCalled();
+      expect(vi.getTimerCount()).toBe(1);
+
+      await stream.clear();
+      expect(vi.getTimerCount()).toBe(0);
+
+      await vi.advanceTimersByTimeAsync(5000);
+      expect(api.sendMessage).not.toHaveBeenCalled();
+      expect(api.editMessageText).not.toHaveBeenCalled();
+    });
+
     it("works with longer text above threshold", async () => {
       const api = createMockApi();
       const stream = createDebouncedStream(api);
diff --git a/extensions/telegram/src/draft-stream.ts b/extensions/telegram/src/draft-stream.ts
@@ -178,6 +178,8 @@ export function createTelegramDraftStream(params: {
   throttleMs?: number;
   /** Minimum chars before sending first message (debounce for push notifications) */
   minInitialChars?: number;
+  /** Maximum time to hold a short first preview before materializing it anyway. */
+  minInitialDelayMs?: number;
   /** Optional preview renderer (e.g. markdown -> HTML + parse mode). */
   renderText?: (text: string) => TelegramDraftPreview;
   /** Called when a late send resolves after forceNewMessage() switched generations. */
@@ -190,6 +192,7 @@ export function createTelegramDraftStream(params: {
   const maxChars = Math.min(params.maxChars ?? transportLimit, transportLimit);
   const throttleMs = Math.max(250, params.throttleMs ?? DEFAULT_THROTTLE_MS);
   const minInitialChars = params.minInitialChars;
+  const minInitialDelayMs = params.minInitialDelayMs;
   const chatId = params.chatId;
   const threadParams = buildTelegramThreadParams(params.thread);
   const replyToMessageId = normalizeTelegramReplyToMessageId(params.replyToMessageId);
@@ -224,6 +227,8 @@ export function createTelegramDraftStream(params: {
   let lastDeliveredText = "";
   let lastRequestedText = "";
   let lastRequestedPreview: TelegramDraftPreview | undefined;
+  let firstShortPreviewSeenMs: number | undefined;
+  let initialPreviewTimer: ReturnType<typeof setTimeout> | undefined;
   let previewRevision = 0;
   let generation = 0;
   let deliveredTextOffset = 0;
@@ -319,6 +324,26 @@ export function createTelegramDraftStream(params: {
     streamVisibleSinceMs = visibleSinceMs;
     return true;
   };
+  const clearInitialPreviewTimer = () => {
+    if (initialPreviewTimer) {
+      clearTimeout(initialPreviewTimer);
+      initialPreviewTimer = undefined;
+    }
+  };
+  const scheduleInitialPreviewFlush = (delayMs: number) => {
+    if (initialPreviewTimer) {
+      return;
+    }
+    initialPreviewTimer = setTimeout(
+      () => {
+        initialPreviewTimer = undefined;
+        void flushInitialPreview().catch((err: unknown) => {
+          params.warn?.(`telegram stream preview delayed send failed: ${formatErrorMessage(err)}`);
+        });
+      },
+      Math.max(0, delayMs),
+    );
+  };
   const stopOversizedPreview = (payloadLength: number): false => {
     streamState.stopped = true;
     params.warn?.(`telegram stream preview stopped (text length ${payloadLength} > ${maxChars})`);
@@ -405,8 +430,24 @@ export function createTelegramDraftStream(params: {
 
     if (typeof streamMessageId !== "number" && minInitialChars != null && !streamState.final) {
       if (renderedText.length < minInitialChars) {
-        return false;
+        if (minInitialDelayMs == null) {
+          return false;
+        }
+        const now = Date.now();
+        firstShortPreviewSeenMs ??= now;
+        const remainingDelayMs = minInitialDelayMs - (now - firstShortPreviewSeenMs);
+        if (remainingDelayMs > 0) {
+          scheduleInitialPreviewFlush(remainingDelayMs);
+          return false;
+        }
+        clearInitialPreviewTimer();
+      } else {
+        firstShortPreviewSeenMs = undefined;
+        clearInitialPreviewTimer();
       }
+    } else {
+      firstShortPreviewSeenMs = undefined;
+      clearInitialPreviewTimer();
     }
 
     const previousSentPreviewKey = lastSentPreviewKey;
@@ -467,6 +508,7 @@ export function createTelegramDraftStream(params: {
     state: streamState,
     sendOrEditStreamMessage,
   });
+  const flushInitialPreview = loop.flush;
 
   const requestDraftUpdate = (text: string, preview?: TelegramDraftPreview) => {
     if (streamState.stopped || streamState.final) {
@@ -513,6 +555,8 @@ export function createTelegramDraftStream(params: {
     messageSendAttempted = false;
     streamMessageId = undefined;
     streamVisibleSinceMs = undefined;
+    firstShortPreviewSeenMs = undefined;
+    clearInitialPreviewTimer();
     lastSentPreviewKey = "";
     if (options?.resetOffset !== false) {
       deliveredTextOffset = 0;
@@ -526,6 +570,7 @@ export function createTelegramDraftStream(params: {
   };
 
   const clear = async () => {
+    clearInitialPreviewTimer();
     const messageId = await takeMessageIdAfterStop({
       stopForClear,
       readMessageId: () => streamMessageId,
@@ -544,6 +589,7 @@ export function createTelegramDraftStream(params: {
   };
 
   const discard = async () => {
+    clearInitialPreviewTimer();
     await stopForClear();
   };
 
diff --git a/src/channels/progress-draft-compositor.test.ts b/src/channels/progress-draft-compositor.test.ts
@@ -21,6 +21,70 @@ describe("createChannelProgressDraftCompositor", () => {
     expect(update).toHaveBeenCalledWith("Shelling", { flush: true, lines: [] });
   });
 
+  it("materializes a label-only progress draft after upstream answer activity", async () => {
+    vi.useFakeTimers();
+    try {
+      const update = vi.fn();
+      const progress = createChannelProgressDraftCompositor({
+        entry: { streaming: { mode: "progress", progress: { label: "Shelling" } } },
+        mode: "progress",
+        active: true,
+        seed: "test",
+        update,
+      });
+
+      await progress.noteActivity();
+      expect(update).not.toHaveBeenCalled();
+
+      await vi.advanceTimersByTimeAsync(DEFAULT_PROGRESS_DRAFT_INITIAL_DELAY_MS);
+
+      expect(update).toHaveBeenCalledWith("Shelling", { flush: true, lines: [] });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not materialize delayed answer activity after final delivery starts or lands", async () => {
+    vi.useFakeTimers();
+    try {
+      for (const markFinal of ["markFinalReplyStarted", "markFinalReplyDelivered"] as const) {
+        const update = vi.fn();
+        const progress = createChannelProgressDraftCompositor({
+          entry: { streaming: { mode: "progress", progress: { label: "Shelling" } } },
+          mode: "progress",
+          active: true,
+          seed: "test",
+          update,
+        });
+
+        await progress.noteActivity();
+        progress[markFinal]();
+        await vi.advanceTimersByTimeAsync(DEFAULT_PROGRESS_DRAFT_INITIAL_DELAY_MS);
+
+        expect(update).not.toHaveBeenCalled();
+        expect(progress.hasStarted).toBe(false);
+      }
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("starts a label-only progress draft on repeated upstream answer activity", async () => {
+    const update = vi.fn();
+    const progress = createChannelProgressDraftCompositor({
+      entry: { streaming: { mode: "progress", progress: { label: "Shelling" } } },
+      mode: "progress",
+      active: true,
+      seed: "test",
+      update,
+    });
+
+    await progress.noteActivity();
+    await progress.noteActivity();
+
+    expect(update).toHaveBeenCalledWith("Shelling", { flush: true, lines: [] });
+  });
+
   it("passes structured progress lines to draft updates", async () => {
     const update = vi.fn();
     const progress = createChannelProgressDraftCompositor({
diff --git a/src/channels/progress-draft-compositor.ts b/src/channels/progress-draft-compositor.ts
@@ -81,7 +81,7 @@ export function createChannelProgressDraftCompositor(params: {
   };
 
   const render = async (options?: { flush?: boolean }): Promise<boolean> => {
-    if (!params.active || params.mode !== "progress") {
+    if (!params.active || params.mode !== "progress" || finalReplyStarted || finalReplyDelivered) {
       return false;
     }
     const text = formatDraftText();
@@ -197,9 +197,11 @@ export function createChannelProgressDraftCompositor(params: {
     },
     markFinalReplyStarted() {
       finalReplyStarted = true;
+      gate.cancel();
     },
     markFinalReplyDelivered() {
       finalReplyDelivered = true;
+      gate.cancel();
     },
     reset() {
       clearProgressState(false);
@@ -216,6 +218,27 @@ export function createChannelProgressDraftCompositor(params: {
     start() {
       return gate.startNow();
     },
+    async noteActivity(options?: { startImmediately?: boolean }) {
+      if (
+        !params.active ||
+        params.mode !== "progress" ||
+        progressSuppressed ||
+        finalReplyStarted ||
+        finalReplyDelivered
+      ) {
+        return false;
+      }
+      if (options?.startImmediately) {
+        await gate.startNow();
+        return gate.hasStarted ? await render({ flush: true }) : false;
+      }
+      const alreadyStarted = gate.hasStarted;
+      const progressActive = await gate.noteWork();
+      if ((alreadyStarted || progressActive) && gate.hasStarted) {
+        return await render();
+      }
+      return false;
+    },
     pushToolProgress: noteProgress,
     async pushReasoningProgress(text?: string, options?: { snapshot?: boolean }) {
       if (
