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..0be808e 100644
--- a/src/channel/src/catch-up.test.ts
+++ b/src/channel/src/catch-up.test.ts
@@ -404,7 +404,7 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
-  it("admits malformed timestamps using local time without poisoning the cursor", async () => {
+  it("admits malformed timestamps using provider creation time", 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);
@@ -421,13 +421,13 @@ describe("AgentMail durable REST catch-up", () => {
     expect(receive).toHaveBeenCalledWith(
       expect.objectContaining({
         messageId: "bad_timestamp",
-        receivedAt: 1_000,
+        receivedAt: 1_100,
         arrivedAt: 1_000,
       }),
     );
   });
 
-  it("does not advance the cursor beyond the sweep's fixed upper bound", async () => {
+  it("advances the cursor from an invalid timestamp's fixed sweep 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);
@@ -495,7 +495,7 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
-  it("repairs a stored cursor after the host clock moves backwards", async () => {
+  it("repairs a future baseline without regressing the committed high-water mark", async () => {
     const store = memoryStore<AgentMailCatchUpCursor>();
     const key = sha256Hex("default\ninbox_1");
     await store.register(key, {
@@ -525,11 +525,40 @@ describe("AgentMail durable REST catch-up", () => {
     );
     expect(await store.lookup(key)).toMatchObject({
       baselineAtMs: 1_000_000,
-      highWaterAtMs: 1_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("advances malformed timestamps and admits far-future rows with bounded retention", async () => {
     const store = memoryStore<AgentMailCatchUpCursor>();
     const key = sha256Hex("default\ninbox_1");
     await store.register(key, {
@@ -637,8 +666,8 @@ describe("AgentMail durable REST catch-up", () => {
     await session.run({ receive, abortSignal: new AbortController().signal });
 
     expect(receive.mock.calls.map(([record]) => [record.messageId, record.receivedAt])).toEqual([
-      ["malformed", 1_000_000],
-      ["future", 1_000_000],
+      ["malformed", 900_000],
+      ["future", 1_000_000 + 24 * 60 * 60_000],
     ]);
     expect(list).toHaveBeenNthCalledWith(
       1,
diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts
index 235437d..8163aac 100644
--- a/src/channel/src/catch-up.ts
+++ b/src/channel/src/catch-up.ts
@@ -10,7 +10,10 @@ import {
 } from "./durable-receive.js";
 import {
   AGENTMAIL_RECEIVED_LABEL,
+  capAgentMailProviderTimestampForRetention,
+  isValidAgentMailTimestampMs,
   isReceivedAgentMailMessage,
+  resolveAgentMailTimestampMs,
   resolveReceivedAgentMailMessageTimestampMs,
 } from "./received-message.js";
 import { createBackoff, waitForRetry } from "./retry.js";
@@ -173,22 +176,18 @@ function cursorNamespace(account: ResolvedAgentMailAccount): string {
   return `${AGENTMAIL_REST_CATCH_UP_NAMESPACE}.${sha256Hex(account.accountId)}`;
 }
 
-function validTimestamp(value: unknown): value is number {
-  return typeof value === "number" && Number.isFinite(value) && value >= 0;
-}
-
 function normalizeCheckpoint(value: unknown): AgentMailCatchUpCheckpoint | undefined {
   if (!value || typeof value !== "object") {
     return undefined;
   }
   const checkpoint = value as Partial<AgentMailCatchUpCheckpoint>;
   if (
-    !validTimestamp(checkpoint.afterAtMs) ||
-    !validTimestamp(checkpoint.beforeAtMs) ||
+    !isValidAgentMailTimestampMs(checkpoint.afterAtMs) ||
+    !isValidAgentMailTimestampMs(checkpoint.beforeAtMs) ||
     checkpoint.beforeAtMs < checkpoint.afterAtMs ||
     typeof checkpoint.pageToken !== "string" ||
     checkpoint.pageToken.length === 0 ||
-    !validTimestamp(checkpoint.highWaterAtMs) ||
+    !isValidAgentMailTimestampMs(checkpoint.highWaterAtMs) ||
     typeof checkpoint.sinceBaseline !== "boolean"
   ) {
     return undefined;
@@ -209,8 +208,8 @@ function normalizeCursor(value: unknown): AgentMailCatchUpCursor | null {
   const cursor = value as Partial<AgentMailCatchUpCursor>;
   if (
     cursor.version !== CURSOR_VERSION ||
-    !validTimestamp(cursor.baselineAtMs) ||
-    !validTimestamp(cursor.highWaterAtMs) ||
+    !isValidAgentMailTimestampMs(cursor.baselineAtMs) ||
+    !isValidAgentMailTimestampMs(cursor.highWaterAtMs) ||
     typeof cursor.established !== "boolean"
   ) {
     return null;
@@ -244,11 +243,11 @@ function mergeCursor(
   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 +404,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,
@@ -457,19 +456,21 @@ export async function createAgentMailCatchUpSession(params: {
             params.account.inboxId,
           );
           const arrivedAt = now();
-          const receivedAt = providerReceivedAt !== null
-            ? Math.min(providerReceivedAt, sweepAtMs)
-            : sweepAtMs;
+          const providerCreatedAt = resolveAgentMailTimestampMs(message.createdAt);
+          const receivedAt = capAgentMailProviderTimestampForRetention(
+            providerReceivedAt ?? providerCreatedAt ?? arrivedAt,
+            arrivedAt,
+          );
           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 sender time must not turn a missed event into a permanent drop. Admit using
+            // provider/local arrival and advance to the fixed sweep bound so it is not re-listed
+            // and warned about on every periodic scan.
             params.log?.warn?.(
-              `AgentMail catch-up used arrival time for message ${message.messageId} with an invalid timestamp`,
+              `AgentMail catch-up used provider/local 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.
+            // Sender-authored time can be arbitrarily skewed. The row was admitted above using a
+            // bounded retention timestamp, and cursor advancement remains clamped to this sweep.
             params.log?.warn?.(
               `AgentMail catch-up clamped future timestamp for message ${message.messageId}`,
             );
@@ -496,7 +497,10 @@ export async function createAgentMailCatchUpSession(params: {
             throw error;
           }
           admitted += 1;
-          highWaterAtMs = Math.max(highWaterAtMs, receivedAt);
+          highWaterAtMs = Math.max(
+            highWaterAtMs,
+            providerReceivedAt === null ? sweepAtMs : 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..71cc8b3 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,44 @@ 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);
+    const accept = vi.fn(async () => ({
+      kind: "accepted" as const,
+      duplicate: false,
+      record: {},
+    }));
+    try {
+      await processAgentMailIngress({
+        journal: {
+          accept,
+          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,
+        }),
+      );
+      expect(accept).toHaveBeenCalledWith(
+        createAgentMailDurableInboundId(record),
+        expect.objectContaining({
+          receivedAt: now + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS,
+        }),
+        { receivedAt: 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(
@@ -634,33 +774,45 @@ describe("AgentMail durable ingress", () => {
     expect(fail).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), {
       reason: "dispatch-attempts-exhausted",
       message: "deferred turn abandoned before adoption",
+      failedAt: expect.any(Number),
     });
   });
 
   it("fails a completion marker after its retry ceiling without redispatching", async () => {
+    const now = 1_000_000;
+    const futureRecord = {
+      ...record,
+      receivedAt: now + 365 * 24 * 60 * 60_000,
+    };
+    const dateNow = vi.spyOn(Date, "now").mockReturnValue(now);
     const dispatch = vi.fn(async () => undefined);
     const complete = vi.fn(async () => {
       throw new Error("database read-only");
     });
     const fail = vi.fn(async () => true);
-    await processAgentMailIngress({
-      journal: {
-        accept: async () => ({ kind: "accepted", duplicate: false, record: {} }),
-        complete,
-        release: vi.fn(),
-        fail,
-      } as never,
-      record,
-      dispatch,
-      retryDelayMs: () => 0,
-    });
-    await vi.waitFor(() => expect(fail).toHaveBeenCalledOnce());
-    expect(dispatch).toHaveBeenCalledOnce();
-    expect(complete).toHaveBeenCalledTimes(50);
-    expect(fail).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), {
-      reason: "completion-marker-failed",
-      message: "AgentMail could not persist the completion marker",
-    });
+    try {
+      await processAgentMailIngress({
+        journal: {
+          accept: async () => ({ kind: "accepted", duplicate: false, record: {} }),
+          complete,
+          release: vi.fn(),
+          fail,
+        } as never,
+        record: futureRecord,
+        dispatch,
+        retryDelayMs: () => 0,
+      });
+      await vi.waitFor(() => expect(fail).toHaveBeenCalledOnce());
+      expect(dispatch).toHaveBeenCalledOnce();
+      expect(complete).toHaveBeenCalledTimes(50);
+      expect(fail).toHaveBeenCalledWith(createAgentMailDurableInboundId(futureRecord), {
+        reason: "completion-marker-failed",
+        message: "AgentMail could not persist the completion marker",
+        failedAt: now + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS,
+      });
+    } finally {
+      dateNow.mockRestore();
+    }
   });
 
   it("does not replay a turn after adoption if later dispatch settlement fails", async () => {
@@ -831,7 +983,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 +1007,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 +1037,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 +1167,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/gateway.test.ts b/src/channel/src/gateway.test.ts
index c0453ef..00249b3 100644
--- a/src/channel/src/gateway.test.ts
+++ b/src/channel/src/gateway.test.ts
@@ -18,6 +18,7 @@ const mocks = vi.hoisted(() => ({
   catchUpSettle: vi.fn(async () => undefined),
   createCatchUpSession: vi.fn(async () => ({ run: vi.fn(async () => undefined) })),
   registerError: false,
+  reclaimDeferredMedia: vi.fn(async () => undefined),
   webhookReceives: [] as Array<(record: unknown) => Promise<void>>,
 }));
 const apiVal = "key";
@@ -73,6 +74,11 @@ vi.mock("./ingress.js", () => ({
   replayPendingAgentMailIngress: vi.fn(async () => undefined),
 }));
 
+vi.mock("./inbound.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("./inbound.js")>()),
+  reclaimAbortedAgentMailDeferredMedia: mocks.reclaimDeferredMedia,
+}));
+
 vi.mock("./webhook.js", () => ({
   createAgentMailWebhookVerifier: (secret: string) => (secret === "invalid" ? null : {}),
   createAgentMailWebhookHandler: ({ receive }: { receive: (record: unknown) => Promise<void> }) => {
@@ -166,6 +172,7 @@ describe("AgentMail gateway route ownership", () => {
     mocks.webhookReceives.length = 0;
     mocks.catchUpRequest.mockClear();
     mocks.processIngress.mockReset();
+    mocks.reclaimDeferredMedia.mockClear();
     mocks.processIngress.mockRejectedValueOnce(new Error("queue full"));
     const controller = new AbortController();
     const running = startAgentMailGatewayAccount({
@@ -189,6 +196,7 @@ describe("AgentMail gateway route ownership", () => {
     expect(mocks.catchUpRequest).toHaveBeenCalledTimes(2);
     controller.abort();
     await running;
+    expect(mocks.reclaimDeferredMedia).toHaveBeenCalledWith(controller.signal);
   });
 
   it("releases an account's old path without letting stale cleanup remove its replacement", async () => {
diff --git a/src/channel/src/gateway.ts b/src/channel/src/gateway.ts
index 917265a..3dc813a 100644
--- a/src/channel/src/gateway.ts
+++ b/src/channel/src/gateway.ts
@@ -15,7 +15,11 @@ import {
 import { createAgentMailClient } from "./client.js";
 import { waitForRetry } from "./retry.js";
 import { createAgentMailDurableInboundReceiveJournal } from "./durable-receive.js";
-import { dispatchAgentMailInboundEvent, type AgentMailChannelRuntime } from "./inbound.js";
+import {
+  dispatchAgentMailInboundEvent,
+  reclaimAbortedAgentMailDeferredMedia,
+  type AgentMailChannelRuntime,
+} from "./inbound.js";
 import { processAgentMailIngress, replayPendingAgentMailIngress } from "./ingress.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
 import { createAgentMailWebhookHandler, createAgentMailWebhookVerifier } from "./webhook.js";
@@ -151,14 +155,19 @@ export async function startAgentMailGatewayAccount(params: {
     log: params.log,
   });
 
-  const startWebSocket = () =>
-    startAgentMailWebSocket({
-      account: params.account,
-      abortSignal: params.abortSignal,
-      receive,
-      log: params.log,
-      client,
-    });
+  const startWebSocket = async () => {
+    try {
+      await startAgentMailWebSocket({
+        account: params.account,
+        abortSignal: params.abortSignal,
+        receive,
+        log: params.log,
+        client,
+      });
+    } finally {
+      await reclaimAbortedAgentMailDeferredMedia(params.abortSignal);
+    }
+  };
 
   if (!params.account.webhookSecret) {
     params.log?.info?.(
@@ -281,4 +290,5 @@ export async function startAgentMailGatewayAccount(params: {
     }
   });
   await Promise.allSettled([...periodicWorkers, catchUpSupervisor.settle()]);
+  await reclaimAbortedAgentMailDeferredMedia(params.abortSignal);
 }
diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts
index 985ab39..d334655 100644
--- a/src/channel/src/inbound.test.ts
+++ b/src/channel/src/inbound.test.ts
@@ -4,6 +4,7 @@ import {
   AgentMailLabelPendingError,
   buildAgentMailSessionKey,
   dispatchAgentMailInboundEvent,
+  reclaimAbortedAgentMailDeferredMedia,
   resolveAgentMailMessageText,
 } from "./inbound.js";
 import { AgentMailMediaPolicyError } from "./media.js";
@@ -256,8 +257,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 },
+      },
+    });
     expect(delivery.durable()).toMatchObject({
       to: "message:message_1",
       replyToId: "message_1",
@@ -337,13 +345,16 @@ describe("AgentMail REST-authoritative inbound", () => {
     expect(onTurnAbandoned).toHaveBeenCalledOnce();
   });
 
-  it("cleans up deferred-turn attachments when account shutdown aborts the queued turn", async () => {
+  it("reclaims deferred attachments only after account shutdown settles", 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 +366,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 +381,14 @@ 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 reclaimAbortedAgentMailDeferredMedia(controller.signal);
+    expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true });
+
+    await lifecycle?.onAbandoned();
+    expect(rm).toHaveBeenCalledTimes(1);
   });
 
   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..97d5380 100644
--- a/src/channel/src/inbound.ts
+++ b/src/channel/src/inbound.ts
@@ -18,6 +18,28 @@ import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.j
 const CHANNEL_ID = "agentmail";
 export const HYDRATION_NOT_FOUND_RETRY_WINDOW_MS = 5 * 60_000;
 
+type DeferredMediaCleanup = () => Promise<void>;
+const deferredMediaCleanups = new WeakMap<AbortSignal, Set<DeferredMediaCleanup>>();
+
+/**
+ * Reclaims media retained by deferred turns only after the account's ingress workers have stopped.
+ * Core does not drain its follow-up queue during shutdown, so no queued turn can adopt these files
+ * once the owning account lifecycle has settled.
+ */
+export async function reclaimAbortedAgentMailDeferredMedia(
+  abortSignal: AbortSignal,
+): Promise<void> {
+  if (!abortSignal.aborted) {
+    return;
+  }
+  const cleanups = deferredMediaCleanups.get(abortSignal);
+  if (!cleanups) {
+    return;
+  }
+  deferredMediaCleanups.delete(abortSignal);
+  await Promise.allSettled([...cleanups].map((cleanup) => cleanup()));
+}
+
 // AgentMail message labels the channel gates on. "received" marks authentic inbound mail; the
 // rejected set marks provider-flagged mail that must never reach the agent.
 const AGENTMAIL_REJECTED_LABELS = ["spam", "blocked", "unauthenticated"];
@@ -274,10 +296,27 @@ export async function dispatchAgentMailInboundEvent(params: {
     ).then(() => undefined);
     await mediaCleanup;
   };
+  let detachDeferredMediaCleanup = () => {};
+  const retainDeferredMediaUntilShutdown = () => {
+    const abortSignal = params.abortSignal;
+    if (!abortSignal || inboundMedia.paths.length === 0) {
+      return;
+    }
+    const cleanup = cleanupInboundMedia;
+    const cleanups = deferredMediaCleanups.get(abortSignal) ?? new Set();
+    cleanups.add(cleanup);
+    deferredMediaCleanups.set(abortSignal, cleanups);
+    detachDeferredMediaCleanup = () => {
+      cleanups.delete(cleanup);
+      if (cleanups.size === 0) {
+        deferredMediaCleanups.delete(abortSignal);
+      }
+    };
+  };
   let detachAbortCleanup = () => {};
   if (params.abortSignal) {
     const onAbort = () => {
-      if (!turnAdoptionObserved) {
+      if (!turnAdoptionObserved && !turnDeferred) {
         void cleanupInboundMedia();
       }
     };
@@ -300,14 +339,17 @@ export async function dispatchAgentMailInboundEvent(params: {
       turnAdoptionObserved = true;
       turnAdopted = true;
       detachAbortCleanup();
+      detachDeferredMediaCleanup();
       await params.onTurnAdopted?.();
     },
     onDeferred: () => {
       turnDeferred = true;
+      retainDeferredMediaUntilShutdown();
       params.onTurnDeferred?.();
     },
     onAbandoned: async () => {
       detachAbortCleanup();
+      detachDeferredMediaCleanup();
       if (!turnAdoptionObserved) {
         await cleanupInboundMedia();
       }
diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts
index 6378c94..696f8cd 100644
--- a/src/channel/src/ingress.ts
+++ b/src/channel/src/ingress.ts
@@ -2,18 +2,17 @@ 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,
+  isValidAgentMailTimestampMs,
+} 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 +78,30 @@ const activeDispatches = new Map<string, ActiveDispatch>();
 
 const retryDelayMs = createBackoff(30 * 60_000);
 
+function resolveAgentMailIngressTerminalAt(
+  record: AgentMailIngressRecord,
+  terminalAt = Date.now(),
+): number {
+  const providerReceivedAt = capAgentMailProviderTimestampForRetention(
+    isValidAgentMailTimestampMs(record.receivedAt) ? record.receivedAt : terminalAt,
+    terminalAt,
+  );
+  return Math.max(terminalAt, providerReceivedAt);
+}
+
+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 through the durable recovery horizon after that bounded time, or it can re-enter a
+  // provider-time scan after a locally-timestamped marker has already expired.
+  await params.journal.complete(params.id, {
+    completedAt: resolveAgentMailIngressTerminalAt(params.record),
+  });
+}
+
 // 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 {
@@ -156,11 +179,25 @@ export async function processAgentMailIngress(params: {
   retryDelayMs?: (attempt: number) => number;
   log?: AgentMailLog;
 }): Promise<"accepted" | "duplicate"> {
-  const id = createAgentMailDurableInboundId(params.record);
+  // Enforce the retention invariant at the shared durable boundary as well as transport parsing.
+  // This protects future transports and direct callers from persisting an unbounded timestamp.
+  const arrivedAt = isValidAgentMailTimestampMs(params.record.arrivedAt)
+    ? params.record.arrivedAt
+    : Date.now();
+  const record = {
+    ...params.record,
+    receivedAt: capAgentMailProviderTimestampForRetention(
+      isValidAgentMailTimestampMs(params.record.receivedAt)
+        ? params.record.receivedAt
+        : arrivedAt,
+      arrivedAt,
+    ),
+  };
+  const id = createAgentMailDurableInboundId(record);
   // accept() throws AgentMailIngressCapacityError directly when durable ingress is full, so
   // transports can apply plugin-owned backpressure (reject new mail, keep accepted pending mail).
-  const accepted = await params.journal.accept(id, params.record, {
-    receivedAt: params.record.receivedAt,
+  const accepted = await params.journal.accept(id, record, {
+    receivedAt: record.receivedAt,
   });
   // "completed" is a durable dedupe hit. "failed" is a terminal tombstone in SDK releases that
   // expose one; compare as a string so the defensive branch also compiles where the accept-result
@@ -169,13 +206,13 @@ export async function processAgentMailIngress(params: {
   if (acceptedKind === "completed" || acceptedKind === "failed") {
     return "duplicate";
   }
-  const record = accepted.kind === "pending" ? accepted.record.payload : params.record;
+  const dispatchRecord = accepted.kind === "pending" ? accepted.record.payload : record;
   // Pending duplicates also register as successors. This closes the account-reload race where a
   // replacement replay ran just before the old account admitted its final live event.
   scheduleAgentMailIngressDispatch({
     journal: params.journal,
     id,
-    record,
+    record: dispatchRecord,
     dispatch: params.dispatch,
     abortSignal: params.abortSignal,
     retryDelay: params.retryDelayMs,
@@ -235,7 +272,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");
@@ -252,6 +289,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
                     (await params.journal.fail(params.id, {
                       reason: "completion-marker-failed",
                       message: "AgentMail could not persist the adoption completion marker",
+                      failedAt: resolveAgentMailIngressTerminalAt(params.record),
                     }))
                   ) {
                     params.dispatchCompleted = true;
@@ -321,6 +359,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;
@@ -338,9 +381,10 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
               await params.journal.fail(params.id, {
                 reason: "dispatch-attempts-exhausted",
                 message: lastError,
+                failedAt: resolveAgentMailIngressTerminalAt(params.record),
               });
             } else {
-              await params.journal.complete(params.id);
+              await completeAgentMailIngress(params);
             }
           } catch {
             // Best effort: TTL pruning still reclaims the row if the terminal marker cannot
@@ -387,9 +431,10 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
               await params.journal.fail(params.id, {
                 reason: "dispatch-attempts-exhausted",
                 message: errorText(dispatchError),
+                failedAt: resolveAgentMailIngressTerminalAt(params.record),
               });
             } 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 +476,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;
@@ -443,6 +488,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro
           return await params.journal.fail(params.id, {
             reason: "completion-marker-failed",
             message: "AgentMail could not persist the completion marker",
+            failedAt: resolveAgentMailIngressTerminalAt(params.record),
           });
         }
         return false;
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..7017e40 100644
--- a/src/channel/src/received-message.ts
+++ b/src/channel/src/received-message.ts
@@ -1,6 +1,9 @@
 import type { AgentMail } from "agentmail";
 
 export const AGENTMAIL_RECEIVED_LABEL = "received";
+// Email timestamps are sender-authored and may have arbitrary clock skew. Keep their contribution
+// to durable retention bounded so malformed dates cannot pin tombstones for years.
+export const AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS = 24 * 60 * 60 * 1000;
 
 export function normalizeAgentMailInboxId(value: string): string {
   return value.trim().toLocaleLowerCase("en-US");
@@ -10,6 +13,10 @@ export function agentMailInboxIdsEqual(left: string, right: string): boolean {
   return normalizeAgentMailInboxId(left) === normalizeAgentMailInboxId(right);
 }
 
+export function isValidAgentMailTimestampMs(value: unknown): value is number {
+  return typeof value === "number" && Number.isFinite(value) && value >= 0;
+}
+
 export function resolveAgentMailTimestampMs(value: unknown): number | null {
   const timestampMs =
     value instanceof Date
@@ -17,7 +24,14 @@ export function resolveAgentMailTimestampMs(value: unknown): number | null {
       : typeof value === "string" || typeof value === "number"
         ? new Date(value).getTime()
         : Number.NaN;
-  return Number.isFinite(timestampMs) && timestampMs >= 0 ? timestampMs : null;
+  return isValidAgentMailTimestampMs(timestampMs) ? timestampMs : null;
+}
+
+export function capAgentMailProviderTimestampForRetention(
+  timestampMs: number,
+  observedAtMs: number,
+): number {
+  return Math.min(timestampMs, observedAtMs + AGENTMAIL_PROVIDER_FUTURE_SKEW_MAX_MS);
 }
 
 export function isReceivedAgentMailMessage(
diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts
index c1e00ef..2f2481b 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 the outbound queue time for a transient 404 after a long-running turn", 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 triggering turn can be much older than the reply. Recovery starts when the outbound
+        // row is enqueued, so a recent send still gets the full provider-projection window.
+        enqueuedAt: now - 1_000,
+        retryCount: 2,
+        effectiveReplyToId: "msg_1",
+        payloads: [
+          {
+            text: "Hello",
+            channelData: { agentmail: { triggerArrivedAt: 0 } },
+          },
+        ],
+      } 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..474b86a 100644
--- a/src/channel/src/send.ts
+++ b/src/channel/src/send.ts
@@ -275,8 +275,9 @@ export async function reconcileAgentMailUnknownSend(
       retryable: false,
     };
   }
+  const nowMs = options.now?.() ?? Date.now();
   const recoveryReferenceAt = ctx.platformSendStartedAt ?? ctx.enqueuedAt;
-  const recoveryAgeMs = Math.max(0, (options.now?.() ?? Date.now()) - recoveryReferenceAt);
+  const recoveryAgeMs = Math.max(0, nowMs - recoveryReferenceAt);
   if (recoveryAgeMs >= AGENTMAIL_UNKNOWN_SEND_MAX_AGE_MS) {
     return {
       status: "unresolved",
diff --git a/src/channel/src/webhook.test.ts b/src/channel/src/webhook.test.ts
index 85a6087..4928db2 100644
--- a/src/channel/src/webhook.test.ts
+++ b/src/channel/src/webhook.test.ts
@@ -63,6 +63,7 @@ describe("AgentMail webhook", () => {
         inbox_id: "INBOX_1",
         message_id: "message_1",
         timestamp: "2026-07-15T12:34:56.000Z",
+        created_at: "2026-07-15T12:35:00.000Z",
       },
     });
     const res = response();
@@ -81,6 +82,64 @@ describe("AgentMail webhook", () => {
     );
   });
 
+  it("falls back to provider creation time when the sender timestamp is invalid", async () => {
+    const receive = vi.fn(async () => undefined);
+    const body = JSON.stringify({
+      type: "event",
+      event_type: "message.received",
+      message: {
+        inbox_id: "inbox_1",
+        message_id: "message_created",
+        timestamp: "not-a-date",
+        created_at: "2026-07-15T12:35:00.000Z",
+      },
+    });
+    const res = response();
+    await createAgentMailWebhookHandler({ account: account(), verifier, receive })(
+      request(body, signed(body)),
+      res,
+    );
+
+    expect(receive).toHaveBeenCalledWith(
+      expect.objectContaining({
+        messageId: "message_created",
+        receivedAt: Date.parse("2026-07-15T12:35:00.000Z"),
+      }),
+    );
+  });
+
+  it("admits far-future signed timestamps with bounded retention", 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).toHaveBeenCalledWith(
+        expect.objectContaining({
+          messageId: "message_future",
+          receivedAt: now + 24 * 60 * 60_000,
+          arrivedAt: now,
+        }),
+      );
+    } 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..44b2fea 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,
+  capAgentMailProviderTimestampForRetention,
+  resolveAgentMailTimestampMs,
+} from "./received-message.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
 
 const MAX_WEBHOOK_BODY_BYTES = 1024 * 1024;
@@ -42,7 +46,10 @@ function parseVerifiedEvent(payload: unknown): {
   if (!inboxId || !messageId) {
     return null;
   }
-  const receivedAt = resolveAgentMailTimestampMs(mail.timestamp);
+  // message.timestamp drives REST scan ordering, so retain its bounded value for dedupe. Fall back
+  // to provider ingestion time when the sender-authored timestamp is absent or malformed.
+  const receivedAt =
+    resolveAgentMailTimestampMs(mail.timestamp) ?? resolveAgentMailTimestampMs(mail.created_at);
   return { inboxId, messageId, ...(receivedAt === null ? {} : { receivedAt }) };
 }
 
@@ -117,12 +124,16 @@ export function createAgentMailWebhookHandler(params: {
     }
     try {
       const nowMs = Date.now();
+      const receivedAt = capAgentMailProviderTimestampForRetention(
+        event.receivedAt ?? nowMs,
+        nowMs,
+      );
       await params.receive({
         accountId: params.account.accountId,
         inboxId: params.account.inboxId,
         messageId: event.messageId,
         transport: "webhook",
-        receivedAt: event.receivedAt ?? nowMs,
+        receivedAt,
         arrivedAt: nowMs,
       });
       return respond(res, 200);
diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts
index ba439d6..8b074ff 100644
--- a/src/channel/src/websocket.test.ts
+++ b/src/channel/src/websocket.test.ts
@@ -182,6 +182,84 @@ describe("AgentMail WebSocket ingress", () => {
     expect(waitForOpen).not.toHaveBeenCalled();
   });
 
+  it("admits far-future sender timestamps with bounded retention", 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(receive).toHaveBeenCalledOnce());
+      expect(receive).toHaveBeenCalledWith(
+        expect.objectContaining({
+          messageId: "message_future",
+          receivedAt: now + 24 * 60 * 60_000,
+          arrivedAt: now,
+        }),
+      );
+    } finally {
+      controller.abort();
+      await running;
+      dateNow.mockRestore();
+    }
+  });
+
+  it("admits invalid sender timestamps using provider creation time", async () => {
+    handlers.clear();
+    const now = 1_000_000;
+    const receive = vi.fn(async () => undefined);
+    const controller = new AbortController();
+    const running = startAgentMailWebSocket({
+      account,
+      abortSignal: controller.signal,
+      receive,
+      catchUpSession: { run: catchUpRun },
+      now: () => now,
+    });
+    await vi.waitFor(() => expect(handlers.has("message")).toBe(true));
+    handlers.get("message")?.({
+      type: "event",
+      eventType: "message.received",
+      message: {
+        inboxId: "inbox_1",
+        messageId: "message_created",
+        labels: ["received"],
+        timestamp: new Date(Number.NaN),
+        createdAt: new Date(900_000),
+      },
+    });
+
+    await vi.waitFor(() => expect(receive).toHaveBeenCalledOnce());
+    expect(receive).toHaveBeenCalledWith(
+      expect.objectContaining({
+        messageId: "message_created",
+        receivedAt: 900_000,
+        arrivedAt: now,
+      }),
+    );
+    controller.abort();
+    await running;
+  });
+
   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..f9c1353 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,
+  capAgentMailProviderTimestampForRetention,
   resolveAgentMailTimestampMs,
 } from "./received-message.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
@@ -174,11 +175,12 @@ export async function startAgentMailWebSocket(params: {
     }
     // The event type itself is authoritative for live receipt. Provider label projection can lag
     // the WebSocket frame; durable hydration already retries that condition safely.
+    const arrivedAt = now();
     const messageTimestampMs = resolveAgentMailTimestampMs(event.message.timestamp);
     if (messageTimestampMs === null) {
-      params.log?.warn?.("AgentMail WebSocket received an event with an invalid timestamp");
-      catchUpSupervisor.request();
-      return false;
+      params.log?.warn?.(
+        "AgentMail WebSocket used provider arrival time for an event with an invalid timestamp",
+      );
     }
     if (queuedMessageIds.has(event.message.messageId)) {
       return true;
@@ -190,14 +192,19 @@ export async function startAgentMailWebSocket(params: {
       catchUpSupervisor.request();
       return true;
     }
+    const providerCreatedAtMs = resolveAgentMailTimestampMs(event.message.createdAt);
+    const receivedAt = capAgentMailProviderTimestampForRetention(
+      messageTimestampMs ?? providerCreatedAtMs ?? arrivedAt,
+      arrivedAt,
+    );
     queuedMessageIds.add(event.message.messageId);
     liveQueue.push({
       accountId: params.account.accountId,
       inboxId: params.account.inboxId,
       messageId: event.message.messageId,
       transport: "websocket",
-      receivedAt: messageTimestampMs,
-      arrivedAt: Date.now(),
+      receivedAt,
+      arrivedAt,
     });
     runLiveWorker();
     return true;