fix(telegram): preserve visible draft recovery (#120626)

* fix(telegram): preserve visible draft recovery * fix(telegram): await visible draft send * test(telegram): use const in draft recovery test * test(telegram): add live partial failure proof * test(telegram): register live recovery coverage * ci(qa): support mock Telegram proof scenarios * test(telegram): isolate live recovery fallback * test(telegram): assert live recovery behavior * fix(agents): join partial reply delivery * test(telegram): keep settlement proof at core boundary
2026-08-27 04:47:03 -06:00 · 2026-08-09 02:54:38 -07:00
parent 917fd92686
commit 0df1a89e3a
16 changed files with 551 additions and 36 deletions
@@ -19,6 +19,14 @@ on:
        required: false
        default: telegram-status-command
        type: string
+      provider_mode:
+        description: QA provider mode
+        required: false
+        default: live-frontier
+        type: choice
+        options:
+          - live-frontier
+          - mock-openai
      crabbox_provider:
        description: Crabbox provider for the desktop transcript capture
        required: false
@@ -98,6 +106,7 @@ jobs:
      crabbox_provider: ${{ steps.resolve.outputs.crabbox_provider }}
      lease_id: ${{ steps.resolve.outputs.lease_id }}
      pr_number: ${{ steps.resolve.outputs.pr_number }}
+      provider_mode: ${{ steps.resolve.outputs.provider_mode }}
      reaction_id: ${{ steps.add_reaction.outputs.reaction_id }}
      request_source: ${{ steps.resolve.outputs.request_source }}
      scenario: ${{ steps.resolve.outputs.scenario }}
@@ -117,10 +126,16 @@ jobs:

            if (eventName === "workflow_dispatch") {
              const inputs = context.payload.inputs ?? {};
+              const providerMode = inputs.provider_mode || "live-frontier";
+              if (!["live-frontier", "mock-openai"].includes(providerMode)) {
+                core.setFailed(`Unsupported provider mode for Mantis Telegram: ${providerMode}`);
+                return;
+              }
              setOutput("should_run", "true");
              setOutput("candidate_ref", inputs.candidate_ref || "main");
              setOutput("pr_number", inputs.pr_number || "");
              setOutput("scenario", inputs.scenario || "telegram-status-command");
+              setOutput("provider_mode", providerMode);
              setOutput("crabbox_provider", inputs.crabbox_provider || "aws");
              setOutput("lease_id", inputs.crabbox_lease_id || "");
              setOutput("request_source", "workflow_dispatch");
@@ -158,6 +173,7 @@ jobs:
              setOutput("candidate_ref", "");
              setOutput("pr_number", "");
              setOutput("scenario", "");
+              setOutput("provider_mode", "");
              setOutput("crabbox_provider", "");
              setOutput("lease_id", "");
              setOutput("request_source", "unsupported_issue_comment");
@@ -172,6 +188,7 @@ jobs:
            });
            const candidateMatch = body.match(/(?:candidate|head)[\s:=]+([^\s`]+)/i);
            const scenarioMatch = body.match(/(?:scenario|scenarios)[\s:=]+([^\s`]+)/i);
