fix(release): repair delivery and live channel QA

extensions/qa-lab/src/live-transports/matrix/substrate/config.test.ts

@@ -290,7 +290,10 @@ describe("matrix qa config", () => {
         },
         groupMentionPatterns: ["\\S"],
         groupPolicy: "open",
-        streaming: true,
+        streaming: {
+          mode: "partial",
+          progress: { commandText: "raw" },
+        },
       },
       sutAccessToken: "sut-token",
       sutAccountId: "sut",
@@ -307,11 +310,12 @@ describe("matrix qa config", () => {
       chunkMode: "length",
       mode: "partial",
       preview: { toolProgress: true },
+      progress: { commandText: "raw" },
     });
     expect(config.messages?.groupChat?.mentionPatterns).toEqual(["\\S"]);
   });
 
-  it("resets tool progress when a scalar streaming override follows an opt-out", () => {
+  it("resets QA streaming detail overrides when a scalar override follows", () => {
     const optedOut = buildMatrixQaConfig({} as OpenClawConfig, {
       driverUserId: "@driver:matrix-qa.test",
       homeserver: "http://127.0.0.1:28008/",
@@ -320,6 +324,7 @@ describe("matrix qa config", () => {
         streaming: {
           mode: "quiet",
           preview: { toolProgress: false },
+          progress: { commandText: "raw" },
         },
       },
       sutAccessToken: "sut-token",
@@ -343,6 +348,7 @@ describe("matrix qa config", () => {
       chunkMode: "length",
       mode: "quiet",
       preview: { toolProgress: false },
+      progress: { commandText: "raw" },
     });
     expect(reset.channels?.matrix?.accounts?.sut?.streaming).toEqual({
       block: { enabled: false },

extensions/qa-lab/src/live-transports/matrix/substrate/config.ts

@@ -19,6 +19,9 @@ type MatrixQaStreamingConfig = {
   preview?: {
     toolProgress?: boolean;
   };
+  progress?: {
+    commandText?: "raw" | "status";
+  };
 };
 type MatrixQaAgentDefaultsOverrides = {
   blockStreamingChunk?: {
@@ -128,6 +131,7 @@ type MatrixQaConfigSnapshot = {
   startupVerification?: "if-unverified" | "off";
   streaming: MatrixQaStreamingMode;
   streamingPreviewToolProgress: boolean;
+  streamingProgressCommandText?: "raw" | "status";
   textChunkLimit?: number;
   threadBindings: MatrixQaThreadBindingsConfigOverrides;
   threadReplies: MatrixQaThreadRepliesMode;
@@ -445,6 +449,9 @@ function buildMatrixQaChannelAccountConfig(params: {
       chunkMode: params.snapshot.chunkMode ?? "length",
       mode: params.snapshot.streaming,
       preview: { toolProgress: params.snapshot.streamingPreviewToolProgress },
+      ...(params.snapshot.streamingProgressCommandText !== undefined
+        ? { progress: { commandText: params.snapshot.streamingProgressCommandText } }
+        : {}),
     },
   };
   const startupVerificationConfig =
@@ -497,6 +504,7 @@ function buildMatrixQaConfigSnapshot(params: {
   sutUserId: string;
   topology: MatrixQaProvisionedTopology;
 }): MatrixQaConfigSnapshot {
+  const streaming = params.overrides?.streaming;
   return {
     allowBots: params.overrides?.allowBots,
     autoJoin: params.overrides?.autoJoin ?? "off",
@@ -516,10 +524,11 @@ function buildMatrixQaConfigSnapshot(params: {
     }),
     replyToMode: params.overrides?.replyToMode ?? "off",
     startupVerification: params.overrides?.startupVerification,
-    streaming: resolveMatrixQaStreamingMode(params.overrides?.streaming),
-    streamingPreviewToolProgress: resolveMatrixQaStreamingPreviewToolProgress(
-      params.overrides?.streaming,
-    ),
+    streaming: resolveMatrixQaStreamingMode(streaming),
+    streamingPreviewToolProgress: resolveMatrixQaStreamingPreviewToolProgress(streaming),
+    streamingProgressCommandText: isMatrixQaStreamingConfig(streaming)
+      ? streaming.progress?.commandText
+      : undefined,
     threadBindings: { ...params.overrides?.threadBindings },
     textChunkLimit: params.overrides?.textChunkLimit,
     threadReplies: params.overrides?.threadReplies ?? "inbound",

extensions/qa-lab/src/live-transports/shared/scenario-selection.test.ts

@@ -138,6 +138,8 @@ describe("live transport QA scenario selection", () => {
 
   it.each([
     { channelId: "matrix", scenarioId: "thread-follow-up" },
+    { channelId: "buzz", scenarioId: "channel-canary" },
+    { channelId: "telegram", scenarioId: "channel-canary" },
     { channelId: "telegram", scenarioId: "channel-message-flows" },
   ] as const)(
     "keeps $scenarioId eligible through both $channelId drivers",
@@ -154,4 +156,17 @@ describe("live transport QA scenario selection", () => {
       expect(selectForDriver("crabline")).toEqual([scenarioId]);
     },
   );
+
+  it("rejects the shared channel canary on unsupported Discord drivers", () => {
+    expect(() =>
+      resolveCatalogLiveTransportQaScenarioIds({
+        ...MOCK_LANE,
+        channelId: "discord",
+        channelDriver: "crabline",
+        scenarioIds: ["channel-canary"],
+      }),
+    ).toThrow(
+      "selected QA scenario(s) do not match the current QA lane: channel-canary (channel=qa-channel|telegram|buzz)",
+    );
+  });
 });

extensions/qa-lab/src/scenario-catalog-channels.test.ts

@@ -189,6 +189,35 @@ describe("qa scenario catalog channel contracts", () => {
     expect(scenario.gatewayConfigPatch).not.toHaveProperty("channels.telegram.groups");
   });
 
+  it("keeps the shared channel canary eligible for QA Channel, Telegram, and Buzz", () => {
+    const scenario = requireFlowScenario(readQaScenarioById("channel-canary"));
+
+    expect(scenario.execution.channels).toEqual(["qa-channel", "telegram", "buzz"]);
+  });
+
+  it("keeps raw Matrix command text scoped to the mention-safety scenario", () => {
+    const scenario = requireFlowScenario(
+      readQaScenarioById("matrix-room-tool-progress-mention-safety"),
+    );
+    const ordinaryProgressScenario = requireFlowScenario(
+      readQaScenarioById("matrix-room-tool-progress-preview"),
+    );
+
+    expect(scenario.execution.channel).toBe("matrix");
+    expect(scenario.execution.config).toMatchObject({
+      matrixConfigOverrides: {
+        streaming: {
+          mode: "partial",
+          progress: { commandText: "raw" },
+        },
+        toolProfile: "coding",
+      },
+    });
+    expect(ordinaryProgressScenario.execution.config).not.toHaveProperty(
+      "matrixConfigOverrides.streaming.progress.commandText",
+    );
+  });
+
   it("keeps transcript-role delivery on the Crabline driver", () => {
     const scenario = readQaScenarioById("telegram-assistant-transcript-role-boundary");
     const config = readQaScenarioExecutionConfig("telegram-assistant-transcript-role-boundary") as

qa/scenarios/channels/channel-canary.yaml

@@ -21,6 +21,10 @@ scenario:
     - extensions/qa-lab/src/suite.ts
   execution:
     kind: flow
+    channels:
+      - qa-channel
+      - telegram
+      - buzz
     summary: Run the shared channel canary through QA Channel, Crabline, or a live adapter.
     transportPolicy:
       requireGroupMention: true

qa/scenarios/channels/matrix-room-tool-progress-mention-safety.yaml

@@ -12,7 +12,10 @@ scenario:
     retryCount: 0
     config:
       matrixConfigOverrides:
-        streaming: partial
+        streaming:
+          mode: partial
+          progress:
+            commandText: raw
         toolProfile: coding
 
 flow:

src/infra/delivery-queue-sqlite-claim.test.ts

@@ -0,0 +1,87 @@
+import { describe, expect, it, vi } from "vitest";
+import {
+  claimDeliveryQueueEntryPlatformSend,
+  dispatchDeliveryQueueEntryPlatformSend,
+} from "./delivery-queue-sqlite-claim.js";
+import { loadDeliveryQueueEntry, upsertDeliveryQueueEntry } from "./delivery-queue-sqlite.js";
+import { installDeliveryQueueTmpDirHooks } from "./outbound/delivery-queue.test-helpers.js";
+
+describe("delivery queue SQLite dispatch ownership", () => {
+  const { tmpDir } = installDeliveryQueueTmpDirHooks();
+  const queueName = "test-dispatch-owner";
+
+  it("atomically promotes dispatch ownership and rejects expired or replaced claims", () => {
+    vi.useFakeTimers();
+    try {
+      vi.setSystemTime(new Date("2026-08-10T10:00:00.000Z"));
+      const stateDir = tmpDir();
+      const id = "cron-direct-delivery:v1:dispatch-owner";
+      upsertDeliveryQueueEntry({
+        queueName,
+        entry: {
+          id,
+          enqueuedAt: Date.now(),
+          retryCount: 0,
+          completionRetention: {
+            idPrefix: "cron-direct-delivery:v1:",
+            maxAgeMs: 24 * 60 * 60_000,
+            maxEntries: 2,
+          },
+          requiresProducerClaim: true,
+        },
+        stateDir,
+      });
+
+      const expiredClaimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir });
+      if (!expiredClaimId) {
+        throw new Error("test invariant: the first producer claim must be available");
+      }
+      vi.advanceTimersByTime(30_001);
+      expect(
+        dispatchDeliveryQueueEntryPlatformSend({
+          queueName,
+          id,
+          claimId: expiredClaimId,
+          stateDir,
+        }),
+      ).toBe(false);
+
+      const claimId = claimDeliveryQueueEntryPlatformSend({ queueName, id, stateDir });
+      if (!claimId) {
+        throw new Error("test invariant: the replacement producer claim must be available");
+      }
+      expect(
+        dispatchDeliveryQueueEntryPlatformSend({
+          queueName,
+          id,
+          claimId: expiredClaimId,
+          stateDir,
+        }),
+      ).toBe(false);
+      expect(
+        dispatchDeliveryQueueEntryPlatformSend({
+          queueName,
+          id,
+          claimId,
+          stateDir,
+          route: { replyToId: "thread-1" },
+        }),
+      ).toBe(true);
+      expect(loadDeliveryQueueEntry(queueName, id, stateDir)).toMatchObject({
+        recoveryState: "send_attempt_started",
+        platformSendAttemptId: claimId,
+        platformSendStartedAt: Date.now(),
+        effectiveReplyToId: "thread-1",
+        availableAt: Date.now() + 30_000,
+      });
+      expect(loadDeliveryQueueEntry(queueName, id, stateDir)?.producerClaimId).toBeUndefined();
+
+      vi.advanceTimersByTime(30_001);
+      expect(dispatchDeliveryQueueEntryPlatformSend({ queueName, id, claimId, stateDir })).toBe(
+        false,
+      );
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});

src/infra/delivery-queue-sqlite-claim.ts

@@ -56,9 +56,9 @@ export function transitionOwnedDeliveryQueueEntry(
   );
 }
 
-function transitionUnsentDeliveryQueueEntry(
+function transitionDeliveryQueueEntryPlatformSend(
   params: PlatformClaimParams,
-  operation: "claim" | "promote",
+  operation: "claim" | "promote" | "dispatch",
   transition: (entry: DeliveryQueueEntryState, now: number) => DeliveryQueueEntryState | undefined,
 ): boolean {
   // State-database opens reuse the canonical path-owned connection, so both
@@ -70,13 +70,16 @@ function transitionUnsentDeliveryQueueEntry(
     database.db,
     () => {
       const current = loadDeliveryQueueEntry(params.queueName, params.id, params.stateDir);
+      if (!current) {
+        return false;
+      }
       if (
-        !current ||
-        (current.platformSendStartedAt !== undefined &&
-          (operation !== "claim" ||
-            current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt ||
-            current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId ||
-            typeof current.platformSendAttemptId !== "string"))
+        current.platformSendStartedAt !== undefined &&
+        (operation === "promote" ||
+          (operation === "claim" &&
+            (current.platformSendStartedAt !== params.reconciledPlatformSendStartedAt ||
+              current.platformSendAttemptId !== params.reconciledPlatformSendAttemptId ||
+              typeof current.platformSendAttemptId !== "string")))
       ) {
         return false;
       }
@@ -102,7 +105,7 @@ export function claimDeliveryQueueEntryPlatformSend(
   params: PlatformClaimParams,
 ): string | undefined {
   const claimId = generateSecureUuid();
-  return transitionUnsentDeliveryQueueEntry(params, "claim", (entry, now) => {
+  return transitionDeliveryQueueEntryPlatformSend(params, "claim", (entry, now) => {
     const reconciledNotSent =
       entry.recoveryState === "send_attempt_started" &&
       typeof params.reconciledPlatformSendStartedAt === "number" &&
@@ -185,7 +188,7 @@ export function promoteDeliveryQueueEntryPlatformSend(
     route?: { replyToId?: string | null };
   },
 ): boolean {
-  return transitionUnsentDeliveryQueueEntry(params, "promote", (entry, now) =>
+  return transitionDeliveryQueueEntryPlatformSend(params, "promote", (entry, now) =>
     entry.recoveryState === "producer_claimed" &&
     entry.producerClaimId === params.claimId &&
     typeof entry.availableAt === "number" &&
@@ -207,3 +210,49 @@ export function promoteDeliveryQueueEntryPlatformSend(
       : undefined,
   );
 }
+
+/** Atomically authorize dispatch, promoting a producer claim into the active attempt. */
+export function dispatchDeliveryQueueEntryPlatformSend(
+  params: PlatformClaimParams & {
+    claimId: string;
+    route?: { replyToId?: string | null };
+  },
+): boolean {
+  return transitionDeliveryQueueEntryPlatformSend(params, "dispatch", (entry, now) => {
+    const producerOwned =
+      entry.recoveryState === "producer_claimed" &&
+      entry.producerClaimId === params.claimId &&
+      typeof entry.availableAt === "number" &&
+      entry.availableAt > now;
+    const attemptOwned =
+      (entry.recoveryState === "send_attempt_started" ||
+        entry.recoveryState === "unknown_after_send") &&
+      entry.platformSendAttemptId === params.claimId &&
+      (entry.requiresProducerClaim !== true ||
+        (typeof entry.availableAt === "number" && entry.availableAt > now));
+    if (!producerOwned && !attemptOwned) {
+      return undefined;
+    }
+    return {
+      ...entry,
+      // Exact reconciliation can skip pre-send promotion, so publish attempt identity
+      // atomically; later batch dispatches retain stronger unknown-after-send evidence.
+      availableAt:
+        entry.requiresProducerClaim === true
+          ? producerOwned
+            ? now + PLATFORM_SEND_OWNER_LEASE_MS
+            : entry.availableAt
+          : undefined,
+      producerClaimId: undefined,
+      platformSendAttemptId: params.claimId,
+      platformSendStartedAt: now,
+      ...(params.route && "replyToId" in params.route
+        ? { effectiveReplyToId: params.route.replyToId ?? null }
+        : {}),
+      recoveryState:
+        entry.recoveryState === "unknown_after_send"
+          ? "unknown_after_send"
+          : "send_attempt_started",
+    };
+  });
+}

src/infra/outbound/deliver-queue.exact-reconciliation.integration.test.ts

@@ -0,0 +1,101 @@
+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { createMessageReceiptFromOutboundResults } from "../../channels/message/receipt.js";
+import type { ChannelMessageSendTextContext } from "../../channels/message/types.js";
+import type { OpenClawConfig } from "../../config/config.js";
+import { createEmptyPluginRegistry } from "../../plugins/registry.js";
+import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../../plugins/runtime.js";
+import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js";
+import { getDeliveryQueueEntryStatus } from "../delivery-queue-sqlite.js";
+import {
+  boundedCronCompletionRetention,
+  drainMatrixReconnect,
+  matrixOutboundForQueueTest,
+} from "./deliver.queue-integration.test-support.js";
+import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js";
+import type { DeliverFn } from "./delivery-queue.js";
+import { installDeliveryQueueTmpDirHooks } from "./delivery-queue.test-helpers.js";
+
+let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads;
+
+describe("exact Matrix delivery queue reconciliation", () => {
+  const fixtures = installDeliveryQueueTmpDirHooks();
+  let tmpDir: string;
+
+  beforeAll(async () => {
+    ({ deliverOutboundPayloads } = await import("./deliver.js"));
+  });
+
+  beforeEach(() => {
+    tmpDir = fixtures.tmpDir();
+  });
+
+  afterEach(() => {
+    resetPluginRuntimeStateForTest();
+    setActivePluginRegistry(createEmptyPluginRegistry());
+  });
+
+  it.each(["required", "best_effort"] as const)(
+    "settles one exact Matrix %s send without restart replay",
+    async (queuePolicy) => {
+      process.env.OPENCLAW_STATE_DIR = tmpDir;
+      const deliveryIntentId = `cron-direct-delivery:v1:exact-${queuePolicy}-completion`;
+      const messageId = `exact-${queuePolicy}-message`;
+      const reconcileUnknownSend = vi.fn();
+      const sendText = vi.fn(async (ctx: ChannelMessageSendTextContext) => {
+        expect(ctx.deliveryQueueId).toBe(deliveryIntentId);
+        await ctx.onPlatformSendDispatch?.();
+        return {
+          messageId,
+          receipt: createMessageReceiptFromOutboundResults({
+            results: [{ channel: "matrix", messageId }],
+            kind: "text",
+          }),
+        };
+      });
+      setActivePluginRegistry(
+        createTestRegistry([
+          {
+            pluginId: "matrix",
+            source: "test",
+            plugin: {
+              ...createOutboundTestPlugin({ id: "matrix", outbound: matrixOutboundForQueueTest }),
+              message: {
+                id: "matrix",
+                durableFinal: {
+                  capabilities: { text: true, reconcileUnknownSend: true },
+                  reconcileUnknownSendKinds: { text: true },
+                  reconcileUnknownSend,
+                },
+                send: { text: sendText },
+              },
+            },
+          },
+        ]),
+      );
+      const params = {
+        cfg: {} as OpenClawConfig,
+        channel: "matrix" as const,
+        to: "!room:example",
+        payloads: [{ text: "send exactly once with durable platform identity" }],
+        queuePolicy,
+        ...(queuePolicy === "best_effort" ? { bestEffort: true } : {}),
+        deliveryIntentId,
+        completionRetention: boundedCronCompletionRetention,
+        reusePendingDeliveryIntent: true,
+        requireUnknownSendReconciliation: true,
+      };
+
+      await expect(deliverOutboundPayloads(params)).resolves.toMatchObject([{ messageId }]);
+      expect(sendText).toHaveBeenCalledOnce();
+      expect(reconcileUnknownSend).not.toHaveBeenCalled();
+      expect(
+        getDeliveryQueueEntryStatus(OUTBOUND_DELIVERY_QUEUE_NAME, deliveryIntentId, tmpDir),
+      ).toBe("completed");
+
+      const recoveryDeliver = vi.fn<DeliverFn>(async () => []);
+      await drainMatrixReconnect({ deliver: recoveryDeliver, stateDir: tmpDir });
+      expect(recoveryDeliver).not.toHaveBeenCalled();
+      expect(sendText).toHaveBeenCalledOnce();
+    },
+  );
+});

src/infra/outbound/delivery-queue-platform-lease.ts

@@ -1,9 +1,26 @@
 import {
   claimDeliveryQueueEntryPlatformSend,
+  dispatchDeliveryQueueEntryPlatformSend,
   renewDeliveryQueueEntryPlatformSendLease,
 } from "../delivery-queue-sqlite-claim.js";
 import { OUTBOUND_DELIVERY_QUEUE_NAME } from "./delivery-queue-media-staging.js";
 
+/** Atomically transfer a stable pending producer intent to one platform sender. */
+export async function claimDeliveryPlatformSendAttempt(
+  id: string,
+  stateDir?: string,
+  reconciledPlatformSendStartedAt?: number,
+  reconciledPlatformSendAttemptId?: string,
+): Promise<string | undefined> {
+  return claimDeliveryQueueEntryPlatformSend({
+    queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
+    id,
+    stateDir,
+    ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}),
+    ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}),
+  });
+}
+
 /** Claim and atomically upgrade a live reusable producer to renewable ownership. */
 export async function claimReusableDeliveryPlatformSendAttempt(
   id: string,
@@ -30,3 +47,22 @@ export async function renewDeliveryPlatformSendLease(
     claimId,
   });
 }
+
+/** Promote or refresh the exact live owner at recipient-visible dispatch. */
+export function markOwnedDeliveryPlatformSendDispatched(
+  id: string,
+  stateDir: string | undefined,
+  route: { replyToId?: string | null } | undefined,
+  claimId: string,
+): void {
+  const dispatched = dispatchDeliveryQueueEntryPlatformSend({
+    queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
+    id,
+    stateDir,
+    route,
+    claimId,
+  });
+  if (!dispatched) {
+    throw new Error(`Stable delivery platform claim was lost: ${id}`);
+  }
+}

src/infra/outbound/delivery-queue-storage.ts

@@ -9,7 +9,6 @@ import type {
 import type { ReplyToMode } from "../../config/types.js";
 import type { PluginHookReplyPayloadSendingContext } from "../../plugins/hook-types.js";
 import {
-  claimDeliveryQueueEntryPlatformSend,
   promoteDeliveryQueueEntryPlatformSend,
   transitionOwnedDeliveryQueueEntry,
 } from "../delivery-queue-sqlite-claim.js";
@@ -43,6 +42,7 @@ import {
   OUTBOUND_DELIVERY_QUEUE_NAME,
   OUTBOUND_LEGACY_PREPARATION_QUEUE_NAME,
 } from "./delivery-queue-media-staging.js";
+import { markOwnedDeliveryPlatformSendDispatched } from "./delivery-queue-platform-lease.js";
 import {
   StableDeliveryPreparationLostError,
   type StableDeliveryPreparation,
@@ -474,21 +474,7 @@ export async function failDeliveryAfterPlatformSend(
   );
 }
 
-/** Atomically transfer a stable pending producer intent to one platform sender. */
-export async function claimDeliveryPlatformSendAttempt(
-  id: string,
-  stateDir?: string,
-  reconciledPlatformSendStartedAt?: number,
-  reconciledPlatformSendAttemptId?: string,
-): Promise<string | undefined> {
-  return claimDeliveryQueueEntryPlatformSend({
-    queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
-    id,
-    stateDir,
-    ...(reconciledPlatformSendStartedAt !== undefined ? { reconciledPlatformSendStartedAt } : {}),
-    ...(reconciledPlatformSendAttemptId !== undefined ? { reconciledPlatformSendAttemptId } : {}),
-  });
-}
+export { claimDeliveryPlatformSendAttempt } from "./delivery-queue-platform-lease.js";
 
 /** Reserve one durable delivery call before invoking the provider path. */
 export async function reserveDeliveryAttempt(
@@ -572,18 +558,23 @@ export async function markDeliveryPlatformSendDispatched(
   route?: { replyToId?: string | null },
   expectedPlatformSendAttemptId?: string | null,
 ): Promise<void> {
+  if (typeof expectedPlatformSendAttemptId === "string") {
+    markOwnedDeliveryPlatformSendDispatched(id, stateDir, route, expectedPlatformSendAttemptId);
+    return;
+  }
   updateQueuedDelivery(
     id,
     stateDir,
     (entry) => ({
       ...entry,
-      // Dispatch still belongs to the promoted producer until provider I/O
-      // settles; clearing its lease lets another process replay an active send.
-      availableAt: expectedPlatformSendAttemptId ? entry.availableAt : undefined,
+      availableAt: undefined,
       producerClaimId: undefined,
       platformSendStartedAt: Date.now(),
       ...(route && "replyToId" in route ? { effectiveReplyToId: route.replyToId ?? null } : {}),
-      recoveryState: "send_attempt_started",
+      // A later batch send must not erase concrete evidence from an earlier result;
+      // recovery could otherwise replay the whole batch and duplicate that delivery.
+      recoveryState:
+        entry.recoveryState === "unknown_after_send" ? entry.recoveryState : "send_attempt_started",
     }),
     expectedPlatformSendAttemptId,
   );

src/infra/outbound/delivery-queue.storage.test.ts

@@ -547,6 +547,25 @@ describe("delivery-queue storage", () => {
       expect(entry.recoveryState).toBe("send_attempt_started");
     });
 
+    it("keeps ambiguous post-send evidence across a later unclaimed batch dispatch", async () => {
+      const id = await enqueueTextDelivery(
+        {
+          channel: "forum",
+          to: "123",
+          payloads: [{ text: "test" }],
+        },
+        tmpDir(),
+      );
+
+      await markDeliveryPlatformSendAttemptStarted(id, tmpDir());
+      await markDeliveryPlatformOutcomeUnknown(id, tmpDir());
+      await markDeliveryPlatformSendDispatched(id, tmpDir());
+
+      // Downgrading to send_attempt_started would let recovery replay the whole
+      // batch as not_sent and duplicate the payload that already reached the platform.
+      expect(readQueuedEntry(tmpDir(), id).recoveryState).toBe("unknown_after_send");
+    });
+
     it("increments retryCount, records attempt time, and sets lastError", async () => {
       const id = await enqueueTextDelivery(
         {