fix(telegram): show progress for queued replies (#114590)

* fix(telegram): preserve queued followup progress * fix(auto-reply): restore queued progress lifecycle --------- Co-authored-by: Ayaan Zaidi <hi@obviy.us>
2026-08-02 08:41:38 +00:00 · 2026-07-28 08:12:17 +09:00
parent c1b56266e4
commit b37e819d74
11 changed files with 166 additions and 25 deletions
--- a/extensions/telegram/src/bot-message-dispatch-draft.ts
+++ b/extensions/telegram/src/bot-message-dispatch-draft.ts
@@ -458,6 +458,15 @@ export function createTelegramDraftController(params: {
    answerLane,
    reasoningLane,
    lanes,
+    beginQueuedFollowup: () => {
+      for (const lane of [answerLane, reasoningLane]) {
+        if (!lane.stream) {
+          continue;
+        }
+        lane.stream.forceNewMessage();
+        resetLaneState(lane);
+      }
+    },
    canPushAnswerDraft: () => Boolean(answerLane.stream),
    cleanup: async (superseded: boolean) => {
      for (const lane of [answerLane, reasoningLane]) {
--- a/extensions/telegram/src/bot-message-dispatch-progress.ts
+++ b/extensions/telegram/src/bot-message-dispatch-progress.ts
@@ -326,6 +326,12 @@ export function createTelegramProgressController(params: {

  return {
    applyCollapseSummary,
+    beginQueuedFollowup: () => {
+      finalAnswerDeliveryStarted = false;
+      finalAnswerDelivered = false;
+      sawProgressFinal = false;
+      compositor.beginNewTurn({ force: true });
+    },
    canPushToolProgress,
    cancel: () => compositor.cancel(),
    closeReasoningBurst: () => summary.closeReasoningBurst(),
--- a/extensions/telegram/src/bot-message-dispatch-turn.ts
+++ b/extensions/telegram/src/bot-message-dispatch-turn.ts
@@ -208,6 +208,15 @@ export async function runTelegramDispatchTurn(params: {
                    params.progress.reset();
                  })
              : () => params.progress.closeReasoningBurst(),
+            onQueuedFollowupAdmitted: () => {
+              params.draft.beginQueuedFollowup();
+              params.progress.beginQueuedFollowup();
+            },
+            onQueuedFollowupSettled: async () => {
+              params.progress.cancel();
+              await params.draft.waitForEvents();
+              await params.draft.cleanup(params.isSuperseded());
+            },
            suppressDefaultToolProgressMessages:
              !params.draft.streamDeliveryEnabled || Boolean(params.draft.answerLane.stream),
            forceToolResultProgress:
--- a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts
+++ b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts
@@ -1,5 +1,6 @@
 import { expect, it } from "vitest";
 import {
+  type DispatchReplyWithBufferedBlockDispatcherArgs,
  describeTelegramDispatch,
  createContext,
  createReasoningStreamContext,
@@ -255,6 +256,34 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => {
    expect(draftStream.flush).toHaveBeenCalled();
  });

+  it("reopens progress drafts for queued followups after the source dispatch settles", async () => {
+    const draftStream = createSequencedDraftStream(2001);
+    createTelegramDraftStream.mockReturnValue(draftStream);
+    let queuedReplyOptions: DispatchReplyWithBufferedBlockDispatcherArgs["replyOptions"];
+    dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => {
+      queuedReplyOptions = replyOptions;
+      return { queuedFinal: false };
+    });
+
+    await dispatchWithContext({
+      context: createContext(),
+      streamMode: "progress",
+      telegramCfg: { streaming: { mode: "progress", progress: { label: "Shelling" } } },
+    });
+
+    expect(draftStream.clear).toHaveBeenCalledTimes(1);
+    await queuedReplyOptions?.onQueuedFollowupAdmitted?.();
+    await queuedReplyOptions?.onToolStart?.({ name: "exec", phase: "start" });
+
+    expect(draftStream.forceNewMessage).toHaveBeenCalledTimes(1);
+    expect(draftStream.updatePreview).toHaveBeenCalledWith(
+      telegramProgressPreview("Shelling\n\n🛠️ Exec", "<b>Shelling</b>\n<b>🛠️ Exec</b>"),
+    );
+
+    await queuedReplyOptions?.onQueuedFollowupSettled?.();
+    expect(draftStream.clear).toHaveBeenCalledTimes(2);
+  });
+
  it("keeps eight rolling tool rows beneath a preamble with verbose off", async () => {
    const draftStream = createSequencedDraftStream(2001);
    createTelegramDraftStream.mockReturnValue(draftStream);
--- a/src/auto-reply/get-reply-options.types.ts
+++ b/src/auto-reply/get-reply-options.types.ts
@@ -309,6 +309,8 @@ export type GetReplyOptions = {
  queuedDeliveryCorrelations?: QueuedReplyDeliveryCorrelation[];
  /** Called after a queued followup owns the reply lane, before its model run starts. */
  onQueuedFollowupAdmitted?: () => Promise<void> | void;
+  /** Called after an admitted queued followup finishes, including failed attempts. */
+  onQueuedFollowupSettled?: () => Promise<void> | void;
  /** Allow channel-owned progress UI while final/source reply delivery remains message-tool-only. */
  allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean;
  /** Called when a suppressed source reply mode observes visible delivery through another path. */
--- a/src/auto-reply/reply/followup-runner.test.ts
+++ b/src/auto-reply/reply/followup-runner.test.ts
@@ -29,6 +29,13 @@ vi.mock("./agent-runner-result-accounting.js", () => ({

 vi.mock("./followup-turn-admission.js", () => ({
  admitFollowupTurn: (...args: unknown[]) => state.admit(...args),
+  settleQueuedFollowupPresentation: async (defaults: {
+    opts?: { onQueuedFollowupSettled?: () => Promise<void> | void };
+  }) => {
+    try {
+      await defaults.opts?.onQueuedFollowupSettled?.();
+    } catch {}
+  },
 }));

 vi.mock("./followup-turn-execution.js", () => ({
@@ -253,15 +260,23 @@ describe("createFollowupRunner", () => {
    });
    state.completeLifecycle.mockImplementation(() => order.push("lifecycle-complete"));

-    await createFollowupRunner({ typing, typingMode: "instant", defaultModel: "claude" })(
-      turn.queued,
-    );
+    await createFollowupRunner({
+      typing,
+      typingMode: "instant",
+      defaultModel: "claude",
+      opts: {
+        onQueuedFollowupSettled: () => {
+          order.push("presentation-settled");
+        },
+      },
+    })(turn.queued);

    expect(order).toEqual([
      "progress-drained",
      "accounted",
      "decision",
      "delivered",
+      "presentation-settled",
      "lifecycle-complete",
      "operation-complete",
    ]);
--- a/src/auto-reply/reply/followup-runner.ts
+++ b/src/auto-reply/reply/followup-runner.ts
@@ -7,6 +7,7 @@ import { accountFollowupTurn } from "./agent-runner-result-accounting.js";
 import { deliverFollowupDecision, resolveFollowupDeliveryDecision } from "./followup-delivery.js";
 import {
  admitFollowupTurn,
+  settleQueuedFollowupPresentation,
  type AdmittedFollowupTurn,
  type FollowupRunnerParams,
 } from "./followup-turn-admission.js";
@@ -32,6 +33,7 @@ export function createFollowupRunner(
    let operation: ReplyOperation | undefined;
    let admittedRunId: string | undefined;
    let executionStarted = false;
+    let queuedFollowupAdmitted = false;
    const initiallyAborted =
      queued.abortSignal?.aborted === true || queued.queueAbortSignal?.aborted === true;
    const endDeliveryCorrelations = initiallyAborted
@@ -73,6 +75,7 @@ export function createFollowupRunner(
      const turn: AdmittedFollowupTurn = admission.turn;
      admittedRunId = turn.runId;
      operation = turn.operation;
+      queuedFollowupAdmitted = true;
      const execution = await executeFollowupTurn({
        turn,
        defaults,
@@ -151,6 +154,9 @@ export function createFollowupRunner(
        disposition = { kind: "retry", error };
      }
    } finally {
+      if (queuedFollowupAdmitted) {
+        await settleQueuedFollowupPresentation(defaults);
+      }
      for (const end of endDeliveryCorrelations.toReversed()) {
        try {
          end();
--- a/src/auto-reply/reply/followup-turn-admission.test.ts
+++ b/src/auto-reply/reply/followup-turn-admission.test.ts
@@ -193,6 +193,29 @@ describe("admitFollowupTurn", () => {
    );
  });

+  it("settles presentation and releases the reply operation when dispatcher setup fails", async () => {
+    const operation = createOperation();
+    const failure = new Error("dispatcher reset failed");
+    const onQueuedFollowupSettled = vi.fn(async () => {});
+    state.admitReply.mockResolvedValue({ status: "owned", operation });
+
+    await expect(
+      admitFollowupTurn({
+        queued: createRun(),
+        defaults: createDefaults({
+          opts: {
+            onQueuedFollowupAdmitted: vi.fn(async () => {
+              throw failure;
+            }),
+            onQueuedFollowupSettled,
+          },
+        }),
+      }),
+    ).rejects.toBe(failure);
+    expect(onQueuedFollowupSettled).toHaveBeenCalledOnce();
+    expect(operation.complete).toHaveBeenCalledOnce();
+  });
+
  it("prefers the admitted snapshot over unchanged stale in-memory state", async () => {
    const operation = createOperation("admitted-session");
    const queuedEntry: SessionEntry = { sessionId: "queued-session", updatedAt: 1 };
@@ -252,26 +275,6 @@ describe("admitFollowupTurn", () => {
    }
  });

-  it("releases the reply operation when post-admission dispatcher setup fails", async () => {
-    const operation = createOperation();
-    const failure = new Error("dispatcher reset failed");
-    state.admitReply.mockResolvedValue({ status: "owned", operation });
-
-    await expect(
-      admitFollowupTurn({
-        queued: createRun(),
-        defaults: createDefaults({
-          opts: {
-            onQueuedFollowupAdmitted: vi.fn(async () => {
-              throw failure;
-            }),
-          },
-        }),
-      }),
-    ).rejects.toBe(failure);
-    expect(operation.complete).toHaveBeenCalledOnce();
-  });
-
  it("does not pass stale enqueue-time state into a rotated session generation", async () => {
    const operation = createOperation("rotated-session");
    const staleEntry: SessionEntry = {
--- a/src/auto-reply/reply/followup-turn-admission.ts
+++ b/src/auto-reply/reply/followup-turn-admission.ts
@@ -7,6 +7,7 @@ import { loadSessionEntry } from "../../config/sessions/session-accessor.js";
 import type { TypingMode } from "../../config/types.js";
 import type { OpenClawConfig } from "../../config/types.openclaw.js";
 import { formatErrorMessage } from "../../infra/errors.js";
+import { defaultRuntime } from "../../runtime.js";
 import { resolveSendPolicy } from "../../sessions/send-policy.js";
 import { sessionDeliveryChannel } from "../../utils/delivery-context.shared.js";
 import { markReplyPayloadForSourceSuppressionDelivery } from "../reply-payload.js";
@@ -49,6 +50,18 @@ export type FollowupRunnerParams = {
  toolProgressDetail?: "explain" | "raw";
 };

+export async function settleQueuedFollowupPresentation(
+  defaults: FollowupRunnerParams,
+): Promise<void> {
+  try {
+    await defaults.opts?.onQueuedFollowupSettled?.();
+  } catch (error) {
+    defaultRuntime.error?.(
+      `followup queue: queued presentation cleanup failed: ${formatErrorMessage(error)}`,
+    );
+  }
+}
+
 type FollowupSessionOwner =
  | {
      kind: "detached";
@@ -296,6 +309,7 @@ export async function admitFollowupTurn(params: {
  }
  const operation = admission.operation;
  operation.retainFailureUntilComplete();
+  let queuedFollowupAdmitted = false;
  try {
    await admitFollowupRunLifecycle(params.queued);
    if (isFollowupRunAborted(params.queued)) {
@@ -304,6 +318,7 @@ export async function admitFollowupTurn(params: {

    // Queue drains retain the latest live runner closure per key. Keep local dispatcher
    // callbacks in that closure so retried non-routable items use the newest transport owner.
+    queuedFollowupAdmitted = true;
    await params.defaults.opts?.onQueuedFollowupAdmitted?.();
    if (operation.sessionId !== run.sessionId) {
      run = {
@@ -646,6 +661,9 @@ export async function admitFollowupTurn(params: {
    }
    return { kind: "admitted", turn };
  } catch (error) {
+    if (queuedFollowupAdmitted) {
+      await settleQueuedFollowupPresentation(params.defaults);
+    }
    operation.complete();
    throw error instanceof Error ? error : new Error(formatErrorMessage(error));
  }
--- a/src/auto-reply/reply/followup-turn-execution.test.ts
+++ b/src/auto-reply/reply/followup-turn-execution.test.ts
@@ -240,7 +240,13 @@ describe("executeFollowupTurn", () => {
        typing,
        typingMode: "instant",
        defaultModel: "claude",
-        opts: { onCompactionStart, onCompactionEnd, onReasoningEnd, onNarrationUpdate },
+        opts: {
+          forceToolResultProgress: true,
+          onCompactionStart,
+          onCompactionEnd,
+          onReasoningEnd,
+          onNarrationUpdate,
+        },
      },
      onToolResult,
      onCompactionNoticePayload: vi.fn(async () => {}),
@@ -256,6 +262,41 @@ describe("executeFollowupTurn", () => {
    expect(onNarrationUpdate).not.toHaveBeenCalled();
  });

+  it("honors channel-forced tool progress when verbosity is off", async () => {
+    const onToolStart = vi.fn(async () => {});
+    const onToolResult = vi.fn(async () => {});
+    const turn = createTurn({
+      session: {
+        kind: "session",
+        key: "main",
+        current: () => ({ sessionId: "session", updatedAt: 1, verboseLevel: "off" }),
+        publish: () => undefined,
+        adopt: () => undefined,
+      },
+    });
+    state.execute.mockImplementation(async (params: AgentTurnParams) => {
+      await params.opts?.onToolStart?.({ name: "read", phase: "start" });
+      await params.opts?.onToolResult?.({ text: "working" });
+      return { runId: "run-1", outcome: { kind: "rejected", payload: { text: "done" } } };
+    });
+
+    const result = await executeFollowupTurn({
+      turn,
+      defaults: {
+        typing: createTypingController(),
+        typingMode: "never",
+        defaultModel: "claude",
+        opts: { forceToolResultProgress: true, onToolStart },
+      },
+      onToolResult,
+      onCompactionNoticePayload: vi.fn(async () => {}),
+    });
+    await result.progress.drain();
+
+    expect(onToolStart).toHaveBeenCalledOnce();
+    expect(onToolResult).toHaveBeenCalledWith({ text: "working" }, { runId: "run-1" });
+  });
+
  it("allows explicitly opted-in tool lifecycle while ordinary progress is hidden", async () => {
    const onToolStart = vi.fn(async () => {});
    const turn = createTurn({
--- a/src/auto-reply/reply/followup-turn-execution.ts
+++ b/src/auto-reply/reply/followup-turn-execution.ts
@@ -97,7 +97,10 @@ export async function executeFollowupTurn(params: {
    return level === "on" || level === "full" ? level : "off";
  };
  const shouldEmitToolResult = () =>
-    progressAllowed() && (currentVerboseLevel() === "on" || currentVerboseLevel() === "full");
+    progressAllowed() &&
+    (defaults.opts?.forceToolResultProgress === true ||
+      currentVerboseLevel() === "on" ||
+      currentVerboseLevel() === "full");
  const shouldEmitToolOutput = () => progressAllowed() && currentVerboseLevel() === "full";
  const shouldEmitToolLifecycle = () =>
    progressAllowed() &&