From 2ee72af28b405a04fc74b691145b230aabcbf636 Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Tue, 22 Sep 2026 07:02:20 +0000
Subject: [PATCH] fix(messaging): set introPostedAt only after intro delivery
 succeeds
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Group intros were stamped as posted before the outbox drain ran. When
delivery enqueue failed or the intro row exhausted retries, the channel
stayed marked and no retry path existed — unlinked participants never saw
the onboarding message.

Stamp introPostedAt in the drain path after a successful provider send,
matching the messagingMirroredAt pattern from #42.

Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>
---
 apps/api/src/messaging-inbound.test.ts        | 14 +++++
 apps/api/src/messaging-inbound.ts             |  4 --
 .../adapters/src/messaging-delivery.test.ts   | 53 +++++++++++++++++++
 packages/adapters/src/messaging-delivery.ts   |  6 +++
 4 files changed, 73 insertions(+), 4 deletions(-)

diff --git a/apps/api/src/messaging-inbound.test.ts b/apps/api/src/messaging-inbound.test.ts
index 5f6991d..a331198 100644
--- a/apps/api/src/messaging-inbound.test.ts
+++ b/apps/api/src/messaging-inbound.test.ts
@@ -804,6 +804,20 @@ describe("createMessagingInboundHandler channel routing", () => {
     expect(deps.sendUserMessage).not.toHaveBeenCalled();
   });
 
+  it("does not mark intro posted before delivery enqueue succeeds", async () => {
+    const deps = createDeps();
+    deps.jobs.enqueue = vi.fn(async () => {
+      throw new Error("queue unavailable");
+    });
+    const handle = createMessagingInboundHandler(deps);
+    await handle(groupEvent);
+
+    expect(deps.outboundRows).toEqual(
+      expect.arrayContaining([expect.objectContaining({ kind: "intro", idempotencyKey: "intro:ch-1" })]),
+    );
+    expect(deps.prisma.messagingChannel.update).not.toHaveBeenCalled();
+  });
+
   it("does not post a second intro once one was posted", async () => {
     const deps = createDeps();
     deps.prisma.messagingChannel.upsert = vi.fn(async () => ({
diff --git a/apps/api/src/messaging-inbound.ts b/apps/api/src/messaging-inbound.ts
index 0a00b7a..673b026 100644
--- a/apps/api/src/messaging-inbound.ts
+++ b/apps/api/src/messaging-inbound.ts
@@ -560,10 +560,6 @@ async function handleChannelEvent(
       ],
       skipDuplicates: true,
     });
-    await deps.prisma.messagingChannel.update({
-      where: { id: channel.id },
-      data: { introPostedAt: new Date() },
-    });
     await deps.jobs.enqueue(messagingDeliverJob()).catch((error) => {
       getLogger().error("messaging intro enqueue error", error);
     });
diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts
index 2d06a41..143b698 100644
--- a/packages/adapters/src/messaging-delivery.test.ts
+++ b/packages/adapters/src/messaging-delivery.test.ts
@@ -110,6 +110,9 @@ function createDeps(overrides: {
         ? []
         : [{ status: overrides.connectionStatus ?? "pending" }],
     ),
+    messagingChannel: {
+      updateMany: vi.fn(async () => ({ count: 1 })),
+    },
     messagingOutbound: {
       findUnique: vi.fn(async ({ where }: { where?: { id?: string } } = {}) => {
         if (overrides.existingOutbox) return overrides.existingOutbox;
@@ -339,6 +342,56 @@ describe("deliverMessagingOutbound", () => {
     expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "pending", attempts: 1 }));
   });
 
+  it("marks introPostedAt only after a pending intro row is delivered", async () => {
+    const deps = createDeps({
+      run: null,
+      outboundRows: [
+        {
+          id: "out-intro",
+          idempotencyKey: "intro:ch-1",
+          kind: "intro",
+          threadId: "sendblue:grp-1",
+          body: "Hi. This line hosts Rakazo personal agents.",
+          status: "pending",
+          providerHandle: null,
+        },
+      ],
+    });
+    await deliverMessagingOutbound(deps, {}, context);
+
+    expect(deps.sendToThread).toHaveBeenCalledWith(
+      { threadId: "sendblue:grp-1", body: "Hi. This line hosts Rakazo personal agents." },
+      context,
+    );
+    expect(deps.prisma.messagingChannel.updateMany).toHaveBeenCalledWith({
+      where: { threadId: "sendblue:grp-1", introPostedAt: null },
+      data: { introPostedAt: expect.any(Date) },
+    });
+  });
+
+  it("does not mark introPostedAt when intro delivery fails", async () => {
+    const deps = createDeps({
+      run: null,
+      sendError: new Error("provider down"),
+      outboundRows: [
+        {
+          id: "out-intro",
+          idempotencyKey: "intro:ch-1",
+          kind: "intro",
+          threadId: "sendblue:grp-1",
+          body: "Hi. This line hosts Rakazo personal agents.",
+          status: "pending",
+          attempts: 0,
+          providerHandle: null,
+        },
+      ],
+    });
+    await deliverMessagingOutbound(deps, {}, context);
+
+    expect(deps.prisma.messagingChannel.updateMany).not.toHaveBeenCalled();
+    expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "pending", attempts: 1 }));
+  });
+
   it("drains leftover pending rows without a run id", async () => {
     const deps = createDeps({
       run: null,
diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts
index b372f61..8d8522e 100644
--- a/packages/adapters/src/messaging-delivery.ts
+++ b/packages/adapters/src/messaging-delivery.ts
@@ -359,6 +359,12 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
           where: { id: row.id },
           data: { providerHandle: sent.handle },
         });
+        if (row.kind === "intro") {
+          await deps.prisma.messagingChannel.updateMany({
+            where: { threadId: row.threadId, introPostedAt: null },
+            data: { introPostedAt: new Date() },
+          });
+        }
         continue;
       }
       const identity = row.identityId