From b37e819d74baf63abc830da0e56c12fb09a128dd Mon Sep 17 00:00:00 2001 From: Ayaan Zaidi Date: Tue, 28 Jul 2026 08:12:17 +0900 Subject: [PATCH] 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 --- .../src/bot-message-dispatch-draft.ts | 9 ++++ .../src/bot-message-dispatch-progress.ts | 6 +++ .../telegram/src/bot-message-dispatch-turn.ts | 9 ++++ ...-message-dispatch.progress-updates.test.ts | 29 +++++++++++++ src/auto-reply/get-reply-options.types.ts | 2 + src/auto-reply/reply/followup-runner.test.ts | 21 +++++++-- src/auto-reply/reply/followup-runner.ts | 6 +++ .../reply/followup-turn-admission.test.ts | 43 ++++++++++--------- .../reply/followup-turn-admission.ts | 18 ++++++++ .../reply/followup-turn-execution.test.ts | 43 ++++++++++++++++++- .../reply/followup-turn-execution.ts | 5 ++- 11 files changed, 166 insertions(+), 25 deletions(-) diff --git a/extensions/telegram/src/bot-message-dispatch-draft.ts b/extensions/telegram/src/bot-message-dispatch-draft.ts index 658007468c32..416d578d77de 100644 --- 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]) { diff --git a/extensions/telegram/src/bot-message-dispatch-progress.ts b/extensions/telegram/src/bot-message-dispatch-progress.ts index 053208a5b9d1..fb85ea02492e 100644 --- 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(), diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index 8fb88c7a181f..cb077abd7068 100644 --- 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: diff --git a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts index b52f637016c2..2980a5cff217 100644 --- 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", "Shelling\n🛠️ Exec"), + ); + + 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); diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index 89763bdec745..2cb5cb47bce4 100644 --- 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; + /** Called after an admitted queued followup finishes, including failed attempts. */ + onQueuedFollowupSettled?: () => Promise | 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. */ diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index 9980f930a8c6..4caef68eefff 100644 --- 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 }; + }) => { + 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", ]); diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index ed0a7a0b092f..4734ec4049bc 100644 --- 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(); diff --git a/src/auto-reply/reply/followup-turn-admission.test.ts b/src/auto-reply/reply/followup-turn-admission.test.ts index 4eed723e94ff..5f486991d62e 100644 --- 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 = { diff --git a/src/auto-reply/reply/followup-turn-admission.ts b/src/auto-reply/reply/followup-turn-admission.ts index 612e68811bdb..9924ec52c444 100644 --- 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 { + 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)); } diff --git a/src/auto-reply/reply/followup-turn-execution.test.ts b/src/auto-reply/reply/followup-turn-execution.test.ts index c5a098178a4a..a56619e83bcf 100644 --- 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({ diff --git a/src/auto-reply/reply/followup-turn-execution.ts b/src/auto-reply/reply/followup-turn-execution.ts index 6fef8cc73520..b3937eeae730 100644 --- 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() &&