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 {