fix(outbound): prevent partial-send recovery replay

obviyus · obviyus · commit 210ea659f77f · 2026-06-25T08:16:56.000-07:00
diff --git a/src/infra/outbound/deliver.queue-integration.test.ts b/src/infra/outbound/deliver.queue-integration.test.ts
@@ -1,9 +1,3 @@
-// Integration test: real SQLite delivery queue + mock adapter.
-// Verifies the patch end-to-end at the queue layer: when a required-mode
-// batch send fails mid-batch after an earlier payload already succeeded,
-// the queue entry advances to recovery_state=unknown_after_send (not left
-// in send_attempt_started), so reconnect-drain routes it through
-// reconcileUnknownQueuedDelivery instead of blind replay.
 import { describe, expect, it, vi, beforeAll, beforeEach } from "vitest";
 import type { OpenClawConfig } from "../../config/config.js";
 import { drainPendingDeliveries, type DeliverFn, loadPendingDeliveries } from "./delivery-queue.js";
@@ -14,7 +8,6 @@ import {
 
 let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads;
 
-// Minimal reconnect drain helper (no adapter → reconcileUnknownQueuedDelivery returns null).
 async function drainMatrixReconnect(opts: { deliver: DeliverFn; stateDir: string }): Promise<void> {
   await drainPendingDeliveries({
     drainKey: "matrix:reconnect-test",
@@ -27,6 +20,27 @@ async function drainMatrixReconnect(opts: { deliver: DeliverFn; stateDir: string
   });
 }
 
+function createPartialSendFailure() {
+  return vi
+    .fn()
+    .mockResolvedValueOnce({ messageId: "m1" })
+    .mockRejectedValueOnce(new Error("second payload send failed"));
+}
+
+async function deliverPartialMatrixBatch(sendMatrix: ReturnType<typeof vi.fn>, tmpDir: string) {
+  process.env.OPENCLAW_STATE_DIR = tmpDir;
+  await expect(
+    deliverOutboundPayloads({
+      cfg: {} as OpenClawConfig,
+      channel: "matrix",
+      to: "!room:example",
+      payloads: [{ text: "first" }, { text: "second" }],
+      deps: { matrix: sendMatrix },
+      queuePolicy: "required",
+    }),
+  ).rejects.toThrow("second payload send failed");
+}
+
 describe("deliverOutboundPayloads queue integration: mid-batch failure with send evidence", () => {
   const fixtures = installDeliveryQueueTmpDirHooks();
   let tmpDir: string;
@@ -40,79 +54,36 @@ describe("deliverOutboundPayloads queue integration: mid-batch failure with send
   });
 
   it("advances queued entry to unknown_after_send when a later payload fails after an earlier one succeeded", async () => {
-    process.env.OPENCLAW_STATE_DIR = tmpDir;
-    // First payload succeeds (send evidence), second payload throws.
-    const sendMatrix = vi
-      .fn()
-      .mockResolvedValueOnce({ messageId: "m1" })
-      .mockRejectedValueOnce(new Error("second payload send failed"));
+    const sendMatrix = createPartialSendFailure();
 
-    await expect(
-      deliverOutboundPayloads({
-        cfg: {} as OpenClawConfig,
-        channel: "matrix",
-        to: "!room:example",
-        payloads: [{ text: "first" }, { text: "second" }],
-        deps: { matrix: sendMatrix },
-        queuePolicy: "required",
-      }),
-    ).rejects.toThrow("second payload send failed");
+    await deliverPartialMatrixBatch(sendMatrix, tmpDir);
 
-    // The entry must exist in the real SQLite queue and be in unknown_after_send.
-    const entries = await import("./delivery-queue.js").then((m) =>
-      m.loadPendingDeliveries(tmpDir),
-    );
+    const entries = await loadPendingDeliveries(tmpDir);
     expect(entries).toHaveLength(1);
     const entry = entries[0];
     expect(entry.recoveryState).toBe("unknown_after_send");
     expect(entry.retryCount).toBe(0);
     expect(entry.lastError).toBeUndefined();
-    // Sanity: the send actually happened for the first payload.
     expect(sendMatrix).toHaveBeenCalledTimes(2);
   });
 
   it("drain does not replay an unknown_after_send entry when no adapter reconciliation is available", async () => {
-    // Regression guard for the recovery/drain semantics: an entry in
-    // unknown_after_send (written by the patch above) must NOT be blindly
-    // replayed when the channel adapter cannot reconcile the unknown send.
-    // Without the patch the entry would stay in send_attempt_started, which
-    // has the same drain behaviour — but this test pins the contract so that
-    // any future regression that accidentally advances the state in a way that
-    // re-enables blind replay is caught.
-    process.env.OPENCLAW_STATE_DIR = tmpDir;
-    const sendMatrix = vi
-      .fn()
-      .mockResolvedValueOnce({ messageId: "m1" })
-      .mockRejectedValueOnce(new Error("second payload send failed"));
+    const sendMatrix = createPartialSendFailure();
 
-    // Drive the patch: entry lands in unknown_after_send.
-    await expect(
-      deliverOutboundPayloads({
-        cfg: {} as OpenClawConfig,
-        channel: "matrix",
-        to: "!room:example",
-        payloads: [{ text: "first" }, { text: "second" }],
-        deps: { matrix: sendMatrix },
-        queuePolicy: "required",
-      }),
-    ).rejects.toThrow("second payload send failed");
+    await deliverPartialMatrixBatch(sendMatrix, tmpDir);
 
     const beforeDrain = await loadPendingDeliveries(tmpDir);
     expect(beforeDrain[0]?.recoveryState).toBe("unknown_after_send");
 
-    // Reconnect drain with no adapter (cfg={}) — reconcileUnknownQueuedDelivery
-    // returns null → "refusing blind replay" branch → deliver is never called.
     const deliver = vi.fn<DeliverFn>(async () => {});
     await drainMatrixReconnect({ deliver, stateDir: tmpDir });
 
     expect(deliver).not.toHaveBeenCalled();
-    // The entry is moved to failed (not re-queued as pending), closing the drain loop.
     expect(await loadPendingDeliveries(tmpDir)).toHaveLength(0);
   });
 
-  it("leaves entry for retry (failDelivery, recovery_state stays null) when no send evidence", async () => {
+  it("leaves entry for retry in send_attempt_started when no send evidence exists", async () => {
     process.env.OPENCLAW_STATE_DIR = tmpDir;
-    // First (and only) payload fails immediately — no send evidence.
     const sendMatrix = vi.fn().mockRejectedValueOnce(new Error("first payload send failed"));
 
     await expect(
@@ -131,7 +102,6 @@ describe("deliverOutboundPayloads queue integration: mid-batch failure with send
     );
     expect(entries).toHaveLength(1);
     const entry = entries[0];
-    // No send evidence -> failDelivery path: retryCount bumped, recovery_state not advanced.
     expect(entry.retryCount).toBe(1);
     expect(entry.recoveryState).toBe("send_attempt_started");
     expect(entry.lastError).toContain("first payload send failed");
diff --git a/src/infra/outbound/deliver.test.ts b/src/infra/outbound/deliver.test.ts
@@ -1046,14 +1046,6 @@ describe("deliverOutboundPayloads", () => {
   });
 
   it("marks queued delivery as unknown-after-send (not failed) when a later payload fails after an earlier one succeeded", async () => {
-    // Regression: required-mode batch send where an earlier payload succeeded
-    // (results.length > 0, OutboundDeliveryError.sentBeforeError === true) but a
-    // later payload throws. Previously the wrapper catch called failDelivery,
-    // leaving the entry in `send_attempt_started` so reconnect drain later
-    // replayed it as "not yet sent" — producing duplicate messages when the
-    // adapter's unknown-send reconciliation misreported `not_sent`. The fix
-    // advances to `unknown_after_send` so drain routes the entry through
-    // reconcileUnknownQueuedDelivery instead of blind replay.
     const sendMatrix = vi
       .fn()
       .mockResolvedValueOnce({ messageId: "m1" })
@@ -1072,16 +1064,11 @@ describe("deliverOutboundPayloads", () => {
 
     expect(sendMatrix).toHaveBeenCalledTimes(2);
     expect(queueMocks.markDeliveryPlatformOutcomeUnknown).toHaveBeenCalledWith("mock-queue-id");
-    // Must NOT failDelivery — that would leave send_attempt_started for drain to replay.
     expect(queueMocks.failDelivery).not.toHaveBeenCalled();
-    // Must NOT ack — the batch did not fully succeed; reconcile must confirm the
-    // partial send rather than silently dropping the queue entry.
     expect(queueMocks.ackDelivery).not.toHaveBeenCalled();
   });
 
   it("still calls failDelivery when a payload fails before any send succeeded", async () => {
-    // No send evidence (sentBeforeError === false): failDelivery is correct —
-    // nothing reached the channel, so leaving the entry for retry is safe.
     const sendMatrix = vi.fn().mockRejectedValueOnce(new Error("first payload send failed"));
 
     await expect(
diff --git a/src/infra/outbound/deliver.ts b/src/infra/outbound/deliver.ts
@@ -1390,29 +1390,13 @@ async function deliverOutboundPayloadsWithQueueCleanup(
       if (isDeliveryAbortError(err)) {
         await ackDelivery(queueId).catch(() => {});
       } else if (!platformResultsReturned) {
-        // If the platform send already started and the error carries partial
-        // send evidence (OutboundDeliveryError with sentBeforeError), the
-        // message may already have reached the channel. Calling failDelivery
-        // here leaves the entry in `send_attempt_started`, which reconnect
-        // drain later replays as "not yet sent" — producing duplicate
-        // messages when the adapter's unknown-send reconciliation misreports
-        // `not_sent`. Advance to `unknown_after_send` instead so drain routes
-        // the entry through reconcileUnknownQueuedDelivery (query the adapter
-        // for actual send state) rather than blind replay.
         const sendEvidence =
           platformSendStarted && err instanceof OutboundDeliveryError && err.sentBeforeError;
         if (sendEvidence) {
           await markQueuedPlatformOutcomeUnknown({
             queueId,
             queuePolicy,
           }).catch((markErr: unknown) => {
-            // markQueuedPlatformOutcomeUnknown failed (e.g. DB write error).
-            // Fall back to failDelivery so the entry is not silently abandoned.
-            // This is a last-resort path: the entry will re-enter
-            // send_attempt_started and be subject to the same drain/reconcile
-            // semantics as before the patch. It does NOT mean failDelivery is
-            // correct when send evidence exists — only that we cannot safely
-            // advance the state without a successful DB write.
             log.warn(
               `failed to mark queued delivery ${queueId} as platform-outcome-unknown after mid-send error; falling back to fail: ${formatErrorMessage(markErr)}`,
             );
diff --git a/src/infra/outbound/delivery-queue-recovery.ts b/src/infra/outbound/delivery-queue-recovery.ts
@@ -389,15 +389,19 @@ async function drainQueuedEntry(opts: {
         return "failed";
       }
     }
-    if (reconciliation?.status === "not_sent") {
+    const reconciliationProvedPreSendFailure =
+      reconciliation?.status === "not_sent" && entry.recoveryState === "send_attempt_started";
+    if (reconciliationProvedPreSendFailure) {
       opts.log.info(
         `Delivery entry ${entry.id} reconciled ${entry.recoveryState} as not sent; replaying`,
       );
     } else {
-      const errMsg =
-        reconciliation?.status === "unresolved" && reconciliation.error
-          ? `delivery state is ${entry.recoveryState} and reconciliation is unresolved: ${reconciliation.error}`
-          : `delivery state is ${entry.recoveryState}; refusing blind replay without adapter reconciliation`;
+      let errMsg = `delivery state is ${entry.recoveryState}; refusing blind replay without adapter reconciliation`;
+      if (reconciliation?.status === "not_sent") {
+        errMsg = `delivery state is ${entry.recoveryState}; refusing full replay after post-send evidence`;
+      } else if (reconciliation?.status === "unresolved" && reconciliation.error) {
+        errMsg = `delivery state is ${entry.recoveryState} and reconciliation is unresolved: ${reconciliation.error}`;
+      }
       opts.log.warn(`Delivery entry ${entry.id} ${errMsg}`);
       opts.onFailed?.(entry, errMsg);
       if (reconciliation?.status === "unresolved" && reconciliation.retryable === true) {
diff --git a/src/infra/outbound/delivery-queue.recovery.test.ts b/src/infra/outbound/delivery-queue.recovery.test.ts
@@ -328,7 +328,7 @@ describe("delivery-queue recovery", () => {
     expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0);
   });
 
-  it("replays unknown-after-send entries only after adapter proves they were not sent", async () => {
+  it("moves unknown-after-send entries to failed when adapter reports not sent", async () => {
     const id = await enqueueDelivery(
       { channel: "demo-channel-a", to: "+1", payloads: [{ text: "not sent" }] },
       tmpDir(),
@@ -346,24 +346,19 @@ describe("delivery-queue recovery", () => {
     });
 
     const deliver = vi.fn().mockResolvedValue([]);
-    const { result } = await runRecovery({ deliver });
+    const log = createRecoveryLog();
+    const { result } = await runRecovery({ deliver, log });
 
-    expect(deliver).toHaveBeenCalledTimes(1);
-    const deliverInput = mockCallArg(deliver) as {
-      channel?: string;
-      to?: string;
-      skipQueue?: boolean;
-    };
-    expect(deliverInput.channel).toBe("demo-channel-a");
-    expect(deliverInput.to).toBe("+1");
-    expect(deliverInput.skipQueue).toBe(true);
+    expect(deliver).not.toHaveBeenCalled();
     expect(result).toEqual({
-      recovered: 1,
-      failed: 0,
+      recovered: 0,
+      failed: 1,
       skippedMaxRetries: 0,
       deferredBackoff: 0,
     });
     expect(await loadPendingDeliveries(tmpDir())).toHaveLength(0);
+    expect(readOutboundQueueStatus(tmpDir(), id)).toBe("failed");
+    expectMockMessageContaining(log.warn, "refusing full replay after post-send evidence");
   });
 
   it("keeps retryable unresolved unknown-after-send entries on the queue without replaying", async () => {
