fix(messaging): hold general outbound on ambiguous send failures

Apply the connect-invite at-most-once pattern to mirrored DMs, group
posts, channel invites, and intros. Provider send errors now keep the
outbox row claimed as sent without a handle instead of reverting to
pending and risking duplicate delivery.

Pre-send setup failures (thread resolution) still retry with backoff.
Pass outbox idempotency keys through to sendToThread for future provider
dedupe.

Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>

packages/adapters/src/messaging-delivery.test.ts

@@ -197,7 +197,7 @@ describe("deliverMessagingOutbound", () => {
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
 
     expect(deps.sendToThread).toHaveBeenCalledWith(
-      { threadId: "sendblue:dm-1", body: "Hello from your bot" },
+      { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" },
       context,
     );
     expect(deps.rows).toEqual([
@@ -237,12 +237,20 @@ describe("deliverMessagingOutbound", () => {
     });
     expect(deps.sendToThread).toHaveBeenNthCalledWith(
       1,
-      { threadId: "sendblue:opened:+15551234567", body: "first" },
+      {
+        threadId: "sendblue:opened:+15551234567",
+        body: "first",
+        idempotencyKey: "msg:m-1",
+      },
       context,
     );
     expect(deps.sendToThread).toHaveBeenNthCalledWith(
       2,
-      { threadId: "sendblue:opened:+15551234567", body: "second" },
+      {
+        threadId: "sendblue:opened:+15551234567",
+        body: "second",
+        idempotencyKey: "msg:m-2",
+      },
       context,
     );
   });
@@ -271,7 +279,7 @@ describe("deliverMessagingOutbound", () => {
     const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } });
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
     expect(deps.sendToThread).toHaveBeenCalledWith(
-      { threadId: "sendblue:dm-1", body: "Hello from your bot" },
+      { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" },
       context,
     );
   });
@@ -305,22 +313,16 @@ describe("deliverMessagingOutbound", () => {
     ]);
   });
 
