diff --git a/extensions/telegram/src/bot-message-dispatch-delivery.ts b/extensions/telegram/src/bot-message-dispatch-delivery.ts index 13f928ece74f..a2f68fbe99ce 100644 --- a/extensions/telegram/src/bot-message-dispatch-delivery.ts +++ b/extensions/telegram/src/bot-message-dispatch-delivery.ts @@ -6,6 +6,7 @@ import { resolveTranscriptBackedChannelFinalText, } from "openclaw/plugin-sdk/channel-outbound"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { createSubsystemLogger } from "openclaw/plugin-sdk/logging-core"; import type { ReplyPayload } from "openclaw/plugin-sdk/reply-payload"; import { isSingleUseReplyToMode } from "openclaw/plugin-sdk/reply-reference"; import { logVerbose } from "openclaw/plugin-sdk/runtime-env"; @@ -73,6 +74,8 @@ type TelegramSendPayloadOptions = { bindPendingFinalDelivery?: (payload: T) => T; }; +const deliveryDowngradeLog = createSubsystemLogger("gateway/channels/telegram").child("delivery"); + const projectPayloadForDelivery = (turn: Turn, payload: ReplyPayload): ReplyPayload | undefined => projectOutboundPayloadPlanForDelivery( createOutboundPayloadPlan([payload], { @@ -303,6 +306,20 @@ export async function sendPayload( await projectionSequence.fail(); return false; } + // Any other status silently downgrades a final from the durable custody + // funnel to the direct funnel; that downgrade must be attributable when a + // delivery later settles ambiguous (spurious "couldn't confirm" notices). + deliveryDowngradeLog.warn( + `durable final delivery not handled (status=${durable.status}${ + "reason" in durable && durable.reason ? ` reason=${durable.reason}` : "" + }); falling back to direct send`, + ); + } else if (options?.durable) { + deliveryDowngradeLog.warn( + `durable final delivery skipped (${ + durableDelivery ? "prompt context sequence not fresh" : "durable deliverer unavailable" + }); falling back to direct send`, + ); } try { const transcriptMirror = createTranscriptMirror(turn); diff --git a/src/channels/turn/lifecycle.ts b/src/channels/turn/lifecycle.ts index 3973298077a1..2bd6a7682c74 100644 --- a/src/channels/turn/lifecycle.ts +++ b/src/channels/turn/lifecycle.ts @@ -19,6 +19,7 @@ import { import { settlePendingFinalDelivery } from "../../infra/outbound/delivery-completion.js"; import { createMessageSentEmitter } from "../../infra/outbound/message-sent-hook.js"; import { summarizeOutboundPayloadForTransport } from "../../infra/outbound/payloads.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; import { getGlobalHookRunner } from "../../plugins/hook-runner-global.js"; import { resolveMessageReceiptPrimaryId } from "../message/receipt.js"; import { createChannelReplyPipeline } from "../message/reply-pipeline.js"; @@ -53,6 +54,8 @@ import type { PreparedChannelTurn, } from "./types.js"; +const settleLog = createSubsystemLogger("channels/turn/delivery-custody"); + type RoutedAssembledChannelTurn = Omit< AssembledChannelTurn, "delivery" | "dispatchReplyWithBufferedBlockDispatcher" @@ -245,6 +248,11 @@ async function settleFailedPendingFinalDelivery( } else if (isPlatformMessageNotDispatchedError(error)) { await settlePendingFinalDelivery(completion, "prepared", ["queued", "unknown"]); } else { + // Unknown custody can surface later as a recovery notice; without this line + // the causing error is invisible and the notice looks spontaneous. + settleLog.warn( + `pending final delivery settled unknown after send error: intent=${completion.intentId} ${formatErrorMessage(error)}`, + ); await settlePendingFinalDelivery(completion, "unknown", ["queued", "unknown"]); } } diff --git a/src/channels/turn/run-channel-turn.finalization-custody.test.ts b/src/channels/turn/run-channel-turn.finalization-custody.test.ts new file mode 100644 index 000000000000..7ba63045c1e5 --- /dev/null +++ b/src/channels/turn/run-channel-turn.finalization-custody.test.ts @@ -0,0 +1,125 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { setReplyPayloadMetadata, type ReplyPayload } from "../../auto-reply/reply-payload.js"; +import type { DispatchReplyWithDispatcher } from "../../auto-reply/reply/provider-dispatcher.types.js"; +import type { FinalizedMsgContext } from "../../auto-reply/templating.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import { createChannelPartialDeliveryError } from "./delivery-result.js"; +import { dispatchRoutedChannelTurn } from "./lifecycle.js"; + +const dispatchReplyWithRoutedChannelDispatcherCore = vi.hoisted(() => vi.fn()); +const getGlobalHookRunner = vi.hoisted(() => vi.fn()); +const settlePendingFinalDelivery = vi.hoisted(() => + vi.fn(async (_completion: unknown, state: string) => ({ state })), +); + +vi.mock("../../auto-reply/dispatch.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + dispatchInboundMessageWithRoutedChannelDispatcher: dispatchReplyWithRoutedChannelDispatcherCore, + }; +}); + +vi.mock("../../plugins/hook-runner-global.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, getGlobalHookRunner }; +}); + +vi.mock("../../config/sessions/transcript.js", () => ({ + readRecentUserAssistantTextForSession: vi.fn(async () => []), +})); + +vi.mock("../../infra/outbound/delivery-completion.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { ...actual, settlePendingFinalDelivery }; +}); + +const cfg: OpenClawConfig = {}; + +function createCtx(overrides: Partial = {}): FinalizedMsgContext { + return { + Body: "hello", + RawBody: "hello", + CommandBody: "hello", + CommandAuthorized: false, + From: "sender", + To: "target", + SessionKey: "agent:main:test:peer", + Provider: "test", + Surface: "test", + ...overrides, + }; +} + +describe("deferred finalization custody after a visible send", () => { + const completion = { + deliveryId: "delivery-final", + intentId: "intent-final", + sessionId: "session-final", + sessionKey: "agent:main:telegram:peer", + storePath: "/tmp/sessions.json", + }; + + beforeEach(() => { + vi.clearAllMocks(); + getGlobalHookRunner.mockReturnValue(null); + settlePendingFinalDelivery.mockImplementation(async (_completion, state: string) => ({ + state, + })); + }); + + const run = (finalization: Promise<{ visibleReplySent: boolean; messageIds?: string[] }>) => { + const sourcePayload = setReplyPayloadMetadata( + { text: "reply" }, + { pendingFinalDeliveryCompletion: completion }, + ); + const dispatch: DispatchReplyWithDispatcher = async (params) => { + await params.dispatcherOptions.deliver(sourcePayload, { kind: "final" }); + return { queuedFinal: true, counts: { tool: 0, block: 0, final: 1 } }; + }; + dispatchReplyWithRoutedChannelDispatcherCore.mockImplementationOnce(dispatch); + return dispatchRoutedChannelTurn({ + cfg, + channel: "telegram", + accountId: "acct", + route: { agentId: "main", sessionKey: completion.sessionKey }, + ctxPayload: createCtx({ Surface: "telegram", OriginatingTo: "chat-1" }), + delivery: { + deliver: async (_payload: ReplyPayload) => ({ + visibleReplySent: true, + finalization, + }), + }, + }); + }; + + it("settles delivered when deferred finalization resolves", async () => { + await run(Promise.resolve({ visibleReplySent: true, messageIds: ["56067"] })); + + expect(settlePendingFinalDelivery).toHaveBeenLastCalledWith( + { kind: "pending-final", ...completion }, + "delivered", + ); + }); + + it("keeps unknown custody when finalization rejects with a true partial", async () => { + // A partial proves something was visible, not that everything was; the + // remainder may be lost. Unknown custody plus the recovery notice ("ask + // for any missing remainder") is the designed outcome. Channels must not + // reject content-complete deliveries this way — cosmetic post-content + // failures stay channel-side (see telegram progress-window guards). + const partial = createChannelPartialDeliveryError(new Error("chunk 2 send failed"), { + visibleReplySent: true, + messageIds: ["56067"], + }); + + await expect(run(Promise.reject(partial))).rejects.toBeDefined(); + + expect(settlePendingFinalDelivery).toHaveBeenLastCalledWith( + { kind: "pending-final", ...completion }, + "unknown", + ["queued", "unknown"], + ); + }); +});