From f9ffe06aeb58587cac83af1d819d2f665044854f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 22 Sep 2026 03:28:33 +0000 Subject: [PATCH] fix(adapters): only mark messagingMirroredAt after durable mirror mirrorMessagingOutbound always stamped messagingMirroredAt even when mirrorRun returned early, so the job reconciler never retried completed messaging runs whose channel or DM mirror was skipped or not yet durable. Gate the recovery marker on mirror disposition (complete vs retry), only set it for completed messaging runs, and leave it null when mirrorable bot text still exists but was not written to the outbox. Co-authored-by: cynarAI --- .../adapters/src/messaging-delivery.test.ts | 79 ++++++++++++++++++- packages/adapters/src/messaging-delivery.ts | 61 ++++++++++---- 2 files changed, 124 insertions(+), 16 deletions(-) diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 1a0388d..0952113 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -213,8 +213,17 @@ describe("deliverMessagingOutbound", () => { expect(deps.prisma.messagingIdentity.update).toHaveBeenCalledWith( expect.objectContaining({ data: { outboundSinceInbound: { increment: 1 } } }), ); + expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); + }); + + it("marks a completed messaging DM run after mirroring", async () => { + const deps = createDeps({ + run: { ...messagingRun, status: "completed" }, + }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ - where: { id: "run-1", trigger: "messaging" }, + where: { id: "run-1", trigger: "messaging", status: "completed" }, data: { messagingMirroredAt: expect.any(Date) }, }); }); @@ -840,6 +849,74 @@ function createChannelDeps( }; } +describe("mirrorMessagingOutbound recovery marker", () => { + it("does not mark messagingMirroredAt when a completed channel run still has mirrorable text", async () => { + const deps = createChannelDeps({ + messages: [ + { + id: "ask-1", + blocks: [{ kind: "ask", text: "Details?", status: "answered", answer: "Private detail" }], + }, + { id: "reply-1", blocks: [{ kind: "text", text: "Done — calendar updated." }] }, + ], + }); + const baseRun = await deps.prisma.run.findUnique(); + deps.prisma.run.findUnique = vi.fn(async () => ({ ...baseRun, status: "completed" })); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.rows).toEqual([]); + expect(deps.sendToThread).not.toHaveBeenCalled(); + expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); + }); + + it("does not mark messagingMirroredAt when bot messages are not durable yet", async () => { + const deps = createDeps({ + messages: [], + run: { ...messagingRun, status: "completed" }, + }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.rows).toEqual([]); + expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); + }); + + it("marks a completed messaging run after a successful mirror", async () => { + const deps = createDeps({ + run: { ...messagingRun, status: "completed" }, + }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ + where: { id: "run-1", trigger: "messaging", status: "completed" }, + data: { messagingMirroredAt: expect.any(Date) }, + }); + }); + + it("marks a completed noop when bot messages have no mirrorable text", async () => { + const deps = createDeps({ + run: { ...messagingRun, status: "completed" }, + messages: [{ id: "m-1", blocks: [{ kind: "tool_call", name: "search" }] }], + }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.sendToThread).not.toHaveBeenCalled(); + expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ + where: { id: "run-1", trigger: "messaging", status: "completed" }, + data: { messagingMirroredAt: expect.any(Date) }, + }); + }); + + it("does not mark an in-progress messaging run even after a successful mirror", async () => { + const deps = createDeps({ + run: { ...messagingRun, status: "running" }, + }); + await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + + expect(deps.sendToThread).toHaveBeenCalled(); + expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); + }); +}); + describe("deliverMessagingOutbound channel runs", () => { it("keeps privately answered runs out of the group and peer fan-out on retries", async () => { const deps = createChannelDeps({ diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 5215d92..db8d891 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -56,23 +56,39 @@ export async function deliverMessagingOutbound( await drain(deps, context); } +/** Whether a completed messaging run is safe to mark as durably mirrored. */ +type MessagingMirrorDisposition = "complete" | "retry"; + +type MirrorRunResult = { + disposition: MessagingMirrorDisposition; + trigger: string | null; + status: string | null; +}; + export async function mirrorMessagingOutbound( deps: MessagingDeliveryDeps, runId: string, ): Promise { - await mirrorRun(deps, runId); + const result = await mirrorRun(deps, runId); + if ( + result.disposition !== "complete" || + result.trigger !== "messaging" || + result.status !== "completed" + ) { + return; + } await deps.prisma.run.updateMany({ - where: { id: runId, trigger: "messaging" }, + where: { id: runId, trigger: "messaging", status: "completed" }, data: { messagingMirroredAt: new Date() }, }); } -async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise { +async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise { const run = await deps.prisma.run.findUnique({ where: { id: runId }, include: { sourceMessage: true }, }); - if (!run) return; + if (!run) return { disposition: "retry", trigger: null, status: null }; if (run.trigger === "messaging") { const sourceBlocks = (run.sourceMessage?.blocks ?? []) as MessageBlock[]; @@ -81,19 +97,19 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise row.body); - if (rows.length === 0) return; + if (rows.length === 0) { + return { + disposition: messages.length === 0 ? "retry" : "complete", + trigger: run.trigger, + status: run.status, + }; + } // Atomic dedupe: a concurrent messaging.deliver for the same run loses on // the idempotencyKey unique key instead of throwing P2002. await deps.prisma.messagingOutbound.createMany({ data: rows, skipDuplicates: true }); + return { disposition: "complete", trigger: run.trigger, status: run.status }; } /** @@ -126,15 +149,15 @@ async function mirrorChannelRun( deps: MessagingDeliveryDeps, run: { id: string; botId: string }, channelBlock: Extract, -): Promise { +): Promise { const identity = await deps.prisma.messagingIdentity.findUnique({ where: { botId: run.botId }, }); - if (!identity) return; + if (!identity) return "complete"; const channel = await deps.prisma.messagingChannel.findUnique({ where: { id: channelBlock.channelId }, }); - if (!channel) return; + if (!channel) return "complete"; const owner = await deps.prisma.user.findUnique({ where: { id: identity.userId }, select: { name: true }, @@ -155,12 +178,19 @@ async function mirrorChannelRun( (block) => block.kind === "ask" && block.status === "answered", ), ) - ) - return; + ) { + const hasMirrorableReply = replies.some( + (message) => + !(message.blocks as MessageBlock[]).some( + (block) => block.kind === "ask" && block.status === "answered", + ) && extractText(message.blocks), + ); + return hasMirrorableReply ? "retry" : "complete"; + } const messages = replies .map((message) => ({ message, text: extractText(message.blocks) })) .filter((entry) => entry.text); - if (messages.length === 0) return; + if (messages.length === 0) return replies.length === 0 ? "retry" : "complete"; await deps.prisma.messagingOutbound.createMany({ data: messages.map(({ message, text }) => ({ @@ -251,6 +281,7 @@ async function mirrorChannelRun( await deps.events.notify(peerThread.id, event.seq).catch(() => undefined); } } + return "complete"; } function escapeRegExp(value: string): string {