-  it("returns a transient send failure to pending and schedules a delayed retry", async () => {
+  it("does not re-queue a mirrored DM when the provider send fails", async () => {
     const deps = createDeps({ sendError: new Error("messaging provider 500") });
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
 
-    expect(deps.rows).toEqual([
+    // At-most-once: keep the claim rather than risk a duplicate DM.
-      expect.objectContaining({
+    expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "dm", status: "sent" }));
-        kind: "dm",
-        status: "pending",
-        attempts: 1,
-        nextAttemptAt: expect.any(Date),
-      }),
-    ]);
     expect(deps.rows[0]!.providerHandle ?? null).toBeNull();
-    expect(deps.jobs.enqueue).toHaveBeenCalledWith(
+    expect(deps.rows[0]!.attempts ?? 0).toBe(0);
-      expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }),
+    expect(deps.jobs.enqueue).not.toHaveBeenCalled();
-    );
+    expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled();
   });
 
   it("skips pending rows whose nextAttemptAt is still in the future", async () => {
@@ -364,7 +366,11 @@ describe("deliverMessagingOutbound", () => {
     await deliverMessagingOutbound(deps, {}, context);
 
     expect(deps.sendToThread).toHaveBeenCalledWith(
-      { threadId: "sendblue:dm-1", body: "pending earlier" },
+      {
+        threadId: "sendblue:dm-1",
+        body: "pending earlier",
+        idempotencyKey: "invite:ch-1:mi-1",
+      },
       context,
     );
     expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" }));
@@ -519,6 +525,28 @@ describe("deliverMessagingOutbound", () => {
     expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled();
   });
 
+  it("does not re-queue a channel invite when the provider send fails", async () => {
+    const deps = createDeps({
+      run: null,
+      sendError: new Error("provider down"),
+      outboundRows: [
+        {
+          id: "out-invite",
+          idempotencyKey: "invite:ch-1:mi-1",
+          kind: "dm",
+          identityId: "mi-1",
+          body: "join the group",
+          status: "pending",
+          providerHandle: null,
+        },
+      ],
+    });
+    await deliverMessagingOutbound(deps, {}, context);
+
+    expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" }));
+    expect(deps.jobs.enqueue).not.toHaveBeenCalled();
+  });
+
   it("does not re-queue a connect invite after a delivery transaction timeout", async () => {
     const deps = createDeps({
       run: null,
@@ -840,7 +868,11 @@ describe("deliverMessagingOutbound channel runs", () => {
     });
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
     expect(deps.sendToThread).toHaveBeenCalledWith(
-      { threadId: "sendblue:grp-1", body: "Alice's agent: Public update" },
+      {
+        threadId: "sendblue:grp-1",
+        body: "Alice's agent: Public update",
+        idempotencyKey: "msg:reply-1",
+      },
       context,
     );
   });
@@ -850,7 +882,11 @@ describe("deliverMessagingOutbound channel runs", () => {
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
 
     expect(deps.sendToThread).toHaveBeenCalledWith(
-      { threadId: "sendblue:grp-1", body: "Alice's agent: found it" },
+      {
+        threadId: "sendblue:grp-1",
+        body: "Alice's agent: found it",
+        idempotencyKey: "msg:m-1",
+      },
       context,
     );
     expect(deps.rows).toEqual([
@@ -949,6 +985,16 @@ describe("deliverMessagingOutbound channel runs", () => {
     );
   });
 
+  it("does not re-queue a group post when the provider send fails", async () => {
+    const deps = createChannelDeps();
+    deps.sendToThread.mockRejectedValueOnce(new Error("provider down"));
+    await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
+
+    expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "group", status: "sent" }));
+    expect(deps.rows[0]!.providerHandle ?? null).toBeNull();
+    expect(deps.enqueue).not.toHaveBeenCalled();
+  });
+
   it("does not wake anyone when the hop budget is exhausted", async () => {
     const deps = createChannelDeps({ text: "@Helper again?", sourceHop: 6 });
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
@@ -962,26 +1008,61 @@ describe("deliverMessagingOutbound channel runs", () => {
   });
 });
 
-describe("deliverMessagingOutbound transient failure retry", () => {
+describe("deliverMessagingOutbound pre-send failure retry", () => {
-  it("marks the row failed only once the attempt budget is exhausted", async () => {
+  it("returns a thread-resolution failure to pending and schedules a delayed retry", async () => {
     const deps = createDeps({
       run: null,
-      sendError: new Error("messaging provider 500"),
+      identity: { dmThreadId: null },
       outboundRows: [
         {
           id: "out-9",
-          idempotencyKey: "msg:m-9",
+          idempotencyKey: "invite:ch-1:mi-1",
           kind: "dm",
           identityId: "mi-1",
-          body: "fourth failure",
+          body: "needs thread lookup",
+          status: "pending",
+          providerHandle: null,
+          attempts: 0,
+        },
+      ],
+    });
+    deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503"));
+    await deliverMessagingOutbound(deps, {}, context);
+
+    expect(deps.sendToThread).not.toHaveBeenCalled();
+    expect(deps.rows).toEqual([
+      expect.objectContaining({
+        status: "pending",
+        attempts: 1,
+        nextAttemptAt: expect.any(Date),
+      }),
+    ]);
+    expect(deps.jobs.enqueue).toHaveBeenCalledWith(
+      expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }),
+    );
+  });
+
+  it("marks the row failed only once the pre-send attempt budget is exhausted", async () => {
+    const deps = createDeps({
+      run: null,
+      identity: { dmThreadId: null },
+      outboundRows: [
+        {
+          id: "out-9",
+          idempotencyKey: "invite:ch-1:mi-1",
+          kind: "dm",
+          identityId: "mi-1",
+          body: "last lookup try",
           status: "pending",
           providerHandle: null,
           attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS - 1,
         },
       ],
     });
+    deps.openDirectThread.mockRejectedValue(new Error("provider lookup 503"));
     await deliverMessagingOutbound(deps, {}, context);
 
+    expect(deps.sendToThread).not.toHaveBeenCalled();
     expect(deps.rows).toEqual([
       expect.objectContaining({ status: "failed", attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS }),
     ]);
@@ -991,8 +1072,8 @@ describe("deliverMessagingOutbound transient failure retry", () => {
 
 describe("deliverMessagingOutbound retry enqueue failure", () => {
   it("propagates the enqueue failure so the job queue retries the drain", async () => {
-    const deps = createDeps({});
+    const deps = createDeps({ identity: { dmThreadId: null } });
-    deps.sendToThread.mockRejectedValueOnce(new Error("messaging provider 500"));
+    deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503"));
     deps.jobs.enqueue.mockRejectedValueOnce(new Error("queue down"));
 
     // A swallowed enqueue failure would strand the row in pending forever:
@@ -1000,6 +1081,7 @@ describe("deliverMessagingOutbound retry enqueue failure", () => {
     await expect(deliverMessagingOutbound(deps, { runId: "run-1" }, context)).rejects.toThrow(
       "queue down",
     );
+    expect(deps.sendToThread).not.toHaveBeenCalled();
     expect(deps.rows).toEqual([
       expect.objectContaining({
         kind: "dm",
@@ -1011,8 +1093,9 @@ describe("deliverMessagingOutbound retry enqueue failure", () => {
 
     // The delayed messaging.deliver job fires at nextAttemptAt: make the row due.
     deps.rows[0]!.nextAttemptAt = new Date(Date.now() - 1);
+    deps.openDirectThread.mockResolvedValueOnce("sendblue:opened:+15551234567");
     await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
-    expect(deps.sendToThread).toHaveBeenCalledTimes(2);
+    expect(deps.sendToThread).toHaveBeenCalledTimes(1);
     expect(deps.rows).toEqual([
       expect.objectContaining({ kind: "dm", status: "sent", providerHandle: "handle-out-1" }),
     ]);

packages/adapters/src/messaging-delivery.ts

@@ -277,6 +277,53 @@ function connectInvitePair(
  * restore pending and duplicate the YES/NO DM. A lost invite is fine;
  * reconnect starts a fresh cycle.
  */
+/**
+ * At-most-once provider send: any error after the call (or a lost response)
+ * keeps the outer claim as `sent` so drain does not restore pending and
+ * duplicate the message. A lost send is acceptable; the outbox row stays
+ * held without a providerHandle until status webhooks or manual cleanup.
+ */
+async function sendOutboundAtMostOnce(
+  deps: MessagingDeliveryDeps,
+  row: { id: string; idempotencyKey: string; body: string },
+  threadId: string,
+  context: AdapterContext,
+): Promise<"delivered" | "held"> {
+  try {
+    const sent = await deps.messaging.sendToThread(
+      { threadId, body: row.body, idempotencyKey: row.idempotencyKey },
+      context,
+    );
+    await deps.prisma.messagingOutbound.updateMany({
+      where: { id: row.id },
+      data: { providerHandle: sent.handle },
+    });
+    return "delivered";
+  } catch {
+    return "held";
+  }
+}
+
+async function revertOutboundToPendingWithRetry(
+  deps: MessagingDeliveryDeps,
+  row: { id: string; attempts?: number | null },
+): Promise<void> {
+  const attempts = (row.attempts ?? 0) + 1;
+  const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS;
+  const retryAt = exhausted ? null : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts));
+  await deps.prisma.messagingOutbound.updateMany({
+    where: { id: row.id },
+    data: {
+      attempts,
+      status: exhausted ? "failed" : "pending",
+      nextAttemptAt: retryAt,
+    },
+  });
+  if (!exhausted && retryAt) {
+    await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt));
+  }
+}
+
 async function sendConnectInvite(
   deps: MessagingDeliveryDeps,
   row: { id: string; body: string },
@@ -343,99 +390,81 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
       data: { status: "sent", nextAttemptAt: null },
     });
     if (claim.count === 0) continue;
-    try {
+    if (row.kind === "group" || row.kind === "intro") {
-      if (row.kind === "group" || row.kind === "intro") {
+      if (!row.threadId) {
-        if (!row.threadId) {
-          await deps.prisma.messagingOutbound.update({
-            where: { id: row.id },
-            data: { status: "failed" },
-          });
-          continue;
-        }
-        const sent = await deps.messaging.sendToThread(
-          { threadId: row.threadId, body: row.body },
-          context,
-        );
-        await deps.prisma.messagingOutbound.update({
-          where: { id: row.id },
-          data: { providerHandle: sent.handle },
-        });
-        continue;
-      }
-      const identity = row.identityId
-        ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } })
-        : null;
-      if (!identity) {
         await deps.prisma.messagingOutbound.update({
           where: { id: row.id },
           data: { status: "failed" },
         });
         continue;
       }
-      // Sendblue's consecutive-outbound vendor cap; other providers have no
+      await sendOutboundAtMostOnce(
-      // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs.
+        deps,
-      if (
+        { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body },
-        identity.provider === "sendblue" &&
+        row.threadId,
-        identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP
+        context,
-      ) {
+      );
-        // Cap holds are not failures: release the claim back to pending.
+      continue;
-        await deps.prisma.messagingOutbound.update({
+    }
-          where: { id: row.id },
+    const identity = row.identityId
-          data: { status: "pending" },
+      ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } })
-        });
+      : null;
-        continue;
+    if (!identity) {
-      }
+      await deps.prisma.messagingOutbound.update({
-      const threadId = await resolveDirectThread(deps, identity, context);
-      const invitePair = connectInvitePair(row.idempotencyKey);
-      if (invitePair) {
-        const result = await sendConnectInvite(
-          deps,
-          { id: row.id, body: row.body },
-          threadId,
-          invitePair,
-          context,
-        );
-        if (result === "delivered") {
-          await deps.prisma.messagingIdentity.update({
-            where: { id: identity.id },
-            data: { outboundSinceInbound: { increment: 1 } },
-          });
-        }
-        continue;
-      }
-      const sent = await deps.messaging.sendToThread({ threadId, body: row.body }, context);
-      await deps.prisma.messagingOutbound.updateMany({
         where: { id: row.id },
-        data: { providerHandle: sent.handle },
+        data: { status: "failed" },
       });
+      continue;
+    }
+    // Sendblue's consecutive-outbound vendor cap; other providers have no
+    // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs.
+    if (
+      identity.provider === "sendblue" &&
+      identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP
+    ) {
+      // Cap holds are not failures: release the claim back to pending.
+      await deps.prisma.messagingOutbound.update({
+        where: { id: row.id },
+        data: { status: "pending" },
+      });
+      continue;
+    }
+    let threadId: string;
+    try {
+      threadId = await resolveDirectThread(deps, identity, context);
+    } catch {
+      // Pre-send setup failed; no provider call was made, so retry is safe.
+      await revertOutboundToPendingWithRetry(deps, row);
+      continue;
+    }
+    const invitePair = connectInvitePair(row.idempotencyKey);
+    if (invitePair) {
+      const result = await sendConnectInvite(
+        deps,
+        { id: row.id, body: row.body },
+        threadId,
+        invitePair,
+        context,
+      );
+      if (result === "delivered") {
+        await deps.prisma.messagingIdentity.update({
+          where: { id: identity.id },
+          data: { outboundSinceInbound: { increment: 1 } },
+        });
+      }
+      continue;
+    }
+    const result = await sendOutboundAtMostOnce(
+      deps,
+      { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body },
+      threadId,
+      context,
+    );
+    if (result === "delivered") {
       await deps.prisma.messagingIdentity.update({
         where: { id: identity.id },
         data: { outboundSinceInbound: { increment: 1 } },
       });
-    } catch {
-      // Transient provider errors go back to pending with a backed-off
-      // retry; only an exhausted budget is terminal.
-      const attempts = (row.attempts ?? 0) + 1;
-      const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS;
-      const retryAt = exhausted
-        ? null
-        : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts));
-      await deps.prisma.messagingOutbound.updateMany({
-        where: { id: row.id },
-        data: {
-          attempts,
-          status: exhausted ? "failed" : "pending",
-          nextAttemptAt: retryAt,
-        },
-      });
-      if (!exhausted && retryAt) {
-        // Propagate an enqueue failure: the messaging.deliver job then fails
-        // and the queue's own retry re-runs the drain. Swallowing it would
-        // strand the row in pending — no reconciler reclaims outbox rows.
-        // Re-entry is safe: the row is pending again and nextAttemptAt keeps
-        // other drains from racing the backoff window.
-        await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt));
-      }
     }
   }
 }