diff --git a/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts b/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts index cc5e25399720..8ffd5694a4c5 100644 --- a/extensions/telegram/src/bot-message-dispatch.draft-failures-progress.test.ts +++ b/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 }); diff --git a/extensions/telegram/src/lane-delivery-text-deliverer.ts b/extensions/telegram/src/lane-delivery-text-deliverer.ts index 77ecfb3eb50a..dae02206edfa 100644 --- a/extensions/telegram/src/lane-delivery-text-deliverer.ts +++ b/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 => { 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) { diff --git a/src/auto-reply/reply/agent-runner-core.ts b/src/auto-reply/reply/agent-runner-core.ts index 417af90823c2..e8a64a4134d8 100644 --- a/src/auto-reply/reply/agent-runner-core.ts +++ b/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 { 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. diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 08d87626540c..8ff75d9d437f 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/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, }) diff --git a/src/auto-reply/reply/agent-runner-failure-reply.ts b/src/auto-reply/reply/agent-runner-failure-reply.ts index b6519697838c..c5c358bf3ed0 100644 --- a/src/auto-reply/reply/agent-runner-failure-reply.ts +++ b/src/auto-reply/reply/agent-runner-failure-reply.ts @@ -471,18 +471,24 @@ export function markAgentRunFailureReplyPayload(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, + }), }); } diff --git a/src/auto-reply/reply/agent-runner-run.ts b/src/auto-reply/reply/agent-runner-run.ts index 447643b1ee66..d7052d1cd78f 100644 --- a/src/auto-reply/reply/agent-runner-run.ts +++ b/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>[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, diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index a7f8a746891d..09bce8416e8f 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/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", () => { diff --git a/src/auto-reply/reply/dispatch-from-config.execute.ts b/src/auto-reply/reply/dispatch-from-config.execute.ts index 8365af58fc98..0aa1aeb8396c 100644 --- a/src/auto-reply/reply/dispatch-from-config.execute.ts +++ b/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"); diff --git a/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts b/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts index 03636349c685..93d86884ba00 100644 --- a/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts +++ b/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 | undefined; + const replyResolver: NonNullable = 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") }), + ); + }); });