+            const providerModeMatch = body.match(/(?:provider_mode|provider-mode)[\s:=]+([^\s`]+)/i);
            const providerMatch = body.match(/(?:provider|crabbox_provider)[\s:=]+([^\s`]+)/i);
            const leaseMatch = body.match(/(?:lease|lease_id|crabbox_lease_id)[\s:=]+([^\s`]+)/i);
            const rawCandidate = candidateMatch?.[1];
@@ -184,11 +201,17 @@ jobs:
              core.setFailed(`Unsupported Crabbox provider for Mantis Telegram: ${provider}`);
              return;
            }
+            const providerMode = providerModeMatch?.[1] || "live-frontier";
+            if (!["live-frontier", "mock-openai"].includes(providerMode)) {
+              core.setFailed(`Unsupported provider mode for Mantis Telegram: ${providerMode}`);
+              return;
+            }

            setOutput("should_run", "true");
            setOutput("candidate_ref", candidate);
            setOutput("pr_number", String(issue.number));
            setOutput("scenario", scenarioMatch?.[1] || "telegram-status-command");
+            setOutput("provider_mode", providerMode);
            setOutput("crabbox_provider", provider);
            setOutput("lease_id", leaseMatch?.[1] || "");
            setOutput("request_source", "issue_comment");
@@ -412,6 +435,7 @@ jobs:
          CRABBOX_CAPACITY_REGIONS: ${{ env.CRABBOX_CAPACITY_REGIONS }}
          CRABBOX_LEASE_ID: ${{ needs.resolve_request.outputs.lease_id }}
          CRABBOX_PROVIDER: ${{ needs.resolve_request.outputs.crabbox_provider }}
+          PROVIDER_MODE: ${{ needs.resolve_request.outputs.provider_mode }}
          SCENARIO_INPUT: ${{ needs.resolve_request.outputs.scenario }}
          CANDIDATE_SHA: ${{ needs.validate_ref.outputs.candidate_revision }}
        shell: bash
@@ -430,7 +454,6 @@ jobs:
          CRABBOX_COORDINATOR_TOKEN="${CRABBOX_COORDINATOR_TOKEN:-${OPENCLAW_QA_MANTIS_CRABBOX_COORDINATOR_TOKEN:-}}"
          export CRABBOX_COORDINATOR CRABBOX_COORDINATOR_TOKEN

-          require_var OPENAI_API_KEY
          require_var OPENCLAW_QA_CONVEX_SITE_URL
          require_var OPENCLAW_QA_CONVEX_SECRET_CI
          require_var CRABBOX_COORDINATOR_TOKEN
@@ -439,7 +462,29 @@ jobs:
          output_rel=".artifacts/qa-e2e/mantis/telegram-live"
          root="$candidate_repo/$output_rel"
          echo "output_dir=${root}" >> "$GITHUB_OUTPUT"
-          model="${OPENCLAW_CI_OPENAI_MODEL:-openai/gpt-5.6-luna}"
+
+          qa_args=(
+            --repo-root "$candidate_repo"
+            --output-dir "$output_rel"
+            --provider-mode "$PROVIDER_MODE"
+          )
+          case "$PROVIDER_MODE" in
+            live-frontier)
+              require_var OPENAI_API_KEY
+              model="${OPENCLAW_CI_OPENAI_MODEL:-openai/gpt-5.6-luna}"
+              qa_args+=(--model "$model" --alt-model "$model" --fast)
+              ;;
+            mock-openai) ;;
+            *)
+              echo "Unsupported provider mode for Mantis Telegram: ${PROVIDER_MODE}" >&2
+              exit 1
+              ;;
+          esac
+          qa_args+=(
+            --credential-source convex
+            --credential-role ci
+            --allow-failures
+          )

          scenario_args=()
          if [[ -n "${SCENARIO_INPUT// }" ]]; then
@@ -453,17 +498,7 @@ jobs:
          fi

          set +e
-          pnpm --dir "$candidate_repo" openclaw qa telegram \
-            --repo-root "$candidate_repo" \
-            --output-dir "$output_rel" \
-            --provider-mode live-frontier \
-            --model "$model" \
-            --alt-model "$model" \
-            --fast \
-            --credential-source convex \
-            --credential-role ci \
-            --allow-failures \
-            "${scenario_args[@]}"
+          pnpm --dir "$candidate_repo" openclaw qa telegram "${qa_args[@]}" "${scenario_args[@]}"
          telegram_exit=$?
          set -e

@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
 import type { StreamEvent } from "./mock-openai-contracts.js";
 import {
  buildAssistantEvents,
+  buildPartialFailureEvents,
  buildAssistantThenToolCallEvents,
  buildFailedResponseEvents,
  buildReasoningAndAssistantEvents,
@@ -31,13 +32,35 @@ function readOutputItemSlots(events: StreamEvent[]) {

 describe("mock OpenAI Responses output item slots", () => {
  it("emits the provider no-details failure used by repeated-request recovery QA", () => {
-    expect(buildFailedResponseEvents()).toEqual([
+    const events = buildFailedResponseEvents();
+    expect(events).toEqual([
      expect.objectContaining({ type: "response.created" }),
      expect.objectContaining({
        type: "response.failed",
        response: expect.not.objectContaining({ error: expect.anything() }),
      }),
    ]);
+    expect(events.some((event) => event.type === "response.output_text.delta")).toBe(false);
+  });
+
+  it("emits an unfinished assistant delta before the failed response", () => {
+    const marker = "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE";
+    const events = buildPartialFailureEvents(marker);
+
+    expect(events.map((event) => event.type)).toEqual([
+      "response.created",
+      "response.output_item.added",
+      "response.output_text.delta",
+      "response.failed",
+    ]);
+    expect(events[1]).toMatchObject({
+      item: { type: "message", role: "assistant", status: "in_progress" },
+    });
+    expect(events[2]).toMatchObject({
+      type: "response.output_text.delta",
+      delta: marker,
+    });
+    expect(events.some((event) => event.type === "response.output_item.done")).toBe(false);
  });

  it("indexes preview deltas and the final answer on the same assistant slot", () => {
@@ -21,6 +21,40 @@ export function buildFailedResponseEvents(): StreamEvent[] {
  ];
 }

+export function buildPartialFailureEvents(partialText: string): StreamEvent[] {
+  const responseId = "resp_qa_partial_failed_1";
+  const itemId = "msg_qa_partial_failed_1";
+  return [
+    { type: "response.created", response: { id: responseId } },
+    {
+      type: "response.output_item.added",
+      output_index: 0,
+      item: {
+        type: "message",
+        id: itemId,
+        role: "assistant",
+        phase: "final_answer",
+        content: [],
+        status: "in_progress",
+      },
+    },
+    {
+      type: "response.output_text.delta",
+      item_id: itemId,
+      output_index: 0,
+      content_index: 0,
+      delta: partialText,
+    },
+    {
+      type: "response.failed",
+      response: {
+        id: responseId,
+        status: "failed",
+      },
+    },
+  ];
+}
+
 export function buildToolCallEvents(prompt: string): StreamEvent[] {
  const targetPath = readTargetFromPrompt(prompt);
  return buildToolCallEventsWithArgs("read", { path: targetPath });
@@ -219,6 +219,13 @@ function expectOpenAiStreamingResponsesText(server: MockServer, body: Record<str
  return expectStreamingResponsesText(server, { model: "gpt-5.6-luna", ...body });
 }

+function parseStreamingResponseEvents(body: string): StreamEvent[] {
+  return body
+    .split("\n")
+    .filter((line) => line.startsWith("data: {") && line.endsWith("}"))
+    .map((line) => JSON.parse(line.slice("data: ".length)) as StreamEvent);
+}
+
 const requireRecord = createRequireRecord("record", "expected-label-capitalized");

 function requireArray(value: unknown, label: string): unknown[] {
@@ -913,6 +920,36 @@ describe("qa mock openai server", () => {
    expect(blockContinuationBody).not.toContain('"item_id":"msg_mock_block_1"');
  });

+  it("serves Telegram visible and unsent failure directives", async () => {
+    const server = await startMockServer();
+    const visibleEvents = parseStreamingResponseEvents(
+      await expectOpenAiStreamingResponsesText(server, {
+        input: [makeUserInput("Telegram visible partial failure QA check")],
+      }),
+    );
+    const unsentEvents = parseStreamingResponseEvents(
+      await expectOpenAiStreamingResponsesText(server, {
+        input: [makeUserInput("Telegram unsent failure QA check")],
+      }),
+    );
+
+    expect(visibleEvents.map((event) => event.type)).toEqual([
+      "response.created",
+      "response.output_item.added",
+      "response.output_text.delta",
+      "response.failed",
+    ]);
+    expect(visibleEvents[2]).toMatchObject({
+      type: "response.output_text.delta",
+      delta: "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE",
+    });
+    expect(unsentEvents.map((event) => event.type)).toEqual([
+      "response.created",
+      "response.failed",
+    ]);
+    expect(unsentEvents.some((event) => event.type === "response.output_text.delta")).toBe(false);
+  });
+
  it("plans deterministic tool-progress reads from prompt paths", async () => {
    const server = await startMockServer();

@@ -144,6 +144,7 @@ import {
  buildQaLongFinalText,
  buildAssistantThenToolCallEvents,
  buildAssistantEvents,
+  buildPartialFailureEvents,
  buildReasoningOnlyEvents,
  buildReasoningAndAssistantEvents,
  buildFailedResponseEvents,
@@ -288,6 +289,9 @@ const QA_STREAMING_TOOL_PROGRESS_CONTINUATION_RE =
  /^Continue with (?:the current Matrix QA scenario|the QA scenario plan and report worked, failed, and blocked items)\.$/i;
 const QA_CODE_MODE_TARGET_MARKER = "qa-code-mode-target:";
 const QA_FAILED_TOOL_TERMINAL_RECOVERY_PROMPT_RE = /failed tool terminal recovery qa check/i;
+const QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_PROMPT_RE = /telegram visible partial failure qa check/i;
+const QA_TELEGRAM_UNSENT_FAILURE_PROMPT_RE = /telegram unsent failure qa check/i;
+const QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_MARKER = "TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE";
 // Keep each real provider request active long enough for retries to span the
 // unchanged five-minute recovery bound while remaining below first-byte timeout.
 const QA_REPEATED_REQUEST_RESPONSE_PAUSE_MS = 110_000;
@@ -868,6 +872,12 @@ async function buildResponsesPayload(
  if (QA_REPEATED_REQUEST_QUEUED_REPLY_PROMPT_RE.test(prompt)) {
    return buildAssistantEvents(QA_REPEATED_REQUEST_QUEUED_REPLY_MARKER);
  }
+  if (QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_PROMPT_RE.test(prompt)) {
+    return buildPartialFailureEvents(QA_TELEGRAM_VISIBLE_PARTIAL_FAILURE_MARKER);
+  }
+  if (QA_TELEGRAM_UNSENT_FAILURE_PROMPT_RE.test(prompt)) {
+    return buildFailedResponseEvents();
+  }
  if (QA_REPEATED_REQUEST_RECOVERY_PROMPT_RE.test(allInputText)) {
    return buildFailedResponseEvents();
  }
@@ -158,7 +158,16 @@ export async function runTelegramDispatchTurn(params: {
                    const queued = params.draft.enqueueEvent(async () => {
                      await params.draft.ingestDraftLaneSegments(payload);
                    });
-                    return queued.then(() => false);
+                    // Queue settlement records draft intent; a numeric provider message ID
+                    // proves operator visibility for terminal recovery.
+                    return queued.then(async () => {
+                      const answerStream = params.draft.answerLane.stream;
+                      await answerStream?.waitForInFlight();
+                      const providerMessageId = answerStream?.messageId();
+                      return (
+                        typeof providerMessageId === "number" && Number.isFinite(providerMessageId)
+                      );
+                    });
                  }
                : undefined,
            onBlockReplyQueued: params.draft.answerLane.stream
@@ -18,6 +18,7 @@ import {
  setupDraftStreams,
  telegramProgressPreview,
 } from "./bot-message-dispatch.test-harness.js";
+import { createTestDraftStream } from "./draft-stream.test-helpers.js";

 const draftWarn = vi.hoisted(() => vi.fn());

@@ -80,36 +81,72 @@ describeTelegramDispatch("dispatchTelegramMessage draft-failures-progress", () =
  });

  it.each([
-    { label: "direct chat", createSessionPayload: createDirectSessionPayload },
+    {
+      label: "direct chat",
+      createMessageContext: () =>
+        createContext({
+          ctxPayload: createDirectSessionPayload(),
+        }),
+    },
    {
      label: "group chat",
-      createSessionPayload: () => ({
-        ...createDirectSessionPayload(),
-        SessionKey: "agent:test:telegram:group:-100123",
-        ChatType: "group" as const,
-      }),
+      createMessageContext: () =>
+        createContext({
+          chatId: -100123,
+          isGroup: true,
+          ctxPayload: {
+            ...createDirectSessionPayload(),
+            SessionKey: "agent:test:telegram:group:-100123",
+            ChatType: "group",
+          },
+          primaryCtx: {
+            ...createContext().primaryCtx,
+            message: {
+              chat: { id: -100123, type: "supergroup", title: "Test group" },
+              date: 0,
+              message_id: 456,
+            },
+          },
+          msg: {
+            chat: { id: -100123, type: "supergroup", title: "Test group" },
+            date: 0,
+            message_id: 456,
+            message_thread_id: undefined,
+          },
+          threadSpec: { id: undefined, scope: "none" },
+          replyThreadId: undefined,
+        }),
    },
  ])(
    "finalizes the default streamed draft in place after an unexpected reply failure in a $label",
-    async ({ createSessionPayload }) => {
-      const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
+    async ({ createMessageContext }) => {
+      const answerDraftStream = createTestDraftStream({
+        onWaitForInFlight: () => answerDraftStream.setMessageId(2001),
+      });
+      const reasoningDraftStream = createTestDraftStream();
+      createTelegramDraftStream
+        .mockImplementationOnce(() => answerDraftStream)
+        .mockImplementationOnce(() => reasoningDraftStream);
+      let partialAccepted: boolean | void = undefined;
      dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async (params) => {
        expect(params.replyOptions?.disableBlockStreaming).toBe(true);
        return await dispatchReplyWithBufferedBlockDispatcherRuntime({
          ...params,
          replyResolver: async (_ctx, opts) => {
-            await opts?.onPartialReply?.({ text: "partial answer" });
+            partialAccepted = await opts?.onPartialReply?.({ text: "partial answer" });
            throw new Error("unexpected model failure");
          },
        });
      });

      await dispatchWithContext({
-        context: createContext({ ctxPayload: createSessionPayload() }),
+        context: createMessageContext(),
        streamMode: "partial",
        telegramCfg: { streaming: { mode: "partial" } },
      });

+      expect(partialAccepted).toBeUndefined();
+      expect(answerDraftStream.waitForInFlight).toHaveBeenCalledOnce();
      expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "partial answer");
      expect(answerDraftStream.update).toHaveBeenCalledTimes(2);
      expect(answerDraftStream.update).toHaveBeenLastCalledWith(
@@ -122,6 +159,35 @@ describeTelegramDispatch("dispatchTelegramMessage draft-failures-progress", () =
    },
  );

+  it("clears a pending partial and sends one fallback after an unexpected reply failure", async () => {
+    const { answerDraftStream } = setupDraftStreams();
+    let partialAccepted: boolean | void = undefined;
+    dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async (params) => {
+      return await dispatchReplyWithBufferedBlockDispatcherRuntime({
+        ...params,
+        replyResolver: async (_ctx, opts) => {
+          partialAccepted = await opts?.onPartialReply?.({ text: "partial answer" });
+          throw new Error("unexpected model failure");
+        },
+      });
+    });
+
+    await dispatchWithContext({
+      context: createContext({ ctxPayload: createDirectSessionPayload() }),
+      streamMode: "partial",
+      telegramCfg: { streaming: { mode: "partial" } },
+    });
+
+    expect(partialAccepted).toBe(false);
+    expect(answerDraftStream.update).toHaveBeenCalledOnce();
+    expect(answerDraftStream.update).toHaveBeenCalledWith("partial answer");
+    expect(answerDraftStream.clear).toHaveBeenCalledOnce();
+    expect(deliverReplies).toHaveBeenCalledOnce();
+    expectDeliveredReply(0, {
+      text: "Something went wrong while processing your request. Please try again.",
+    });
+  });
+
  it("returns retryable when dispatch fails after partial output and the fallback is not delivered", async () => {
    deliverReplies.mockResolvedValueOnce({ delivered: true });
    deliverReplies.mockResolvedValueOnce({ delivered: false });
@@ -187,7 +187,7 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => {

    expect(answerDraftStream.clear).not.toHaveBeenCalled();
    expect(answerDraftStream.update).toHaveBeenCalledWith("Photo");
-    expect(partialAccepted).toBe(false);
+    expect(partialAccepted).toBe(true);
    expectDeliverRepliesParams({ mediaMaxBytes });
    expectDeliveredReply(0, { text: undefined, mediaUrl: "https://example.com/a.png" });
    expect(emitTelegramMessageSentHooks).toHaveBeenCalledTimes(1);
@@ -11,6 +11,7 @@ type TestDraftStream = {
  updateLazy: ReturnType<typeof vi.fn<(resolveText: () => string | undefined) => void>>;
  updatePreview: ReturnType<typeof vi.fn<(preview: TelegramDraftPreview) => void>>;
  flush: ReturnType<typeof vi.fn<() => Promise<void>>>;
+  waitForInFlight: ReturnType<typeof vi.fn<() => Promise<void>>>;
  messageId: ReturnType<typeof vi.fn<() => number | undefined>>;
  lastDeliveredText: ReturnType<typeof vi.fn<() => string>>;
  currentMessageSnapshot: ReturnType<typeof vi.fn<() => TelegramDraftMessageSnapshot | undefined>>;
@@ -31,6 +32,7 @@ type TestDraftStream = {
 export function createTestDraftStream(params?: {
  messageId?: number;
  onUpdate?: (text: string) => void;
+  onWaitForInFlight?: () => void | Promise<void>;
  onStop?: () => void | Promise<void>;
  onDiscard?: () => void | Promise<void>;
  clearMessageIdOnForceNew?: boolean;
@@ -64,6 +66,9 @@ export function createTestDraftStream(params?: {
      params?.onUpdate?.(preview.text);
    }),
    flush: vi.fn().mockResolvedValue(undefined),
+    waitForInFlight: vi.fn().mockImplementation(async () => {
+      await params?.onWaitForInFlight?.();
+    }),
    messageId: vi.fn().mockImplementation(() => messageId),
    lastDeliveredText: vi.fn().mockImplementation(() => lastDeliveredText),
    currentMessageSnapshot: vi
@@ -142,6 +147,7 @@ export function createSequencedTestDraftStream(startMessageId = 1001): TestDraft
      lastDeliveredText = preview.text.trimEnd();
    }),
    flush: vi.fn().mockResolvedValue(undefined),
+    waitForInFlight: vi.fn().mockResolvedValue(undefined),
    messageId: vi.fn().mockImplementation(() => activeMessageId),
    lastDeliveredText: vi.fn().mockImplementation(() => lastDeliveredText),
    currentMessageSnapshot: vi
@@ -194,7 +194,7 @@ describe("createTelegramDraftStream", () => {
    const stream = createDraftStream(api, { validateProviderMessage });

    stream.update("First preview");
-    await expect(stream.flush()).rejects.toBe(validationError);
+    await expect(stream.waitForInFlight()).rejects.toBe(validationError);
    stream.update("Second preview");
    await expect(stream.flush()).rejects.toBe(validationError);

@@ -65,6 +65,7 @@ export type TelegramDraftStream = {
  updateLazy: (resolveText: () => string | undefined) => void;
  updatePreview: (preview: TelegramDraftPreview) => void;
  flush: () => Promise<void>;
+  waitForInFlight: () => Promise<void>;
  messageId: () => number | undefined;
  lastDeliveredText?: () => string;
  currentMessageSnapshot?: () => TelegramDraftMessageSnapshot | undefined;
@@ -758,8 +759,12 @@ export function createTelegramDraftStream(params: {
      throw terminalDeliveryError;
    }
  };
-  const flush = async () => {
+  const waitForInFlight = async () => {
    await loop.waitForInFlight();
+    throwTerminalDeliveryError();
+  };
+  const flush = async () => {
+    await waitForInFlight();
    if (!streamState.stopped) {
      await loop.flush();
    }
@@ -1048,6 +1053,7 @@ export function createTelegramDraftStream(params: {
    updateLazy: requestLazyDraftUpdate,
    updatePreview,
    flush,
+    waitForInFlight,
    messageId: () => streamMessageId,
    lastDeliveredText: () => lastDeliveredText,
    currentMessageSnapshot: () => streamMessageSnapshot,
@@ -19,10 +19,11 @@ describe("Telegram outbound web app presentation", () => {
  it.each(["-1001234567890", "@channelname"])(
    "falls back to a link for non-DM target %s",
    async (to) => {
+      const payload = { text: "Open app:" };
      const rendered = await telegramOutbound.renderPresentation?.({
-        payload: { text: "Open app:" },
+        payload,
        presentation: webAppPresentation,
-        ctx: { to } as never,
+        ctx: { cfg: {}, to, text: payload.text, payload },
      });

      expect(rendered).toEqual({
@@ -0,0 +1,107 @@
+title: Telegram partial failure recovery
+
+scenario:
+  id: telegram-partial-failure-recovery
+  surface: channels
+  category: channels.conversation-routing-and-delivery
+  coverage:
+    primary:
+      - channels.streaming-final-reply
+  regressionRefs:
+    - openclaw/openclaw#120626
+  objective: Verify Telegram recovers provider failures in one logical message whether or not a streaming preview became visible.
+  successCriteria:
+    - A visible partial is retained and the terminal recovery edits the same Telegram message.
+    - A failure before any visible partial produces one fallback Telegram message.
+  codeRefs:
+    - extensions/telegram/src/bot-message-dispatch-turn.ts
+    - extensions/telegram/src/draft-stream.ts
+    - extensions/qa-lab/src/providers/mock-openai/server.ts
+    - extensions/qa-lab/src/live-transports/telegram/adapter.runtime.ts
+  gatewayConfigPatch:
+    agents:
+      defaults:
+        model:
+          fallbacks: []
+      entries:
+        qa:
+          model:
+            fallbacks: []
+    channels:
+      telegram:
+        streaming:
+          mode: partial
+  execution:
+    kind: flow
+    channel: telegram
+    summary: Inject provider failures before and after Telegram accepts a streaming preview, then verify one logical recovery message remains.
+    config:
+      requiredProviderMode: mock-openai
+      partialMarker: TELEGRAM-VISIBLE-PARTIAL-BEFORE-FAILURE
+
+flow:
+  steps:
+    - name: preserves a visible partial in one recovered message
+      actions:
+        - assert:
+            expr: env.providerMode === config.requiredProviderMode
+            message: this Telegram recovery scenario requires mock-openai
+        - call: waitForGatewayHealthy
+          args: [{ ref: env }, 60000]
+        - call: waitForTransportReady
+          args: [{ ref: env }, 60000]
+        - resetTransport: true
+        - set: visibleOutboundStart
+          value:
+            expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').length"
+        - sendInbound:
+            conversation: { id: telegram-partial-failure-room, kind: group }
+            senderId: qa-telegram-recovery-operator
+            senderName: Telegram Recovery Operator
+            text: "@openclaw Telegram visible partial failure QA check"
+        - waitForOutbound:
+            conversation: { id: telegram-partial-failure-room, kind: group }
+            sinceIndex: { ref: visibleOutboundStart }
+            textIncludes: { ref: config.partialMarker }
+            timeoutMs: 75000
+          saveAs: visiblePartial
+        - set: visiblePartialText
+          value:
+            expr: "visiblePartial.text"
+        - call: sleep
+          args: [4000]
+        - set: visibleMessages
+          value:
+            expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').slice(visibleOutboundStart)"
+        - assert:
+            expr: "visibleMessages.length === 1 && visiblePartialText.includes(config.partialMarker) && visibleMessages[0].text !== visiblePartialText && visibleMessages[0].text.trim().length > 0"
+            message:
+              expr: "`expected one Telegram message edited away from the visible partial; saw ${visibleMessages.length}: initial=${visiblePartialText} final=${visibleMessages.map((message) => message.text).join(' | ')}`"
+      detailsExpr: "JSON.stringify({ count: visibleMessages.length, initialText: visiblePartialText, finalText: visibleMessages.map((message) => message.text).join(' | ') })"
+
+    - name: sends one fallback when no partial became visible
+      actions:
+        - resetTransport: true
+        - set: unsentOutboundStart
+          value:
+            expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').length"
+        - sendInbound:
+            conversation: { id: telegram-partial-failure-room, kind: group }
+            senderId: qa-telegram-recovery-operator
+            senderName: Telegram Recovery Operator
+            text: "@openclaw Telegram unsent failure QA check"
+        - waitForOutbound:
+            conversation: { id: telegram-partial-failure-room, kind: group }
+            sinceIndex: { ref: unsentOutboundStart }
+            timeoutMs: 75000
+          saveAs: unsentReply
+        - call: sleep
+          args: [4000]
+        - set: unsentMessages
+          value:
+            expr: "getTransportSnapshot().messages.filter((message) => message.direction === 'outbound').slice(unsentOutboundStart)"
+        - assert:
+            expr: "unsentMessages.length === 1 && unsentReply.text.trim().length > 0 && unsentMessages[0].text.trim().length > 0 && !unsentMessages[0].text.includes(config.partialMarker)"
+            message:
+              expr: "`expected one non-empty Telegram fallback without partial text; saw ${unsentMessages.length}: ${unsentMessages.map((message) => message.text).join(' | ')}`"
+      detailsExpr: "JSON.stringify({ count: unsentMessages.length, text: unsentMessages.map((message) => message.text).join(' | ') })"
@@ -12,6 +12,7 @@ vi.mock("./attempt-stream-settle.js", () => ({
  settleEmbeddedAttemptStream: mocks.settleStream,
 }));

+import { createSubscribedSessionHarness } from "../../embedded-agent-subscribe.e2e-harness.js";
 import { SessionManager } from "../../sessions/index.js";
 import { finalizeEmbeddedAttemptStreamPhase } from "./attempt-stream-finalize.js";

@@ -102,6 +103,66 @@ beforeEach(() => {
 });

 describe("finalizeEmbeddedAttemptStreamPhase", () => {
+  it("does not settle a provider failure before partial presentation finishes", async () => {
+    let resolvePartial: (() => void) | undefined;
+    const onPartialReply = vi.fn(
+      () =>
+        new Promise<void>((resolve) => {
+          resolvePartial = resolve;
+        }),
+    );
+    const { emit, subscription } = createSubscribedSessionHarness({
+      runId: "run-partial-provider-failure",
+      onBeforeTerminalDelivery: async () => undefined,
+      onPartialReply,
+    });
+    const failedAssistant = {
+      role: "assistant",
+      content: [{ type: "text", text: "partial answer" }],
+      stopReason: "error",
+      errorMessage: "provider failed after partial",
+      provider: "test-provider",
+      model: "test-model",
+    };
+    emit({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: { type: "text_delta", delta: "partial answer" },
+    });
+    emit({ type: "message_end", message: failedAssistant });
+    emit({ type: "agent_end", messages: [failedAssistant], willRetry: false });
+
+    const fixture = createFixture({
+      waitForPendingEvents: subscription.waitForPendingEvents,
+      getBeforeAgentFinalizeRevisionReason: () => undefined,
+    });
+    mocks.settleStream.mockResolvedValue({
+      promptError: new Error("provider failed after partial"),
+      promptErrorSource: "prompt",
+      timedOutDuringCompaction: false,
+      compactionOccurredThisAttempt: false,
+      messagesSnapshot: [failedAssistant],
+      sessionIdUsed: "session-1",
+      lastAssistant: failedAssistant,
+      currentAttemptAssistant: failedAssistant,
+      currentAttemptCompletedAssistant: failedAssistant,
+      attemptUsage: undefined,
+      cacheBreak: null,
+      lastCallUsage: undefined,
+      promptCache: undefined,
+    });
+    mocks.completeAfterTurn.mockResolvedValue({ sessionIdUsed: "session-1" });
+
+    const finalize = finalizeEmbeddedAttemptStreamPhase(fixture.input);
+    await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce());
+    await Promise.resolve();
+    expect(mocks.settleStream).not.toHaveBeenCalled();
+
+    resolvePartial?.();
+    await finalize;
+    expect(mocks.settleStream).toHaveBeenCalledOnce();
+  });
+
  it("rewinds the exact rejected branch before the hidden retry can choose NO_REPLY", async () => {
    const sessionManager = SessionManager.inMemory();
    const promptId = sessionManager.appendMessage({
@@ -0,0 +1,99 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const logger = vi.hoisted(() => ({
+  debug: vi.fn(),
+  error: vi.fn(),
+  fatal: vi.fn(),
+  info: vi.fn(),
+  isEnabled: vi.fn(() => false),
+  trace: vi.fn(),
+  warn: vi.fn(),
+}));
+
+vi.mock("../logging/subsystem.js", () => ({
+  createSubsystemLogger: () => logger,
+}));
+
+import { createSubscribedSessionHarness } from "./embedded-agent-subscribe.e2e-harness.js";
+
+function emitPartialThenProviderFailure(emit: (event: unknown) => void): void {
+  emit({
+    type: "message_update",
+    message: { role: "assistant" },
+    assistantMessageEvent: { type: "text_delta", delta: "partial answer" },
+  });
+  const failedAssistant = {
+    role: "assistant",
+    content: [{ type: "text", text: "partial answer" }],
+    stopReason: "error",
+    errorMessage: "provider failed after partial",
+    provider: "test-provider",
+    model: "test-model",
+  };
+  emit({ type: "message_end", message: failedAssistant });
+  emit({ type: "agent_end", messages: [failedAssistant], willRetry: false });
+}
+
+describe("subscribeEmbeddedAgentSession partial reply lifecycle", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+  });
+
+  it("joins a partial reply task created while terminal events settle", async () => {
+    let resolvePartial: (() => void) | undefined;
+    const onPartialReply = vi.fn(
+      () =>
+        new Promise<void>((resolve) => {
+          resolvePartial = resolve;
+        }),
+    );
+    const { emit, subscription } = createSubscribedSessionHarness({
+      runId: "run-partial-provider-failure",
+      onBeforeTerminalDelivery: async () => undefined,
+      onPartialReply,
+    });
+
+    emitPartialThenProviderFailure(emit);
+    let settled = false;
+    const settlement = subscription.waitForPendingEvents().then(() => {
+      settled = true;
+    });
+
+    await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce());
+    await Promise.resolve();
+    expect(settled).toBe(false);
+
+    resolvePartial?.();
+    await settlement;
+    expect(settled).toBe(true);
+  });
+
+  it("contains and logs a rejected partial reply after unsubscribe", async () => {
+    const callbackError = new Error("draft send rejected");
+    let rejectPartial: ((reason: unknown) => void) | undefined;
+    const onPartialReply = vi.fn(
+      () =>
+        new Promise<void>((_resolve, reject) => {
+          rejectPartial = reject;
+        }),
+    );
+    const { emit, subscription } = createSubscribedSessionHarness({
+      runId: "run-partial-rejection",
+      onPartialReply,
+    });
+
+    emit({
+      type: "message_update",
+      message: { role: "assistant" },
+      assistantMessageEvent: { type: "text_delta", delta: "partial answer" },
+    });
+
+    await vi.waitFor(() => expect(onPartialReply).toHaveBeenCalledOnce());
+    subscription.unsubscribe();
+    rejectPartial?.(callbackError);
+    await expect(subscription.waitForPendingEvents()).resolves.toBeUndefined();
+    expect(logger.warn).toHaveBeenCalledWith(
+      `assistant partial reply callback failed: ${String(callbackError)}`,
+    );
+  });
+});
@@ -308,6 +308,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
  const pendingMessagingTexts = state.pendingMessagingTexts;
  const pendingMessagingTargets = state.pendingMessagingTargets;
  const pendingBlockReplyTasks = new Set<Promise<void>>();
+  const pendingPartialReplyTasks = new Set<Promise<void>>();
  const replyDirectiveAccumulator = createStreamingDirectiveAccumulator();
  const partialReplyDirectiveAccumulator = createStreamingDirectiveAccumulator();
  const shouldAllowSilentTurnText = (text: string | undefined) =>
@@ -333,11 +334,22 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
      });
    }
    if (delivery.emitPartialReply && params.onPartialReply && state.shouldEmitPartialReplies) {
-      runBestEffortCallback({
-        label: "assistant partial reply",
-        log,
-        callback: () => params.onPartialReply?.(data),
-      });
+      try {
+        const maybeTask = params.onPartialReply(data);
+        if (isPromiseLike(maybeTask)) {
+          const task = Promise.resolve(maybeTask)
+            .then(() => undefined)
+            .catch((error: unknown) => {
+              log.warn(`assistant partial reply callback failed: ${String(error)}`);
+            });
+          pendingPartialReplyTasks.add(task);
+          void task.finally(() => {
+            pendingPartialReplyTasks.delete(task);
+          });
+        }
+      } catch (error) {
+        log.warn(`assistant partial reply callback failed: ${String(error)}`);
+      }
    }
  };
  const emitAssistantStreamData = (
@@ -1585,7 +1597,16 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess
    getCompactionCount: () => compactionCount,
    getLastCompactionTokensAfter: () => state.lastCompactionTokensAfter,
    getAssistantTurnCount: () => state.assistantTurnCount,
-    waitForPendingEvents: () => state.pendingEventChain ?? Promise.resolve(),
+    waitForPendingEvents: async () => {
+      // Partial presentation stays concurrent with provider events, but terminal
+      // settlement must observe callbacks launched while the event chain drains.
+      while (state.pendingEventChain || pendingPartialReplyTasks.size > 0) {
+        await Promise.allSettled([
+          ...(state.pendingEventChain ? [state.pendingEventChain] : []),
+          ...pendingPartialReplyTasks,
+        ]);
+      }
+    },
    getItemLifecycle: () => ({
      startedCount: state.itemStartedCount,
      completedCount: state.itemCompletedCount,