diff --git a/README.md b/README.md
index 3cd41f2..08f76e7 100644
--- a/README.md
+++ b/README.md
@@ -108,6 +108,10 @@ Security defaults worth knowing:
 - `dmPolicy: "open"` requires `allowFrom` to include `"*"`.
 - Every reply re-hydrates the triggering message and re-authorizes its `From`, so an untrusted `Reply-To` cannot redirect delivery.
 
+The configured `inboxId` also identifies the durable receive queue. Keep its casing stable across
+upgrades: changing only letter case can create a new queue identity, so messages covered only by
+older completion tombstones may be dispatched once more during the migration.
+
 ## CLI-backed skill
 
 The plugin registers a passthrough command:
diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts
index f0c5066..e75fb42 100644
--- a/src/channel/src/catch-up.test.ts
+++ b/src/channel/src/catch-up.test.ts
@@ -427,7 +427,7 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
-  it("does not advance the cursor beyond the sweep's fixed upper bound", async () => {
+  it("does not advance the cursor from an invalid timestamp's local fallback", async () => {
     const store = memoryStore<never>();
     const malformed = message({ id: "bad_timestamp", timestamp: 1_100 }) as AgentMail.MessageItem;
     (malformed as { timestamp: unknown }).timestamp = new Date(Number.NaN);
@@ -454,7 +454,7 @@ describe("AgentMail durable REST catch-up", () => {
     expect(list).toHaveBeenLastCalledWith(
       "inbox_1",
       expect.objectContaining({
-        after: new Date(2_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS),
+        after: new Date(1_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS),
       }),
       expect.any(Object),
     );
@@ -495,7 +495,7 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
-  it("repairs a stored cursor after the host clock moves backwards", async () => {
+  it("keeps committed cursor state when the host clock moves backwards", async () => {
     const store = memoryStore<AgentMailCatchUpCursor>();
     const key = sha256Hex("default\ninbox_1");
     await store.register(key, {
@@ -524,12 +524,41 @@ describe("AgentMail durable REST catch-up", () => {
       expect.any(Object),
     );
     expect(await store.lookup(key)).toMatchObject({
-      baselineAtMs: 1_000_000,
-      highWaterAtMs: 1_000_000,
+      baselineAtMs: 2_000_000,
+      highWaterAtMs: 3_000_000,
       established: true,
     });
   });
 
+  it("does not invert baseline catch-up bounds when the clock moves backward", async () => {
+    const store = memoryStore<never>();
+    const list = vi.fn(async () => ({ count: 0, messages: [] }));
+    let nowMs = 2_000_000;
+    const session = await createAgentMailCatchUpSession({
+      account,
+      client: { inboxes: { messages: { list } } } as never,
+      store: store as never,
+      now: () => nowMs,
+    });
+    nowMs = 1_000_000;
+
+    await session.run({ receive: vi.fn(), abortSignal: new AbortController().signal });
+    await session.run({
+      receive: vi.fn(),
+      abortSignal: new AbortController().signal,
+      sinceBaseline: true,
+    });
+
+    for (const [, query] of list.mock.calls) {
+      expect(query).toEqual(
+        expect.objectContaining({
+          after: new Date(nowMs),
+          before: new Date(nowMs + 1),
+        }),
+      );
+    }
+  });
+
   it("lists from the monitoring baseline on a deep sweep", async () => {
     const store = memoryStore<never>();
     const list = vi.fn(async () => ({
@@ -605,7 +634,7 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
-  it("advances past malformed timestamps and clamps implausibly future timestamps", async () => {
+  it("does not advance from malformed timestamps or admit implausibly future timestamps", async () => {
     const store = memoryStore<AgentMailCatchUpCursor>();
     const key = sha256Hex("default\ninbox_1");
     await store.register(key, {
@@ -638,7 +667,6 @@ describe("AgentMail durable REST catch-up", () => {
 
     expect(receive.mock.calls.map(([record]) => [record.messageId, record.receivedAt])).toEqual([
       ["malformed", 1_000_000],
-      ["future", 1_000_000],
     ]);
     expect(list).toHaveBeenNthCalledWith(
       1,
@@ -652,12 +680,12 @@ describe("AgentMail durable REST catch-up", () => {
       2,
       "inbox_1",
       expect.objectContaining({
-        after: new Date(1_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS),
+        after: new Date(0),
       }),
       expect.any(Object),
     );
     expect(warn).toHaveBeenCalledWith(expect.stringContaining("invalid timestamp"));
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining("clamped future timestamp"));
+    expect(warn).toHaveBeenCalledWith(expect.stringContaining("too far in the future"));
   });
 
   it("pauses the pass when durable ingress reports capacity", async () => {
diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts
index 235437d..20c1285 100644
--- a/src/channel/src/catch-up.ts
+++ b/src/channel/src/catch-up.ts
@@ -10,6 +10,7 @@ import {
 } from "./durable-receive.js";
 import {
   AGENTMAIL_RECEIVED_LABEL,
+  isAgentMailProviderTimestampWithinFutureSkew,
   isReceivedAgentMailMessage,
   resolveReceivedAgentMailMessageTimestampMs,
 } from "./received-message.js";
@@ -236,19 +237,15 @@ function mergeCursor(
   checkpoint: AgentMailCatchUpCheckpoint | null,
 ): AgentMailCatchUpCursor {
   const current = normalizeCursor(currentValue);
-  const baselineAtMs = Math.min(
-    upperBoundAtMs,
-    current?.baselineAtMs ?? next.baselineAtMs,
-    next.baselineAtMs,
-  );
+  const baselineAtMs = current?.baselineAtMs ?? next.baselineAtMs;
   return {
     version: CURSOR_VERSION,
     baselineAtMs,
-    // A wall-clock rollback must be allowed to lower a cursor that is now in the future. Within
-    // the current clock horizon, concurrent updates still merge monotonically.
+    // Clamp only this run's contribution. A backward-moving clock must not lower a committed
+    // cursor; query-time effective bounds below keep provider ranges valid until the clock recovers.
     highWaterAtMs: Math.max(
       baselineAtMs,
-      Math.min(upperBoundAtMs, current?.highWaterAtMs ?? 0),
+      current?.highWaterAtMs ?? 0,
       Math.min(upperBoundAtMs, next.highWaterAtMs),
     ),
     established: current?.established === true || next.established,
@@ -405,8 +402,8 @@ export async function createAgentMailCatchUpSession(params: {
       const sweepAtMs = resumableCheckpoint
         ? resumableCheckpoint.beforeAtMs - 1
         : runAtMs;
-      // If the host clock moved backwards, both stored bounds can be in the future. Clamp the
-      // effective cursor for this run and persist the repaired value after a successful sweep.
+      // If the host clock moved backwards, both stored bounds can be in the future. Clamp only the
+      // effective query cursor for this run; committed cursor state remains monotonic.
       const effectiveBaselineAtMs = Math.min(storedCursor.baselineAtMs, sweepAtMs);
       const effectiveHighWaterAtMs = Math.max(
         effectiveBaselineAtMs,
@@ -456,22 +453,32 @@ export async function createAgentMailCatchUpSession(params: {
             message,
             params.account.inboxId,
           );
+          if (
+            providerReceivedAt !== null &&
+            !isAgentMailProviderTimestampWithinFutureSkew(
+              providerReceivedAt,
+              sweepAtMs,
+            )
+          ) {
+            params.log?.warn?.(
+              `AgentMail catch-up ignored message ${message.messageId} timestamped too far in the future`,
+            );
+            continue;
+          }
           const arrivedAt = now();
-          const receivedAt = providerReceivedAt !== null
-            ? Math.min(providerReceivedAt, sweepAtMs)
-            : sweepAtMs;
+          const receivedAt = providerReceivedAt ?? sweepAtMs;
           if (providerReceivedAt === null) {
-            // Use local arrival time so a malformed newest row is admitted once and advances the
-            // cursor instead of being re-listed and warned about forever.
+            // Invalid provider time must not turn a missed event into a permanent drop. Admit with
+            // local observation time, but do not use that synthetic value as cursor evidence.
             params.log?.warn?.(
               `AgentMail catch-up used arrival time for message ${message.messageId} with an invalid timestamp`,
             );
           } else if (providerReceivedAt > sweepAtMs) {
-            // Provider clock skew must not push the cursor into the far future and disable periodic
-            // overlap recovery. Preserve the row while clamping its ordering timestamp to this
-            // sweep's fixed wall-clock bound.
+            // A bounded amount of provider skew is retained on the durable row so its completion
+            // tombstone survives until the message enters provider-time scans. Cursor advancement
+            // remains clamped to this sweep below.
             params.log?.warn?.(
-              `AgentMail catch-up clamped future timestamp for message ${message.messageId}`,
+              `AgentMail catch-up retained bounded future timestamp for message ${message.messageId}`,
             );
           }
           try {
@@ -496,7 +503,12 @@ export async function createAgentMailCatchUpSession(params: {
             throw error;
           }
           admitted += 1;
-          highWaterAtMs = Math.max(highWaterAtMs, receivedAt);
+          if (providerReceivedAt !== null) {
+            highWaterAtMs = Math.max(
+              highWaterAtMs,
+              Math.min(providerReceivedAt, sweepAtMs),
+            );
+          }
         }
         pageCursor = page.nextPageToken;
         if (pageCursor) {
diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts
index ec8dbef..673756c 100644
--- a/src/channel/src/durable-receive.test.ts
+++ b/src/channel/src/durable-receive.test.ts
@@ -7,6 +7,7 @@ import {
   createAgentMailDurableInboundId,
   withAgentMailIngressCapacity,
 } from "./durable-receive.js";
+import { AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS } from "./received-message.js";
 import {
   AgentMailIngressCapacityError,
   processAgentMailIngress,
@@ -62,6 +63,102 @@ describe("AgentMail durable ingress", () => {
     expect(deletePending).toHaveBeenCalledWith(id);
   });
 
+  it("coordinates capacity across journal facades for the same queue", async () => {
+    const pendingRows = new Map<string, { id: string; payload: AgentMailIngressRecord }>();
+    const baseJournal = () =>
+      ({
+        accept: vi.fn(async (id: string, payload: AgentMailIngressRecord) => {
+          const existing = pendingRows.get(id);
+          if (existing) {
+            return {
+              kind: "pending",
+              duplicate: true,
+              record: { ...existing, attempts: 0 },
+            };
+          }
+          const accepted = { id, payload };
+          pendingRows.set(id, accepted);
+          return { kind: "accepted", duplicate: false, record: accepted };
+        }),
+        pending: vi.fn(async () =>
+          [...pendingRows.values()].map((item) => ({ ...item, attempts: 0 })),
+        ),
+        complete: vi.fn(),
+        release: vi.fn(),
+        deletePending: vi.fn(async (id: string) => pendingRows.delete(id)),
+      }) as never;
+    const coordinationKey = `capacity-test-${crypto.randomUUID()}`;
+    const first = withAgentMailIngressCapacity(baseJournal(), 3, coordinationKey);
+    const second = withAgentMailIngressCapacity(baseJournal(), 3, coordinationKey);
+
+    await first.accept("message_1", record);
+    await second.accept("message_2", record);
+    const admissions = await Promise.allSettled([
+      first.accept("message_3", record),
+      second.accept("message_4", record),
+    ]);
+
+    expect(admissions.filter((result) => result.status === "fulfilled")).toHaveLength(1);
+    expect(
+      admissions.filter(
+        (result) =>
+          result.status === "rejected" &&
+          result.reason instanceof AgentMailIngressCapacityError,
+      ),
+    ).toHaveLength(1);
+    expect(pendingRows).toHaveLength(3);
+  });
+
+  it("resynchronizes capacity after accept persists a row and then throws", async () => {
+    const pendingRows = new Map<string, { id: string; payload: AgentMailIngressRecord }>();
+    let failAfterPersist = true;
+    const journal = withAgentMailIngressCapacity(
+      {
+        accept: vi.fn(async (id: string, payload: AgentMailIngressRecord) => {
+          const existing = pendingRows.get(id);
+          if (existing) {
+            return {
+              kind: "pending",
+              duplicate: true,
+              record: { ...existing, attempts: 0 },
+            };
+          }
+          const accepted = { id, payload };
+          pendingRows.set(id, accepted);
+          if (id === "message_3" && failAfterPersist) {
+            failAfterPersist = false;
+            throw new Error("post-enqueue prune failed");
+          }
+          return { kind: "accepted", duplicate: false, record: accepted };
+        }),
+        pending: vi.fn(async () =>
+          [...pendingRows.values()].map((item) => ({ ...item, attempts: 0 })),
+        ),
+        complete: vi.fn(),
+        release: vi.fn(),
+        deletePending: vi.fn(async (id: string) => pendingRows.delete(id)),
+      } as never,
+      3,
+      `partial-admission-test-${crypto.randomUUID()}`,
+    );
+
+    await journal.accept("message_1", record);
+    await journal.accept("message_2", record);
+    await expect(journal.accept("message_3", record)).rejects.toThrow(
+      "post-enqueue prune failed",
+    );
+    // Duplicate redelivery does not consume capacity and must remain dispatchable.
+    await expect(journal.accept("message_3", record)).resolves.toMatchObject({
+      kind: "pending",
+    });
+    // The next new row forces a durable recount, rolls itself back, and preserves the cap.
+    await expect(journal.accept("message_4", record)).rejects.toBeInstanceOf(
+      AgentMailIngressCapacityError,
+    );
+    expect(pendingRows).toHaveLength(3);
+    expect(pendingRows.has("message_4")).toBe(false);
+  });
+
   it("keeps an overflow row retryable when rollback deletion fails", async () => {
     const id = createAgentMailDurableInboundId(record);
     const fail = vi.fn(async () => true);
@@ -99,7 +196,12 @@ describe("AgentMail durable ingress", () => {
     await expect(
       processAgentMailIngress({ journal: journal as never, record, dispatch }),
     ).resolves.toBe("accepted");
-    await vi.waitFor(() => expect(complete).toHaveBeenCalledWith(id));
+    await vi.waitFor(() =>
+      expect(complete).toHaveBeenCalledWith(
+        id,
+        expect.objectContaining({ completedAt: expect.any(Number) }),
+      ),
+    );
     expect(dispatch).toHaveBeenCalledOnce();
     expect(fail).not.toHaveBeenCalled();
   });
@@ -205,6 +307,32 @@ describe("AgentMail durable ingress", () => {
     expect(order).toEqual(["accept", "dispatch", "complete"]);
   });
 
+  it("retains future-dated completion tombstones through the provider scan horizon", async () => {
+    const now = 1_000_000;
+    const futureReceivedAt = now + 365 * 24 * 60 * 60 * 1000;
+    const dateNow = vi.spyOn(Date, "now").mockReturnValue(now);
+    const complete = vi.fn(async () => undefined);
+    try {
+      await processAgentMailIngress({
+        journal: {
+          accept: async () => ({ kind: "accepted", duplicate: false, record: {} }),
+          complete,
+          release: vi.fn(),
+        } as never,
+        record: { ...record, receivedAt: futureReceivedAt },
+        dispatch: vi.fn(async () => undefined),
+      });
+
+      await vi.waitFor(() =>
+        expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), {
+          completedAt: now + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS,
+        }),
+      );
+    } finally {
+      dateNow.mockRestore();
+    }
+  });
+
   it("returns after durable admission without waiting for agent dispatch", async () => {
     let finishDispatch!: () => void;
     const dispatch = vi.fn(
@@ -831,7 +959,13 @@ describe("AgentMail durable ingress", () => {
       record,
       expect.objectContaining({ onTurnAdopted: expect.any(Function) }),
     );
-    expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record));
+    expect(complete).toHaveBeenCalledWith(
+      createAgentMailDurableInboundId(record),
+      expect.objectContaining({ completedAt: expect.any(Number) }),
+    );
+    // Let the completed dispatch remove itself from the process-wide active-dispatch registry
+    // before exercising a fresh pending row with the same durable id.
+    await new Promise((resolve) => setTimeout(resolve, 0));
 
     dispatch.mockClear();
     complete.mockClear();
@@ -849,7 +983,10 @@ describe("AgentMail durable ingress", () => {
       dispatch,
     });
     await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce());
-    expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record));
+    expect(complete).toHaveBeenCalledWith(
+      createAgentMailDurableInboundId(record),
+      expect.objectContaining({ completedAt: expect.any(Number) }),
+    );
   });
 
   it("keeps completed dedupe for the full recovery horizon without an entry cap", () => {
@@ -876,7 +1013,12 @@ describe("AgentMail durable ingress", () => {
       } as never,
       dispatch,
     });
-    await vi.waitFor(() => expect(complete).toHaveBeenCalledWith("durable_1"));
+    await vi.waitFor(() =>
+      expect(complete).toHaveBeenCalledWith(
+        "durable_1",
+        expect.objectContaining({ completedAt: expect.any(Number) }),
+      ),
+    );
     expect(dispatch).toHaveBeenCalledWith(
       record,
       expect.objectContaining({ onTurnAdopted: expect.any(Function) }),
@@ -1001,7 +1143,10 @@ describe("AgentMail durable ingress", () => {
     firstAbort.abort();
     rejectFirst(new Error("old account stopped"));
     await vi.waitFor(() => expect(replacementDispatch).toHaveBeenCalledOnce());
-    expect(replacementComplete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record));
+    expect(replacementComplete).toHaveBeenCalledWith(
+      createAgentMailDurableInboundId(record),
+      expect.objectContaining({ completedAt: expect.any(Number) }),
+    );
     replacementAbort.abort();
   });
 });
diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts
index 6ed092a..00e26cc 100644
--- a/src/channel/src/durable-receive.ts
+++ b/src/channel/src/durable-receive.ts
@@ -48,6 +48,31 @@ type AgentMailJournal = ReturnType<
   ) => Promise<boolean>;
 };
 
+type AgentMailCapacityAdmissionState = {
+  admissionChain: Promise<unknown>;
+  pendingEstimate: number | null;
+};
+
+// Account reloads can briefly leave old and replacement journal facades open over the same
+// persisted queue. Keep their count-and-accept sequence on one chain, keyed by that queue's stable
+// storage identity, rather than coordinating only calls made through one facade.
+const capacityAdmissionStates = new Map<string, AgentMailCapacityAdmissionState>();
+
+function resolveCapacityAdmissionState(
+  coordinationKey: string | undefined,
+): AgentMailCapacityAdmissionState {
+  if (!coordinationKey) {
+    return { admissionChain: Promise.resolve(), pendingEstimate: null };
+  }
+  const existing = capacityAdmissionStates.get(coordinationKey);
+  if (existing) {
+    return existing;
+  }
+  const created = { admissionChain: Promise.resolve(), pendingEstimate: null };
+  capacityAdmissionStates.set(coordinationKey, created);
+  return created;
+}
+
 /**
  * Wraps the store-backed journal with an admission cap. The published SDK's queue journal only
  * evicts on capacity; durable email must instead reject NEW mail while keeping already-accepted
@@ -56,47 +81,62 @@ type AgentMailJournal = ReturnType<
 export function withAgentMailIngressCapacity(
   journal: AgentMailJournal,
   maxPendingEntries: number,
+  coordinationKey?: string,
 ): AgentMailJournal {
   // Serialize check-and-enqueue. The published queue journal has no atomic admission cap, so two
   // concurrent transports (live WebSocket + REST catch-up) could otherwise both observe free space
-  // and push past the bound. Chaining admissions makes the count-then-accept step atomic per
-  // journal; the chain never rejects so one failed admission cannot poison later ones.
-  let admissionChain: Promise<unknown> = Promise.resolve();
+  // and push past the bound. Chaining admissions makes the count-then-accept step atomic across
+  // every facade for the queue; the chain never rejects so one failed admission cannot poison
+  // later ones. This intentionally introduces brief head-of-line coupling for facades sharing one
+  // queue; each critical section contains only the queue admission and occasional capacity scan.
+  const state = resolveCapacityAdmissionState(coordinationKey);
   // Upper-bound estimate of the pending count. It only increases on new admissions (never on
   // completion/retention pruning), so the O(pending) scan is skipped on the common below-cap path
   // and only runs to re-sync from the source of truth when the estimate first reaches the cap.
-  let pendingEstimate: number | null = null;
   return {
     ...journal,
     accept: (id, payload, options) => {
-      const admission = admissionChain.then(async () => {
-        const accepted = await journal.accept(id, payload, options);
-        // Let the queue perform its atomic id lookup first. Completed tombstones and pending
-        // duplicates consume no new capacity and must remain harmless even while the queue is full.
-        if (accepted.kind !== "accepted") {
-          return accepted;
-        }
-        if (pendingEstimate === null || pendingEstimate >= maxPendingEntries) {
-          const pending = await journal.pending();
-          pendingEstimate = pending.length;
-          if (pendingEstimate > maxPendingEntries) {
-            try {
-              if (await journal.deletePending(id)) {
-                pendingEstimate -= 1;
+      const admission = state.admissionChain.then(async () => {
+        try {
+          const accepted = await journal.accept(id, payload, options);
+          // Let the queue perform its atomic id lookup first. Completed tombstones and pending
+          // duplicates consume no new capacity and must remain harmless even while the queue is
+          // full.
+          if (accepted.kind !== "accepted") {
+            return accepted;
+          }
+          if (
+            state.pendingEstimate === null ||
+            state.pendingEstimate >= maxPendingEntries
+          ) {
+            const pending = await journal.pending();
+            state.pendingEstimate = pending.length;
+            if (state.pendingEstimate > maxPendingEntries) {
+              try {
+                if (await journal.deletePending(id)) {
+                  state.pendingEstimate -= 1;
+                }
+              } catch {
+                // Never turn a capacity rejection into a terminal tombstone. If rollback deletion
+                // is temporarily unavailable, preserve the accepted pending row: provider
+                // redelivery or REST catch-up re-admits that duplicate and dispatches it once
+                // capacity recovers.
               }
-            } catch {
-              // Never turn a capacity rejection into a terminal tombstone. If rollback deletion is
-              // temporarily unavailable, preserve the accepted pending row: provider redelivery or
-              // REST catch-up re-admits that duplicate and dispatches it once capacity recovers.
+              throw new AgentMailIngressCapacityError();
             }
-            throw new AgentMailIngressCapacityError();
+          } else {
+            state.pendingEstimate += 1;
           }
-        } else {
-          pendingEstimate += 1;
+          return accepted;
+        } catch (error) {
+          // accept() may persist the row and then fail during its post-enqueue retention prune.
+          // Any failure after entering the critical section makes the cached count uncertain; force
+          // the next new admission to resynchronize from durable storage.
+          state.pendingEstimate = null;
+          throw error;
         }
-        return accepted;
       });
-      admissionChain = admission.then(
+      state.admissionChain = admission.then(
         () => undefined,
         () => undefined,
       );
@@ -110,10 +150,12 @@ export function createAgentMailDurableInboundReceiveJournal(params: {
   inboxId: string;
 }): AgentMailJournal {
   const runtime = getAgentMailRuntime();
+  const stateDir = runtime.state.resolveStateDir();
+  const queueAccountId = sha256Hex(`${params.accountId}\n${params.inboxId}`).slice(0, 24);
   const queue = runtime.state.openChannelIngressQueue<AgentMailIngressRecord, undefined, undefined>(
     {
-      accountId: sha256Hex(`${params.accountId}\n${params.inboxId}`).slice(0, 24),
-      stateDir: runtime.state.resolveStateDir(),
+      accountId: queueAccountId,
+      stateDir,
     },
   );
   const prune = async () => {
@@ -132,7 +174,7 @@ export function createAgentMailDurableInboundReceiveJournal(params: {
         await prune();
         acceptsSincePrune = 0;
       }
-      const result = await queue.enqueue(id.trim(), payload, options);
+      const result = await queue.enqueue(id, payload, options);
       acceptsSincePrune += 1;
       if (result.kind === "accepted") {
         return { kind: "accepted", duplicate: false, record: result.record };
@@ -173,5 +215,6 @@ export function createAgentMailDurableInboundReceiveJournal(params: {
   return withAgentMailIngressCapacity(
     extendedJournal,
     AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES,
+    `${stateDir}\nagentmail\n${queueAccountId}`,
   );
 }
diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts
index 985ab39..6f44f7f 100644
--- a/src/channel/src/inbound.test.ts
+++ b/src/channel/src/inbound.test.ts
@@ -256,8 +256,15 @@ describe("AgentMail REST-authoritative inbound", () => {
         replyToId: "message_1",
         replyToTag: false,
         replyToCurrent: true,
+        channelData: { existing: true, agentmail: { existingAgentMail: true } },
       }),
-    ).toEqual({ text: "reply" });
+    ).toEqual({
+      text: "reply",
+      channelData: {
+        existing: true,
+        agentmail: { existingAgentMail: true, triggerArrivedAt: 1 },
+      },
+    });
     expect(delivery.durable()).toMatchObject({
       to: "message:message_1",
       replyToId: "message_1",
@@ -337,13 +344,16 @@ describe("AgentMail REST-authoritative inbound", () => {
     expect(onTurnAbandoned).toHaveBeenCalledOnce();
   });
 
-  it("cleans up deferred-turn attachments when account shutdown aborts the queued turn", async () => {
+  it("does not delete core-owned deferred attachments when account shutdown aborts", async () => {
     rm.mockClear();
     loadAgentMailInboundAttachments.mockResolvedValueOnce({
       paths: ["/tmp/deferred-abort.bin"],
       types: ["application/octet-stream"],
     });
     const controller = new AbortController();
+    let lifecycle:
+      | { onDeferred: () => void; onAbandoned: () => Promise<void> }
+      | undefined;
     await dispatchAgentMailInboundEvent({
       cfg: {},
       account,
@@ -355,9 +365,10 @@ describe("AgentMail REST-authoritative inbound", () => {
           run: async ({
             turnAdoptionLifecycle,
           }: {
-            turnAdoptionLifecycle: { onDeferred: () => void };
+            turnAdoptionLifecycle: typeof lifecycle;
           }) => {
-            turnAdoptionLifecycle.onDeferred();
+            lifecycle = turnAdoptionLifecycle;
+            turnAdoptionLifecycle?.onDeferred();
           },
         },
         session: { resolveStorePath: () => "/tmp/s.json", recordInboundSession: vi.fn() },
@@ -369,9 +380,11 @@ describe("AgentMail REST-authoritative inbound", () => {
     expect(rm).not.toHaveBeenCalled();
 
     controller.abort();
-    await vi.waitFor(() =>
-      expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true }),
-    );
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(rm).not.toHaveBeenCalled();
+
+    await lifecycle?.onAbandoned();
+    expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true });
   });
 
   it("cleans up attachments when inbound handling resolves without adoption", async () => {
diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts
index a503f86..e3272c0 100644
--- a/src/channel/src/inbound.ts
+++ b/src/channel/src/inbound.ts
@@ -13,6 +13,7 @@ import {
   agentMailInboxIdsEqual,
   resolveAgentMailTimestampMs,
 } from "./received-message.js";
+import { withAgentMailTriggerArrival } from "./reply-metadata.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
 
 const CHANNEL_ID = "agentmail";
@@ -277,7 +278,7 @@ export async function dispatchAgentMailInboundEvent(params: {
   let detachAbortCleanup = () => {};
   if (params.abortSignal) {
     const onAbort = () => {
-      if (!turnAdoptionObserved) {
+      if (!turnAdoptionObserved && !turnDeferred) {
         void cleanupInboundMedia();
       }
     };
@@ -394,7 +395,10 @@ export async function dispatchAgentMailInboundEvent(params: {
             // implicit id owned by the durable delivery contract below, so remove those derived
             // payload directives immediately before delivery.
             preparePayload: (payload) => {
-              const prepared = { ...payload };
+              const prepared = withAgentMailTriggerArrival(
+                payload,
+                params.record.arrivedAt ?? params.record.receivedAt,
+              );
               delete prepared.replyToId;
               delete prepared.replyToTag;
               delete prepared.replyToCurrent;
diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts
index 6378c94..316da5b 100644
--- a/src/channel/src/ingress.ts
+++ b/src/channel/src/ingress.ts
@@ -2,18 +2,14 @@ import { type AgentMailLog, errorText } from "./log.js";
 import type { createAgentMailDurableInboundReceiveJournal } from "./durable-receive.js";
 import { AgentMailIngressCapacityError, createAgentMailDurableInboundId } from "./durable-receive.js";
 import { HYDRATION_NOT_FOUND_RETRY_WINDOW_MS } from "./inbound.js";
+import { capAgentMailProviderTimestampForRetention } from "./received-message.js";
 import { createBackoff, waitForRetry } from "./retry.js";
 import type { AgentMailIngressRecord } from "./types.js";
 
 // Re-exported for transports (websocket) that catch capacity backpressure by class.
 export { AgentMailIngressCapacityError };
 
-type AgentMailJournal = ReturnType<typeof createAgentMailDurableInboundReceiveJournal> & {
-  fail?: (
-    id: string,
-    options: { reason: string; message?: string; failedAt?: number },
-  ) => Promise<boolean>;
-};
+type AgentMailJournal = ReturnType<typeof createAgentMailDurableInboundReceiveJournal>;
 
 type DispatchParams = {
   journal: AgentMailJournal;
@@ -79,6 +75,28 @@ const activeDispatches = new Map<string, ActiveDispatch>();
 
 const retryDelayMs = createBackoff(30 * 60_000);
 
+async function completeAgentMailIngress(params: {
+  journal: AgentMailJournal;
+  id: string;
+  record: AgentMailIngressRecord;
+}): Promise<void> {
+  // REST catch-up scans by provider timestamp. A live message stamped in the future must retain its
+  // tombstone until seven days after that provider time, or it can re-enter the scan window only
+  // after a locally-timestamped completion marker has already expired.
+  const completedAt = Date.now();
+  const rawProviderReceivedAt =
+    Number.isFinite(params.record.receivedAt) && params.record.receivedAt >= 0
+      ? params.record.receivedAt
+      : completedAt;
+  const providerReceivedAt = capAgentMailProviderTimestampForRetention(
+    rawProviderReceivedAt,
+    completedAt,
+  );
+  await params.journal.complete(params.id, {
+    completedAt: Math.max(completedAt, providerReceivedAt),
+  });
+}
+
 // True when a dispatch failure is a provider-projection race: either a 404 (message not yet
 // REST-visible) or a not-yet-projected `received` label. Both resolve on their own within seconds.
 function isHydrationRetryable(error: unknown): boolean {
@@ -235,7 +253,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
         adoptionTask = (async () => {
           while (!params.abortSignal?.aborted) {
             try {
-              await params.journal.complete(params.id);
+              await completeAgentMailIngress(params);
               params.dispatchCompleted = true;
               turnAdopted = true;
               settleDeferred("adopted");
@@ -321,6 +339,11 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
           return false;
         }
         if (outcome === "completion-failed") {
+          // Core may already have adopted the turn, but neither the completion marker nor the
+          // terminal failure marker could be written. Leaving the row pending preserves recovery;
+          // without any durable marker, a restart cannot distinguish adoption from non-adoption
+          // and may replay the turn. That duplicate window is unavoidable during a total marker
+          // store outage.
           return false;
         }
         turnAbandoned = true;
@@ -340,7 +363,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
                 message: lastError,
               });
             } else {
-              await params.journal.complete(params.id);
+              await completeAgentMailIngress(params);
             }
           } catch {
             // Best effort: TTL pruning still reclaims the row if the terminal marker cannot
@@ -389,7 +412,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
                 message: errorText(dispatchError),
               });
             } else {
-              await params.journal.complete(params.id);
+              await completeAgentMailIngress(params);
             }
           } catch {
             // Best effort: TTL pruning still reclaims the row if the terminal marker cannot persist.
@@ -431,7 +454,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
       params.dispatchCompleted = true;
     }
     try {
-      await params.journal.complete(params.id);
+      await completeAgentMailIngress(params);
       return true;
     } catch {
       completionAttempts += 1;
diff --git a/src/channel/src/media.ts b/src/channel/src/media.ts
index 655c687..9518397 100644
--- a/src/channel/src/media.ts
+++ b/src/channel/src/media.ts
@@ -62,6 +62,8 @@ export async function loadAgentMailInboundAttachments(params: {
   }
   let declaredBytes = 0;
   for (const attachment of accepted) {
+    // Hydrated AgentMail.Attachment objects require `size` in the provider SDK contract. Keep the
+    // runtime check as a fail-closed guard for malformed or version-skewed provider payloads.
     const declaredSize: unknown = attachment.size;
     if (
       typeof declaredSize !== "number" ||
diff --git a/src/channel/src/received-message.ts b/src/channel/src/received-message.ts
index c69314f..88167b2 100644
--- a/src/channel/src/received-message.ts
+++ b/src/channel/src/received-message.ts
@@ -1,6 +1,10 @@
 import type { AgentMail } from "agentmail";
 
 export const AGENTMAIL_RECEIVED_LABEL = "received";
+// Email timestamps are provider-authored and may have ordinary clock skew, but allowing arbitrary
+// future values into durable retention can pin tombstones for years. One day is deliberately more
+// permissive than normal clock drift while keeping storage retention bounded.
+export const AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS = 24 * 60 * 60 * 1000;
 
 export function normalizeAgentMailInboxId(value: string): string {
   return value.trim().toLocaleLowerCase("en-US");
@@ -20,6 +24,20 @@ export function resolveAgentMailTimestampMs(value: unknown): number | null {
   return Number.isFinite(timestampMs) && timestampMs >= 0 ? timestampMs : null;
 }
 
+export function isAgentMailProviderTimestampWithinFutureSkew(
+  timestampMs: number,
+  observedAtMs: number,
+): boolean {
+  return timestampMs <= observedAtMs + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS;
+}
+
+export function capAgentMailProviderTimestampForRetention(
+  timestampMs: number,
+  observedAtMs: number,
+): number {
+  return Math.min(timestampMs, observedAtMs + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS);
+}
+
 export function isReceivedAgentMailMessage(
   message: AgentMail.MessageItem | AgentMail.Message,
   inboxId: string,
diff --git a/src/channel/src/reply-metadata.ts b/src/channel/src/reply-metadata.ts
new file mode 100644
index 0000000..a9ceccb
--- /dev/null
+++ b/src/channel/src/reply-metadata.ts
@@ -0,0 +1,44 @@
+import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime";
+
+const AGENTMAIL_CHANNEL_DATA_KEY = "agentmail";
+const TRIGGER_ARRIVED_AT_KEY = "triggerArrivedAt";
+
+function validTimestamp(value: unknown): value is number {
+  return typeof value === "number" && Number.isFinite(value) && value >= 0;
+}
+
+/** Persist local inbound timing with the durable reply so restart reconciliation can classify 404s. */
+export function withAgentMailTriggerArrival(
+  payload: ReplyPayload,
+  triggerArrivedAt: number,
+): ReplyPayload {
+  const channelData = payload.channelData ?? {};
+  const currentAgentMailData = channelData[AGENTMAIL_CHANNEL_DATA_KEY];
+  const agentMailData =
+    currentAgentMailData && typeof currentAgentMailData === "object"
+      ? currentAgentMailData
+      : {};
+  return {
+    ...payload,
+    channelData: {
+      ...channelData,
+      [AGENTMAIL_CHANNEL_DATA_KEY]: {
+        ...agentMailData,
+        [TRIGGER_ARRIVED_AT_KEY]: triggerArrivedAt,
+      },
+    },
+  };
+}
+
+/** Read local inbound timing from a persisted reply, falling back for pre-migration queue rows. */
+export function resolveAgentMailTriggerArrival(
+  payload: ReplyPayload,
+  fallbackAt: number,
+): number {
+  const agentMailData = payload.channelData?.[AGENTMAIL_CHANNEL_DATA_KEY];
+  if (!agentMailData || typeof agentMailData !== "object") {
+    return fallbackAt;
+  }
+  const triggerArrivedAt = (agentMailData as Record<string, unknown>)[TRIGGER_ARRIVED_AT_KEY];
+  return validTimestamp(triggerArrivedAt) ? triggerArrivedAt : fallbackAt;
+}
diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts
index c1e00ef..80115fa 100644
--- a/src/channel/src/send.test.ts
+++ b/src/channel/src/send.test.ts
@@ -590,6 +590,49 @@ describe("AgentMail reply-only outbound", () => {
     expect(reply).not.toHaveBeenCalled();
   });
 
+  it("uses persisted inbound arrival time to classify a 404 after restart", async () => {
+    reply.mockClear();
+    const unavailableClient = {
+      inboxes: {
+        messages: {
+          get: vi.fn(async () => {
+            throw new AgentMailError({ message: "not projected", statusCode: 404 });
+          }),
+          reply,
+        },
+      },
+    } as never;
+    const now = 10 * 60_000;
+    const result = await reconcileAgentMailUnknownSend(
+      {
+        cfg: {
+          channels: {
+            agentmail: { apiKey: "key", inboxId: "inbox_1", allowFrom: ["sender@example.com"] },
+          },
+        },
+        queueId: "queue_1",
+        channel: "agentmail",
+        to: "message:msg_1",
+        accountId: "default",
+        // The recovered queue timestamp can be stale after restart; the durable inbound metadata
+        // remains the authoritative projection-window reference.
+        enqueuedAt: 0,
+        retryCount: 2,
+        effectiveReplyToId: "msg_1",
+        payloads: [
+          {
+            text: "Hello",
+            channelData: { agentmail: { triggerArrivedAt: now - 1_000 } },
+          },
+        ],
+      } as never,
+      { client: unavailableClient, now: () => now },
+    );
+
+    expect(result).toMatchObject({ status: "unresolved", retryable: true });
+    expect(reply).not.toHaveBeenCalled();
+  });
+
   it("treats a triggering-message 404 as terminal after the projection window", async () => {
     const now = 10 * 60_000;
     const client = {
diff --git a/src/channel/src/send.ts b/src/channel/src/send.ts
index b65f5db..2a8a75c 100644
--- a/src/channel/src/send.ts
+++ b/src/channel/src/send.ts
@@ -13,6 +13,7 @@ import { sha256Hex } from "./digest.js";
 import { isAgentMailSenderAllowed, parseSingleFromMailbox } from "./mailbox.js";
 import { AgentMailMediaPolicyError, loadAgentMailOutboundAttachments } from "./media.js";
 import { agentMailInboxIdsEqual } from "./received-message.js";
+import { resolveAgentMailTriggerArrival } from "./reply-metadata.js";
 import { HYDRATION_NOT_FOUND_RETRY_WINDOW_MS } from "./inbound.js";
 
 // A media-size policy violation is genuinely terminal: the same payload will always exceed the
@@ -275,9 +276,10 @@ export async function reconcileAgentMailUnknownSend(
       retryable: false,
     };
   }
-  const recoveryReferenceAt = ctx.platformSendStartedAt ?? ctx.enqueuedAt;
-  const recoveryAgeMs = Math.max(0, (options.now?.() ?? Date.now()) - recoveryReferenceAt);
-  if (recoveryAgeMs >= AGENTMAIL_UNKNOWN_SEND_MAX_AGE_MS) {
+  const nowMs = options.now?.() ?? Date.now();
+  const sendRecoveryReferenceAt = ctx.platformSendStartedAt ?? ctx.enqueuedAt;
+  const sendRecoveryAgeMs = Math.max(0, nowMs - sendRecoveryReferenceAt);
+  if (sendRecoveryAgeMs >= AGENTMAIL_UNKNOWN_SEND_MAX_AGE_MS) {
     return {
       status: "unresolved",
       error: "AgentMail recovery is too close to the provider idempotency-key expiry",
@@ -293,6 +295,8 @@ export async function reconcileAgentMailUnknownSend(
     (value): value is string => typeof value === "string" && value.trim().length > 0,
   );
   const text = rendered?.text ?? payload.text ?? "";
+  const triggerArrivedAt = resolveAgentMailTriggerArrival(payload, ctx.enqueuedAt);
+  const triggerAgeMs = Math.max(0, nowMs - triggerArrivedAt);
   const recoveredPayload = { ...payload, text };
   delete recoveredPayload.mediaUrl;
   delete recoveredPayload.mediaUrls;
@@ -325,7 +329,7 @@ export async function reconcileAgentMailUnknownSend(
     if (isTerminalMediaPolicyFailure(error)) {
       return { status: "unresolved", error: error.message, retryable: false };
     }
-    if (isDeletedTriggerFailure(error, recoveryAgeMs)) {
+    if (isDeletedTriggerFailure(error, triggerAgeMs)) {
       // A 404 that persists beyond the provider projection window is treated as deletion.
       return { status: "unresolved", error: errorText(error), retryable: false };
     }
diff --git a/src/channel/src/webhook.test.ts b/src/channel/src/webhook.test.ts
index 85a6087..8b1c571 100644
--- a/src/channel/src/webhook.test.ts
+++ b/src/channel/src/webhook.test.ts
@@ -81,6 +81,32 @@ describe("AgentMail webhook", () => {
     );
   });
 
+  it("rejects far-future signed timestamps before durable admission", async () => {
+    const now = Date.now();
+    const dateNow = vi.spyOn(Date, "now").mockReturnValue(now);
+    const receive = vi.fn(async () => undefined);
+    const body = JSON.stringify({
+      type: "event",
+      event_type: "message.received",
+      message: {
+        inbox_id: "inbox_1",
+        message_id: "message_future",
+        timestamp: new Date(now + 365 * 24 * 60 * 60_000).toISOString(),
+      },
+    });
+    try {
+      const res = response();
+      await createAgentMailWebhookHandler({ account: account(), verifier, receive })(
+        request(body, signed(body)),
+        res,
+      );
+      expect(res.statusCode).toBe(200);
+      expect(receive).not.toHaveBeenCalled();
+    } finally {
+      dateNow.mockRestore();
+    }
+  });
+
   it("acknowledges but ignores a signed event with empty identifiers", async () => {
     const receive = vi.fn(async () => undefined);
     const body = JSON.stringify({
diff --git a/src/channel/src/webhook.ts b/src/channel/src/webhook.ts
index 03d82cc..b096011 100644
--- a/src/channel/src/webhook.ts
+++ b/src/channel/src/webhook.ts
@@ -5,7 +5,11 @@ import {
 } from "openclaw/plugin-sdk/webhook-ingress";
 import { Webhook } from "svix";
 import { type AgentMailLog, errorText } from "./log.js";
-import { agentMailInboxIdsEqual, resolveAgentMailTimestampMs } from "./received-message.js";
+import {
+  agentMailInboxIdsEqual,
+  isAgentMailProviderTimestampWithinFutureSkew,
+  resolveAgentMailTimestampMs,
+} from "./received-message.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
 
 const MAX_WEBHOOK_BODY_BYTES = 1024 * 1024;
@@ -117,6 +121,15 @@ export function createAgentMailWebhookHandler(params: {
     }
     try {
       const nowMs = Date.now();
+      if (
+        event.receivedAt !== undefined &&
+        !isAgentMailProviderTimestampWithinFutureSkew(event.receivedAt, nowMs)
+      ) {
+        // Acknowledging a signed but policy-invalid event prevents provider retry amplification.
+        // REST catch-up will admit it only if its provider time later enters the bounded window.
+        params.log?.warn?.("AgentMail webhook ignored an event timestamped too far in the future");
+        return respond(res, 200);
+      }
       await params.receive({
         accountId: params.account.accountId,
         inboxId: params.account.inboxId,
diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts
index ba439d6..411c27c 100644
--- a/src/channel/src/websocket.test.ts
+++ b/src/channel/src/websocket.test.ts
@@ -182,6 +182,41 @@ describe("AgentMail WebSocket ingress", () => {
     expect(waitForOpen).not.toHaveBeenCalled();
   });
 
+  it("rejects far-future provider timestamps before durable admission", async () => {
+    handlers.clear();
+    const now = 1_000_000;
+    const dateNow = vi.spyOn(Date, "now").mockReturnValue(now);
+    const receive = vi.fn(async () => undefined);
+    const controller = new AbortController();
+    const running = startAgentMailWebSocket({
+      account,
+      abortSignal: controller.signal,
+      receive,
+      catchUpSession: { run: catchUpRun },
+    });
+    try {
+      await vi.waitFor(() => expect(handlers.has("message")).toBe(true));
+      catchUpRun.mockClear();
+      handlers.get("message")?.({
+        type: "event",
+        eventType: "message.received",
+        message: {
+          inboxId: "inbox_1",
+          messageId: "message_future",
+          labels: ["received"],
+          timestamp: new Date(now + 365 * 24 * 60 * 60_000),
+        },
+      });
+
+      await vi.waitFor(() => expect(catchUpRun).toHaveBeenCalled());
+      expect(receive).not.toHaveBeenCalled();
+    } finally {
+      controller.abort();
+      await running;
+      dateNow.mockRestore();
+    }
+  });
+
   it("durably admits message.received frames before the received label projects", async () => {
     handlers.clear();
     const receive = vi.fn(async () => undefined);
diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts
index 507152a..10bd26c 100644
--- a/src/channel/src/websocket.ts
+++ b/src/channel/src/websocket.ts
@@ -12,6 +12,7 @@ import { AgentMailIngressCapacityError } from "./ingress.js";
 import { createBackoff, waitForRetry } from "./retry.js";
 import {
   agentMailInboxIdsEqual,
+  isAgentMailProviderTimestampWithinFutureSkew,
   resolveAgentMailTimestampMs,
 } from "./received-message.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
@@ -190,6 +191,14 @@ export async function startAgentMailWebSocket(params: {
       catchUpSupervisor.request();
       return true;
     }
+    const arrivedAt = now();
+    if (!isAgentMailProviderTimestampWithinFutureSkew(messageTimestampMs, arrivedAt)) {
+      params.log?.warn?.(
+        "AgentMail WebSocket ignored an event timestamped too far in the future",
+      );
+      catchUpSupervisor.request();
+      return false;
+    }
     queuedMessageIds.add(event.message.messageId);
     liveQueue.push({
       accountId: params.account.accountId,
@@ -197,7 +206,7 @@ export async function startAgentMailWebSocket(params: {
       messageId: event.message.messageId,
       transport: "websocket",
       receivedAt: messageTimestampMs,
-      arrivedAt: Date.now(),
+      arrivedAt,
     });
     runLiveWorker();
     return true;