Harden durable ingress edge cases

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:

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,

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