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 <cynarAI@users.noreply.github.com>

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({

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<void> {
-  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<void> {
+async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<MirrorRunResult> {
   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<vo
         block.kind === "channel_message",
     );
     if (channelBlock) {
-      await mirrorChannelRun(deps, run, channelBlock);
+      const disposition = await mirrorChannelRun(deps, run, channelBlock);
-      return;
+      return { disposition, trigger: run.trigger, status: run.status };
     }
   } else if (run.trigger !== "bot_message") {
     // Mirror inbound messaging runs and delegated bot_message replies only.
     // Do not push in-app user/routine runs out to the linked chat.
-    return;
+    return { disposition: "complete", trigger: run.trigger, status: run.status };
   }
 
   const identity = await deps.prisma.messagingIdentity.findUnique({
     where: { botId: run.botId },
   });
-  if (!identity) return;
+  if (!identity) return { disposition: "complete", trigger: run.trigger, status: run.status };
 
   // bot_message runs can carry the bot's user-facing reply after a delegate wake.
   // extractText only keeps kind:"text", so inter-bot chatter blocks never leak out.
@@ -111,10 +127,17 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<vo
       sourceMessageId: message.id,
     }))
     .filter((row) => 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<MessageBlock, { kind: "channel_message" }>,
-): Promise<void> {
+): Promise<MessagingMirrorDisposition> {
   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 {