fix(reply): preserve streamed Telegram replies on failure

Preserve already-visible Telegram partial replies when a later run failure occurs. Finalize the same draft with the terminal notice while keeping the reply operation in run_failed state.

Co-authored-by: Dinesh Suthar <dineshsld20@gmail.com>
Co-authored-by: Ayaan Zaidi <hi@obviy.us>

extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts

@@ -1,3 +1,4 @@
+import { dispatchReplyWithBufferedBlockDispatcher as dispatchReplyWithBufferedBlockDispatcherRuntime } from "openclaw/plugin-sdk/reply-dispatch-runtime";
 import { expect, it, vi } from "vitest";
 import {
   describeTelegramDispatch,
@@ -78,6 +79,49 @@ describeTelegramDispatch("dispatchTelegramMessage draft-failures-progress", () =
     );
   });
 
+  it.each([
+    { label: "direct chat", createSessionPayload: createDirectSessionPayload },
+    {
+      label: "group chat",
+      createSessionPayload: () => ({
+        ...createDirectSessionPayload(),
+        SessionKey: "agent:test:telegram:group:-100123",
+        ChatType: "group" as const,
+      }),
+    },
+  ])(
+    "finalizes the default streamed draft in place after an unexpected reply failure in a $label",
+    async ({ createSessionPayload }) => {
+      const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
+      dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async (params) => {
+        expect(params.replyOptions?.disableBlockStreaming).toBe(true);
+        return await dispatchReplyWithBufferedBlockDispatcherRuntime({
+          ...params,
+          replyResolver: async (_ctx, opts) => {
+            await opts?.onPartialReply?.({ text: "partial answer" });
+            throw new Error("unexpected model failure");
+          },
+        });
+      });
+
+      await dispatchWithContext({
+        context: createContext({ ctxPayload: createSessionPayload() }),
+        streamMode: "partial",
+        telegramCfg: { streaming: { mode: "partial" } },
+      });
+
+      expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "partial answer");
+      expect(answerDraftStream.update).toHaveBeenCalledTimes(2);
+      expect(answerDraftStream.update).toHaveBeenLastCalledWith(
+        expect.stringMatching(
+          /^partial answer\n\n.*Something went wrong while processing your request\. Please try again, or use \/new to start a fresh session\.$/,
+        ),
+      );
+      expect(answerDraftStream.clear).not.toHaveBeenCalled();
+      expect(deliverReplies).not.toHaveBeenCalled();
+    },
+  );
+
   it("returns retryable when dispatch fails after partial output and the fallback is not delivered", async () => {
     deliverReplies.mockResolvedValueOnce({ delivered: true });
     deliverReplies.mockResolvedValueOnce({ delivered: false });

extensions/telegram/src/lane-delivery-text-deliverer.ts

@@ -275,9 +275,10 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
     buttons: TelegramInlineButtons | undefined,
     promptContextSequence: TelegramPromptContextProjectionSequence,
     followedByDurablePayload = false,
+    allowErrorPayload = false,
   ): Promise<LaneDeliveryResult | undefined> => {
     const stream = lane.stream;
-    if (!stream || text.length === 0 || payload.isError) {
+    if (!stream || text.length === 0 || (payload.isError && !allowErrorPayload)) {
       return undefined;
     }
     rotateFinalizedStream(lane);
@@ -390,17 +391,40 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
     const isDurableFinal = infoKind === "final";
     const finalizePreview = requestedFinalizePreview ?? isDurableFinal;
     const durable = requestedDurable ?? isDurableFinal;
+    const streamedErrorDraftText =
+      isDurableFinal &&
+      payload.isError === true &&
+      laneName === "answer" &&
+      lane.stream &&
+      lane.hasStreamedMessage &&
+      !lane.finalized &&
+      !reply.hasMedia &&
+      text.trim()
+        ? (() => {
+            const existing = (
+              lane.lastPartialText ||
+              lane.stream?.lastDeliveredText?.() ||
+              ""
+            ).trimEnd();
+            const notice = text.trim();
+            return existing && !existing.endsWith(notice)
+              ? `${existing}\n\n${notice}`
+              : existing || notice;
+          })()
+        : undefined;
     const streamed =
       allowStream && !reply.hasMedia
         ? await streamText(
             laneName,
             lane,
-            text,
+            streamedErrorDraftText ?? text,
             payload,
             isDurableFinal,
             finalizePreview,
             buttons,
             promptContextSequence,
+            false,
+            streamedErrorDraftText !== undefined,
           )
         : undefined;
     if (streamed) {

src/auto-reply/reply/agent-runner-core.ts

@@ -30,7 +30,10 @@ import type { TemplateContext } from "../templating.js";
 import type { VerboseLevel } from "../thinking.js";
 import { SILENT_REPLY_TOKEN } from "../tokens.js";
 import type { GetReplyOptions, ReplyPayload } from "../types.js";
-import { buildKnownAgentRunFailureReplyPayload } from "./agent-runner-failure-reply.js";
+import {
+  buildKnownAgentRunFailureReplyPayload,
+  buildTerminalAgentRunFailureReplyPayload,
+} from "./agent-runner-failure-reply.js";
 import type { BlockReplyPipeline } from "./block-reply-pipeline.js";
 import { resolveEffectiveReplyRoute } from "./effective-reply-route.js";
 import type { InternalGetReplyOptions } from "./get-reply.types.js";
@@ -365,6 +368,9 @@ export async function handleReplyAgentRunError(
   error: unknown,
   context: {
     cfg: OpenClawConfig;
+    blockReplyPipeline: BlockReplyPipeline | null;
+    didDeliverVisiblePartialReply: () => boolean;
+    isHeartbeat: boolean;
     isRestartRecoveryArmed: () => boolean;
     replyOperation: ReplyOperation;
     resolvedVerboseLevel: VerboseLevel;
@@ -374,6 +380,9 @@ export async function handleReplyAgentRunError(
 ): Promise<ReplyPayload | undefined> {
   const {
     cfg,
+    blockReplyPipeline,
+    didDeliverVisiblePartialReply,
+    isHeartbeat,
     isRestartRecoveryArmed,
     replyOperation,
     resolvedVerboseLevel,
@@ -426,6 +435,28 @@ export async function handleReplyAgentRunError(
     replyOperation.fail("run_failed", error);
     return returnWithQueuedFollowupDrain(knownFailurePayload);
   }
+  if (blockReplyPipeline) {
+    try {
+      await blockReplyPipeline.flush({ force: true });
+    } catch (flushError) {
+      logVerbose(
+        `failed to flush streamed reply blocks before surfacing run failure: ${String(flushError)}`,
+      );
+    }
+  }
+  const didDeliverVisibleReply =
+    (blockReplyPipeline?.didStreamTerminalReply?.() === true && !blockReplyPipeline.isAborted()) ||
+    didDeliverVisiblePartialReply();
+  if (!isHeartbeat && didDeliverVisibleReply && !replyOperation.abortSignal.aborted) {
+    replyOperation.fail("run_failed", error);
+    return returnWithQueuedFollowupDrain(
+      buildTerminalAgentRunFailureReplyPayload({
+        visibleReplyDelivered: true,
+        sessionCtx,
+        cfg,
+      }),
+    );
+  }
   replyOperation.fail("run_failed", error);
   // Keep the followup queue moving even when an unexpected exception escapes
   // the run path; the caller still receives the original error.

src/auto-reply/reply/agent-runner-execution.ts

@@ -446,6 +446,7 @@ async function executeAgentTurnInternalWithRetryState(
   const terminalFailurePayload = terminalRunFailed
     ? buildTerminalAgentRunFailureReplyPayload({
         isHeartbeat: params.isHeartbeat,
+        visibleReplyDelivered: false,
         sessionCtx: params.sessionCtx,
         cfg: params.followupRun.run.config,
       })

src/auto-reply/reply/agent-runner-failure-reply.ts

@@ -471,18 +471,24 @@ export function markAgentRunFailureReplyPayload<T extends ReplyPayload>(payload:
 
 export function buildTerminalAgentRunFailureReplyPayload(params: {
   isHeartbeat?: boolean;
+  visibleReplyDelivered: boolean;
   sessionCtx: ExternalFailureConversationContext;
   cfg?: OpenClawConfig;
 }): ReplyPayload {
+  const text = params.isHeartbeat
+    ? HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT
+    : GENERIC_EXTERNAL_RUN_FAILURE_TEXT;
+  // Once output is visible, hiding its terminal failure leaves a misleading partial reply.
+  // Keep normal group silence only for failures that produced no visible output.
   return markAgentRunFailureReplyPayload({
-    text: resolveExternalRunFailureTextForConversation({
-      text: params.isHeartbeat
-        ? HEARTBEAT_EXTERNAL_RUN_FAILURE_TEXT
-        : GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
-      sessionCtx: params.sessionCtx,
-      isGenericRunnerFailure: true,
-      cfg: params.cfg,
-    }),
+    text: params.visibleReplyDelivered
+      ? text
+      : resolveExternalRunFailureTextForConversation({
+          text,
+          sessionCtx: params.sessionCtx,
+          isGenericRunnerFailure: true,
+          cfg: params.cfg,
+        }),
   });
 }
 

src/auto-reply/reply/agent-runner-run.ts

@@ -10,6 +10,7 @@ import { hasRestartRecoverySourceClaim } from "../../config/sessions/restart-rec
 import { loadSessionEntry, updateSessionEntry } from "../../config/sessions/session-accessor.js";
 import { logVerbose } from "../../globals.js";
 import { measureDiagnosticsTimelineSpan } from "../../infra/diagnostics-timeline.js";
+import { hasOutboundReplyContent } from "../../plugin-sdk/reply-payload.js";
 import {
   buildHandledBeforeAgentReplyPayloads,
   runBeforeAgentReplyForTurn,
@@ -111,6 +112,18 @@ export async function runReplyAgent(
   const activeRunQueueMode = effectiveResetTriggered ? "interrupt" : resolvedQueue.mode;
 
   const isHeartbeat = opts?.isHeartbeat === true;
+  let didDeliverVisiblePartialReply = false;
+  const runOpts = opts?.onPartialReply
+    ? {
+        ...opts,
+        onPartialReply: async (payload: Parameters<NonNullable<typeof opts.onPartialReply>>[0]) => {
+          await opts.onPartialReply?.(payload);
+          if (hasOutboundReplyContent(payload, { trimText: true })) {
+            didDeliverVisiblePartialReply = true;
+          }
+        },
+      }
+    : opts;
   const replyOperationRunState = resolveReplyOperationRunState(opts);
   const traceAttributes = {
     provider: followupRun.run.provider,
@@ -654,7 +667,7 @@ export async function runReplyAgent(
       getActiveSessionEntry: () => activeSessionEntry,
       isHeartbeat,
       isRestartRecoveryArmed,
-      opts,
+      opts: runOpts,
       pendingToolTasks,
       performSessionReset: resetSession,
       queueKey,
@@ -694,7 +707,10 @@ export async function runReplyAgent(
     });
   } catch (error) {
     return await handleReplyAgentRunError(error, {
+      blockReplyPipeline,
       cfg,
+      didDeliverVisiblePartialReply: () => didDeliverVisiblePartialReply,
+      isHeartbeat,
       isRestartRecoveryArmed,
       replyOperation,
       resolvedVerboseLevel,

src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts

@@ -1128,6 +1128,169 @@ describe("runReplyAgent heartbeat followup guard", () => {
       persistSpy.mockRestore();
     }
   });
+
+  it.each([
+    { label: "direct chat", sessionCtx: {} },
+    {
+      label: "group chat",
+      sessionCtx: {
+        ChatType: "group" as const,
+        SessionKey: "agent:test:telegram:group:-100123",
+      },
+    },
+  ])(
+    "returns a terminal failure in a $label after a delivered partial with block streaming disabled",
+    async ({ sessionCtx }) => {
+      const accounting = await import("./session-run-accounting.js");
+      const persistSpy = vi
+        .spyOn(accounting, "persistRunSessionUsage")
+        .mockRejectedValueOnce(new Error("persist exploded"));
+      const onPartialReply = vi.fn();
+      state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => {
+        await params.onPartialReply?.({ text: "partial answer" });
+        return {
+          payloads: [{ text: "final answer" }],
+          meta: { agentMeta: { usage: { input: 1, output: 1 } } },
+        };
+      });
+
+      try {
+        const { run } = createMinimalRun({
+          blockStreamingEnabled: false,
+          opts: { onPartialReply },
+          sessionCtx,
+        });
+        const result = await run();
+        const payload = Array.isArray(result) ? result[0] : result;
+
+        expect(onPartialReply).toHaveBeenCalledWith({
+          text: "partial answer",
+          mediaUrls: undefined,
+        });
+        expect(payload).toMatchObject({
+          text: GENERIC_EXTERNAL_RUN_FAILURE_TEXT,
+          isError: true,
+        });
+      } finally {
+        persistSpy.mockRestore();
+      }
+    },
+  );
+
+  it("rethrows after a delivered partial without visible content", async () => {
+    const accounting = await import("./session-run-accounting.js");
+    const persistSpy = vi
+      .spyOn(accounting, "persistRunSessionUsage")
+      .mockRejectedValueOnce(new Error("persist exploded"));
+    const onPartialReply = vi.fn();
+    state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => {
+      await params.onPartialReply?.({ text: "   " });
+      return {
+        payloads: [{ text: "final answer" }],
+        meta: { agentMeta: { usage: { input: 1, output: 1 } } },
+      };
+    });
+
+    try {
+      const { run } = createMinimalRun({
+        blockStreamingEnabled: false,
+        opts: { onPartialReply },
+      });
+
+      await expect(run()).rejects.toThrow("persist exploded");
+    } finally {
+      persistSpy.mockRestore();
+    }
+  });
+
+  it("rethrows heartbeat failures after a delivered partial", async () => {
+    const accounting = await import("./session-run-accounting.js");
+    const persistSpy = vi
+      .spyOn(accounting, "persistRunSessionUsage")
+      .mockRejectedValueOnce(new Error("persist exploded"));
+    const onPartialReply = vi.fn();
+    state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => {
+      await params.onPartialReply?.({ text: "heartbeat detail" });
+      return {
+        payloads: [{ text: "HEARTBEAT_OK" }],
+        meta: { agentMeta: { usage: { input: 1, output: 1 } } },
+      };
+    });
+
+    try {
+      const { run } = createMinimalRun({
+        blockStreamingEnabled: false,
+        opts: { isHeartbeat: true, onPartialReply },
+      });
+
+      await expect(run()).rejects.toThrow("persist exploded");
+    } finally {
+      persistSpy.mockRestore();
+    }
+  });
+
+  it("keeps user aborts silent after a delivered partial", async () => {
+    const replyOperation = createReplyOperation({
+      sessionKey: "main",
+      sessionId: "session",
+      resetTriggered: false,
+    });
+    const onPartialReply = vi.fn();
+    state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => {
+      await params.onPartialReply?.({ text: "partial answer" });
+      replyOperation.abortByUser();
+      throw new Error("run stopped");
+    });
+
+    const { run } = createMinimalRun({
+      blockStreamingEnabled: false,
+      opts: { onPartialReply },
+      replyOperation,
+    });
+    const result = await run();
+
+    expect(result).toEqual({ text: "NO_REPLY" });
+  });
+
+  it.each(["reasoning", "commentary"] as const)(
+    "rethrows after %s-only block streaming",
+    async (lane) => {
+      const accounting = await import("./session-run-accounting.js");
+      const persistSpy = vi
+        .spyOn(accounting, "persistRunSessionUsage")
+        .mockRejectedValueOnce(new Error("persist exploded"));
+      const onBlockReply = vi.fn();
+      state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => {
+        await params.onBlockReply?.({
+          text: `internal ${lane}`,
+          ...(lane === "reasoning" ? { isReasoning: true } : { isCommentary: true }),
+        });
+        return {
+          payloads: [{ text: "final answer" }],
+          meta: { agentMeta: { usage: { input: 1, output: 1 } } },
+        };
+      });
+
+      try {
+        const { run } = createMinimalRun({
+          blockStreamingEnabled: true,
+          opts: {
+            onBlockReply,
+            reasoningPayloadsEnabled: lane === "reasoning",
+            commentaryPayloadsEnabled: lane === "commentary",
+          },
+        });
+
+        await expect(run()).rejects.toThrow("persist exploded");
+        expect(onBlockReply).toHaveBeenCalledWith(
+          expect.objectContaining({ text: `internal ${lane}` }),
+          expect.any(Object),
+        );
+      } finally {
+        persistSpy.mockRestore();
+      }
+    },
+  );
 });
 
 describe("runReplyAgent pending final delivery capture", () => {

src/auto-reply/reply/dispatch-from-config.execute.ts

@@ -12,6 +12,7 @@ import {
   isReplyPayloadStatusNotice,
   type ReplyPayload,
 } from "../reply-payload.js";
+import { buildTerminalAgentRunFailureReplyPayload } from "./agent-runner-failure-reply.js";
 import { takeCommandSessionMetadataChanges } from "./command-session-metadata.js";
 import { runWithDispatchAbortSignal } from "./dispatch-from-config.abort.js";
 import {
@@ -31,6 +32,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
     ctx,
     deliveryChannel,
     dispatcher,
+    failDispatchReplyOperation,
     flushPendingCommentaryProgress,
     getDispatchAbortOperation,
     getDispatchAbortSignal,
@@ -64,6 +66,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
     wrapProgressCallback,
   } = state;
   let deliberateSilentTerminalReply = false;
+  let didDeliverVisiblePartialReply = false;
   const replyResult = await runWithDispatchLifecycleAdmission(
     async () =>
       await runWithDispatchAbortSignal(
@@ -89,7 +92,13 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
                 shouldSuppressToolErrorWarnings: state.shouldSuppressToolErrorWarnings,
                 typingPolicy: typing.typingPolicy,
                 suppressTyping: typing.suppressTyping,
-                onPartialReply: wrapProgressCallback(params.replyOptions?.onPartialReply),
+                onPartialReply: wrapProgressCallback(params.replyOptions?.onPartialReply, {
+                  onVisible: (payload) => {
+                    if (hasOutboundReplyContent(payload, { trimText: true })) {
+                      didDeliverVisiblePartialReply = true;
+                    }
+                  },
+                }),
                 onReasoningStream: wrapProgressCallback(params.replyOptions?.onReasoningStream),
                 streamReasoningInNonStreamModes:
                   params.replyOptions?.streamReasoningInNonStreamModes,
@@ -491,7 +500,21 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState)
           ),
         trackDispatchLifecycleWork,
       ),
-  );
+  ).catch((error: unknown) => {
+    if (
+      params.replyOptions?.isHeartbeat === true ||
+      !didDeliverVisiblePartialReply ||
+      isDispatchOperationAborted()
+    ) {
+      throw error;
+    }
+    failDispatchReplyOperation(error);
+    return buildTerminalAgentRunFailureReplyPayload({
+      visibleReplyDelivered: true,
+      sessionCtx: ctx,
+      cfg: replyConfig,
+    });
+  });
   const sessionMetadataChanges = takeCommandSessionMetadataChanges(ctx);
   notifySessionMetadataChanges(sessionMetadataChanges);
   const finalDispatchAcquisition = await state.ensureDispatchReplyOperation("dispatch");

src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts

@@ -97,4 +97,44 @@ describe("dispatchReplyFromConfig terminal visible admission recovery", () => {
     expect(replyResolver).toHaveBeenCalledTimes(1);
     expect(dispatchParams.dispatcher.sendFinalReply).toHaveBeenCalledTimes(1);
   });
+
+  it("records a failed reply operation when recovering a visible partial", async () => {
+    const resolverError = new Error("provider failed after partial");
+    let replyOperation: ReturnType<typeof createReplyOperation> | undefined;
+    const replyResolver: NonNullable<DispatchFromConfigParams["replyResolver"]> = async (
+      _ctx,
+      options,
+    ) => {
+      if (!options) {
+        throw new Error("reply options required for partial recovery");
+      }
+      replyOperation = options.replyOperation;
+      await options.onPartialReply?.({ text: "partial telegram reply" });
+      throw resolverError;
+    };
+    const dispatchParams = {
+      ...createVisibleDispatchParams(replyResolver),
+      replyOptions: {
+        onPartialReply: vi.fn(async () => undefined),
+      },
+    };
+
+    const result = await dispatchReplyFromConfig(dispatchParams);
+
+    expect(replyOperation?.result).toEqual({
+      kind: "failed",
+      code: "run_failed",
+      cause: resolverError,
+    });
+    expect(result).toMatchObject({
+      queuedFinal: true,
+      counts: { tool: 0, block: 0, final: 0 },
+    });
+    expect(dispatchParams.replyOptions.onPartialReply).toHaveBeenCalledWith({
+      text: "partial telegram reply",
+    });
+    expect(dispatchParams.dispatcher.sendFinalReply).toHaveBeenCalledWith(
+      expect.objectContaining({ text: expect.stringContaining("Something went wrong") }),
+    );
+  });
 });