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

@@ -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 () => {

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