fix: queued MEDIA attachments survive delivery retries (#108969)

* fix: queued MEDIA attachments survive delivery retries

Local media carried in a legacy MEDIA: text directive was invisible to
queue staging (which reads only structured mediaUrl(s)), so the raw
directive was persisted with no queue-owned copy and a retry read the
already-deleted producer path — burning the retry budget and dropping the
attachment plus its text.

Project each source payload's effective media through the canonical
createOutboundPayloadPlan before staging and fold directive-derived local
sources into the queue copy's structured fields, so the existing spool
takes custody. The custody copy anchors both mediaUrl and mediaUrls to the
effective set so the staged copy overrides the preserved in-text directive
on replay; the rendered-batch plan is recomputed from the deduplicated
effective media to avoid count inflation. Raw pre-hook text, the copy-free
live send, the same media capability, and sensitive-media fail-closed are
all unchanged. No schema/config/TTL change.

Extends the ownership invariant from #108501 / #108502 to the legacy
MEDIA: text-directive carrier.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor(outbound): localize queue media custody

Trim the direct helper matrix in favor of end-to-end queue recovery proof.

Co-authored-by: Masato Hoshino <g515hoshino@gmail.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/infra/outbound/deliver.ts

@@ -1374,6 +1374,33 @@ function suppressedPayloadOutcome(params: {
   };
 }
 
+/** Adds directive-derived media to the queue copy before spool custody. */
+function materializeQueueCustodyMedia(
+  payloads: readonly ReplyPayload[],
+  plan: readonly OutboundPayloadPlan[],
+): ReplyPayload[] {
+  const effectiveBySource = new Map(
+    plan.map((entry) => [entry.sourceIndex, entry.parts.mediaUrls] as const),
+  );
+  return payloads.map((payload, index) => {
+    const effective = effectiveBySource.get(index);
+    if (!effective?.length) {
+      return payload;
+    }
+    const structured = new Set(
+      [payload.mediaUrl, ...(payload.mediaUrls ?? [])]
+        .map((url) => url?.trim())
+        .filter((url): url is string => Boolean(url)),
+    );
+    if (effective.every((url) => structured.has(url))) {
+      return payload;
+    }
+    // Keep raw pre-hook text for deterministic replay. The singular anchor
+    // prevents recovery from re-adding its original MEDIA: path.
+    return { ...payload, mediaUrl: effective[0], mediaUrls: [...effective] };
+  });
+}
+
 /**
  * @deprecated Direct outbound delivery is compatibility/runtime substrate.
  * New message lifecycle code should use `sendDurableMessageBatch` from
@@ -1428,32 +1455,59 @@ export async function deliverOutboundPayloadsInternal(
   }
   const queuePolicy = params.queuePolicy ?? "best_effort";
   const strippedQueuePayloads = payloads.map(stripInternalRuntimeScaffoldingFromPayload);
-  const queuePayloadsChanged = strippedQueuePayloads.some(
-    (payload, index) => payload !== payloads[index],
-  );
   const renderedBatchPlan =
     params.renderedBatchPlan ?? createRenderedMessageBatchPlan(params.payloads);
-  // Media staging only rewrites source URLs one-for-one, and the plan is read for
-  // its per-payload counts (see resolveMessagePlanMediaCount), so it stays keyed
-  // to the payload shape rather than to which copy the row happens to reference.
-  const queueRenderedBatchPlan = queuePayloadsChanged
-    ? createRenderedMessageBatchPlan(strippedQueuePayloads)
-    : renderedBatchPlan;
 
   const stageAndEnqueueDelivery = async (): Promise<string | null> => {
+    // Legacy `MEDIA:` text directives carry local media that only materializes
+    // into structured fields at send time, so the spool (which reads structured
+    // media) would skip it and a retry would read the vanished producer path.
+    // Project each source payload's effective media through the same canonical
+    // plan the live send uses and fold directive-derived sources into the queue
+    // copy's structured media before staging. The raw payload and its pre-hook
+    // text are untouched, so the live send below stays copy-free on the original.
+    const directiveOptions = await resolveChannelOutboundDirectiveOptions({
+      cfg: params.cfg,
+      channel,
+    });
+    const queueCustodyPayloads = materializeQueueCustodyMedia(
+      strippedQueuePayloads,
+      createOutboundPayloadPlan(strippedQueuePayloads, {
+        cfg: params.cfg,
+        sessionKey: params.session?.policyKey ?? params.session?.key,
+        surface: channel,
+        conversationType: params.session?.conversationType,
+        extractMarkdownImages: directiveOptions.extractMarkdownImages,
+      }),
+    );
+    const queuePayloadsChanged = queueCustodyPayloads.some(
+      (payload, index) => payload !== payloads[index],
+    );
+    // Media staging only rewrites source URLs one-for-one, so the plan stays keyed
+    // to the custody payload counts rather than to which copy the row references;
+    // recovery replays entry.payloads and this plan together. Materialized custody
+    // anchors mediaUrl to the effective set (to override the in-text directive on
+    // replay), so count fan-out from mediaUrls alone for payloads we rewrote to
+    // keep the plan aligned with the deduped effective media recovery re-derives.
+    const renderPlanPayloads = queueCustodyPayloads.map((payload, index) =>
+      payload === strippedQueuePayloads[index] ? payload : { ...payload, mediaUrl: undefined },
+    );
+    const queueRenderedBatchPlan = queuePayloadsChanged
+      ? createRenderedMessageBatchPlan(renderPlanPayloads)
+      : renderedBatchPlan;
     // A durable row must not outlive its media. Producer-owned local sources
     // (TTS temps above all) are deleted when this process exits, so the queue
     // takes its own copy first and the row references that; the live send below
     // keeps the original path and stays copy-free.
     const staged = await stageQueuePayloadMedia({
-      payloads: strippedQueuePayloads,
+      payloads: queueCustodyPayloads,
       // Resolved exactly as the live send resolves it: staging must neither
       // reject media the send would deliver (agent workspace sources are only
       // reachable through the agent-scoped roots) nor read more than the send may.
       mediaAccess: resolveOutboundMediaAccessForSend(
         params,
         channel,
-        collectPayloadMediaSources(strippedQueuePayloads),
+        collectPayloadMediaSources(queueCustodyPayloads),
       ),
       maxBytes: resolveOutboundMediaMaxBytes({
         cfg: params.cfg,

src/infra/outbound/delivery-queue-media-directive-durability.test.ts

@@ -0,0 +1,220 @@
+import fs from "node:fs/promises";
+import path from "node:path";
+import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
+import type {
+  ChannelOutboundAdapter,
+  ChannelOutboundContext,
+} from "../../channels/plugins/types.public.js";
+import type { OpenClawConfig } from "../../config/config.js";
+import { createEmptyPluginRegistry } from "../../plugins/registry.js";
+import {
+  releasePinnedPluginChannelRegistry,
+  setActivePluginRegistry,
+} from "../../plugins/runtime.js";
+import { createOutboundTestPlugin, createTestRegistry } from "../../test-utils/channel-plugins.js";
+import { PlatformMessageNotDispatchedError } from "./deliver-types.js";
+import { collectEntrySpoolPaths } from "./delivery-queue-media-spool.js";
+import { loadPendingDeliveries } from "./delivery-queue-storage.js";
+import { drainPendingDeliveries, type DeliverFn } from "./delivery-queue.js";
+import {
+  createRecoveryLog,
+  installDeliveryQueueTmpDirHooks,
+} from "./delivery-queue.test-helpers.js";
+
+let deliverOutboundPayloads: typeof import("./deliver.js").deliverOutboundPayloads;
+
+const cfg = {} as OpenClawConfig;
+
+function installMatrixAdapter(outbound: ChannelOutboundAdapter): void {
+  setActivePluginRegistry(
+    createTestRegistry([
+      {
+        pluginId: "matrix",
+        source: "test",
+        plugin: createOutboundTestPlugin({ id: "matrix", outbound }),
+      },
+    ]),
+  );
+}
+
+/** Adapter that records what the live send resolves, then proves nothing dispatched. */
+function enqueuePhaseAdapter(mediaPaths: string[]): ChannelOutboundAdapter {
+  return {
+    deliveryMode: "direct",
+    sendText: async () => ({ channel: "matrix", messageId: "t" }),
+    sendMedia: async (ctx: ChannelOutboundContext) => {
+      mediaPaths.push(ctx.mediaUrl ?? "");
+      // Proven-not-dispatched clears send evidence so the durable row stays replayable.
+      throw new PlatformMessageNotDispatchedError("forced not-dispatched (test)", {
+        cause: new Error("test"),
+      });
+    },
+  };
+}
+
+type RecoveredSend = { mediaUrl: string; fromSpool: boolean; bytes?: string; error?: string };
+
+/** Adapter that reads the exact path the replayed send hands it (spool copy vs producer path). */
+function recoveryPhaseAdapter(records: RecoveredSend[], spoolRoot: string): ChannelOutboundAdapter {
+  return {
+    deliveryMode: "direct",
+    sendText: async () => ({ channel: "matrix", messageId: "t" }),
+    sendMedia: async (ctx: ChannelOutboundContext) => {
+      const mediaUrl = ctx.mediaUrl ?? "";
+      const fromSpool = path.dirname(mediaUrl) === spoolRoot;
+      let bytes: string | undefined;
+      let error: string | undefined;
+      try {
+        bytes = (await fs.readFile(mediaUrl)).toString("hex");
+      } catch (err) {
+        error = String(err);
+      }
+      records.push({ mediaUrl, fromSpool, bytes, error });
+      if (error) {
+        throw new Error(error);
+      }
+      return { channel: "matrix", messageId: "recovered" };
+    },
+  };
+}
+
+describe("delivery-queue MEDIA-directive durability (end-to-end)", () => {
+  const fixtures = installDeliveryQueueTmpDirHooks();
+  let tmpDir: string;
+  let sourceDir: string;
+  let spoolRoot: string;
+  const to = "!room:example";
+  // Plain-text document: passes the host-local media send buffer verification.
+  const bytes = Buffer.from("durable media directive proof payload\n", "utf8");
+
+  beforeAll(async () => {
+    ({ deliverOutboundPayloads } = await import("./deliver.js"));
+  });
+
+  beforeEach(async () => {
+    tmpDir = fixtures.tmpDir();
+    process.env.OPENCLAW_STATE_DIR = tmpDir;
+    spoolRoot = path.join(tmpDir, "delivery-queue-media");
+    sourceDir = await fs.realpath(await fs.mkdtemp(path.join(tmpDir, "src-")));
+  });
+
+  afterEach(() => {
+    releasePinnedPluginChannelRegistry();
+    setActivePluginRegistry(createEmptyPluginRegistry());
+  });
+
+  it("replays MEDIA-directive local media from a queue-owned copy after the source is deleted", async () => {
+    const source = path.join(sourceDir, "generated.txt");
+    await fs.writeFile(source, bytes);
+
+    const payloads: ReplyPayload[] = [{ text: `caption\nMEDIA:${source}` }];
+    const inputSnapshot = structuredClone(payloads);
+
+    const liveMediaPaths: string[] = [];
+    installMatrixAdapter(enqueuePhaseAdapter(liveMediaPaths));
+
+    await expect(
+      deliverOutboundPayloads({
+        cfg,
+        channel: "matrix",
+        to,
+        payloads,
+        queuePolicy: "required",
+        mediaAccess: { localRoots: [sourceDir] },
+      }),
+    ).rejects.toThrow("forced not-dispatched");
+
+    // Live send stays copy-free on the original producer path.
+    expect(liveMediaPaths).toEqual([source]);
+    // Input object is not mutated.
+    expect(payloads).toEqual(inputSnapshot);
+
+    // The durable row references a queue-owned spool copy, not the producer path.
+    const [entry] = await loadPendingDeliveries(tmpDir);
+    expect(entry).toBeDefined();
+    // mediaUrl and mediaUrls[0] both anchor to the one staged copy, so dedupe.
+    const spoolPaths = [...new Set(collectEntrySpoolPaths(entry?.payloads ?? [], tmpDir))];
+    expect(spoolPaths).toHaveLength(1);
+    expect(path.dirname(spoolPaths[0] ?? "")).toBe(spoolRoot);
+    // Raw pre-hook text (directive included) is preserved on the row.
+    expect(entry?.payloads[0]?.text).toBe(`caption\nMEDIA:${source}`);
+    // Spool bytes equal source bytes.
+    expect(await fs.readFile(spoolPaths[0] ?? "")).toEqual(bytes);
+
+    // Producer source disappears (process exit / TTS temp cleanup).
+    await fs.rm(source, { force: true });
+    await expect(fs.readFile(source)).rejects.toThrow();
+
+    // Fresh-process recovery replays the row.
+    const recovered: RecoveredSend[] = [];
+    installMatrixAdapter(recoveryPhaseAdapter(recovered, spoolRoot));
+    const deliver = vi.fn<DeliverFn>(async (params) => deliverOutboundPayloads(params));
+    await drainPendingDeliveries({
+      drainKey: "media-directive-test",
+      logLabel: "media-directive drain",
+      cfg,
+      log: createRecoveryLog(),
+      stateDir: tmpDir,
+      deliver,
+      selectEntry: (e) => ({ match: e.channel === "matrix", bypassBackoff: true }),
+    });
+
+    // Recovery delivered from the queue-owned copy, not the deleted producer path.
+    expect(recovered).toHaveLength(1);
+    expect(recovered[0]?.fromSpool).toBe(true);
+    expect(recovered[0]?.error).toBeUndefined();
+    expect(recovered[0]?.bytes).toBe(bytes.toString("hex"));
+    expect(recovered[0]?.mediaUrl).not.toBe(source);
+    // Terminal ack clears the durable row.
+    expect(await loadPendingDeliveries(tmpDir)).toHaveLength(0);
+  });
+
+  it("fails a required send closed for sensitive directive media (no row, no spool)", async () => {
+    const source = path.join(sourceDir, "secret.txt");
+    await fs.writeFile(source, bytes);
+    installMatrixAdapter(enqueuePhaseAdapter([]));
+
+    await expect(
+      deliverOutboundPayloads({
+        cfg,
+        channel: "matrix",
+        to,
+        payloads: [{ text: `MEDIA:${source}`, sensitiveMedia: true }],
+        queuePolicy: "required",
+        mediaAccess: { localRoots: [sourceDir] },
+      }),
+    ).rejects.toThrow(/cannot be persisted|unsupported/i);
+
+    expect(await loadPendingDeliveries(tmpDir)).toHaveLength(0);
+    await expect(fs.readdir(spoolRoot)).rejects.toThrow(); // spool dir never created
+  });
+
+  it("does not spool directive media when the queue is skipped", async () => {
+    const source = path.join(sourceDir, "skip.txt");
+    await fs.writeFile(source, bytes);
+    const liveMediaPaths: string[] = [];
+    installMatrixAdapter({
+      deliveryMode: "direct",
+      sendText: async () => ({ channel: "matrix", messageId: "t" }),
+      sendMedia: async (ctx: ChannelOutboundContext) => {
+        liveMediaPaths.push(ctx.mediaUrl ?? "");
+        return { channel: "matrix", messageId: "sent" };
+      },
+    });
+
+    await deliverOutboundPayloads({
+      cfg,
+      channel: "matrix",
+      to,
+      payloads: [{ text: `MEDIA:${source}` }],
+      skipQueue: true,
+      mediaAccess: { localRoots: [sourceDir] },
+    });
+
+    // Live send used the original path; nothing was queued or spooled.
+    expect(liveMediaPaths).toEqual([source]);
+    expect(await loadPendingDeliveries(tmpDir)).toHaveLength(0);
+    await expect(fs.readdir(spoolRoot)).rejects.toThrow();
+  });
+});