Harden durable AgentMail ingress and plugin loading

README.md

@@ -26,7 +26,9 @@ For development, use `openclaw plugins install --link .` so OpenClaw loads this
 
 ## Configure
 
-Set `AGENTMAIL_API_KEY` in the environment that runs the OpenClaw Gateway. To enable **webhook** ingress for the channel, also set `AGENTMAIL_WEBHOOK_SECRET` (Svix-signed); without it the channel falls back to WebSocket ingress.
+Provide the API key through `AGENTMAIL_API_KEY` in the Gateway environment or as
+`channels.agentmail.apiKey`. To enable **webhook** ingress, also configure
+`AGENTMAIL_WEBHOOK_SECRET` (Svix-signed); without it the channel falls back to WebSocket ingress.
 
 OpenClaw can scope the secrets to this plugin in `~/.openclaw/openclaw.json`:
 
@@ -55,10 +57,9 @@ openclaw plugins inspect agentmail --runtime
 
 ### Tool config (optional SDK settings)
 
-> **Credentials:** the email **tools** authenticate only with the `AGENTMAIL_API_KEY` environment
-> variable, while the **channel** can also take an inline or resolved `apiKey` in `channels.agentmail`.
-> Always set `AGENTMAIL_API_KEY` in the Gateway environment so both surfaces are configured; a
-> channel-only inline key leaves the tools reporting AgentMail as unconfigured.
+> **Credentials:** the email **tools** use the resolved `apiKey` from the default
+> `channels.agentmail` account when configured, with `AGENTMAIL_API_KEY` as the tools-only fallback.
+> This keeps inline and secret-reference channel configuration shared across both surfaces.
 
 Optional AgentMail SDK settings for the **tools** belong under `plugins.entries.agentmail.config`:
 

package.json

@@ -23,7 +23,7 @@
   "scripts": {
     "build": "tsc -p tsconfig.json",
     "plugin:build": "npm run build && node scripts/build-manifest.mjs",
-    "plugin:check": "node scripts/build-manifest.mjs --check",
+    "plugin:check": "npm run build && node scripts/build-manifest.mjs --check",
     "plugin:validate": "npm run build && node scripts/build-manifest.mjs --check && node scripts/validate-plugin.mjs",
     "prepack": "npm run plugin:validate",
     "test": "vitest run --config ./vitest.config.ts"
@@ -49,8 +49,7 @@
   },
   "openclaw": {
     "extensions": [
-      "./dist/tools/index.js",
-      "./dist/channel/index.js"
+      "./dist/index.js"
     ],
     "channel": {
       "id": "agentmail",

scripts/validate-plugin.mjs

@@ -20,6 +20,8 @@ const childEnv = {
   OPENCLAW_STATE_DIR: stateDir,
   OPENCLAW_CONFIG_DIR: stateDir,
   OPENCLAW_HOME: stateDir,
+  NO_COLOR: "1",
+  FORCE_COLOR: "0",
 };
 
 function run(args) {

src/channel/src/accounts.test.ts

@@ -25,7 +25,7 @@ describe("AgentMail account config", () => {
       channels: {
         agentmail: {
           apiKey: paddedApi,
-          inboxId: " inbox_123 ",
+          inboxId: " Agent@AgentMail.TO ",
           webhookSecret: paddedHook,
         },
       },
@@ -34,7 +34,7 @@ describe("AgentMail account config", () => {
     expect(account).toMatchObject({
       accountId: "default",
       apiKey: apiVal,
-      inboxId: "inbox_123",
+      inboxId: "Agent@AgentMail.TO",
       webhookSecret: hookVal,
       webhookPath: "/webhooks/agentmail",
       dmPolicy: "allowlist",

src/channel/src/accounts.ts

@@ -14,6 +14,7 @@ import {
 import { normalizeStringEntries } from "openclaw/plugin-sdk/string-coerce-runtime";
 import { AGENTMAIL_MEDIA_DEFAULT_MB } from "./config-schema.js";
 import { normalizeMailbox } from "./mailbox.js";
+import { normalizeAgentMailInboxId } from "./received-message.js";
 import {
   type AgentMailChannelConfig,
   type ResolvedAgentMailAccount,
@@ -147,7 +148,10 @@ export function resolveAgentMailAccount(
     accountId: id,
     enabled: channel.enabled !== false && account?.enabled !== false,
     apiKey: apiVal,
-    inboxId: merged.inboxId?.trim() ?? "",
+    // Preserve configured spelling as the durable identity. Queue namespaces, catch-up cursor keys,
+    // and conversation ids include this value, so lowercasing it during resolution would strand
+    // persisted state after an upgrade. Normalize provider values only where ids are compared.
+    inboxId: (merged.inboxId ?? "").trim(),
     webhookSecret: hookVal,
     webhookPath:
       configuredPath ||
@@ -187,7 +191,7 @@ export function findConflictingAgentMailInboxOwner(
     if (
       other.enabled &&
       isAgentMailAccountConfigured(other) &&
-      other.inboxId === account.inboxId
+      normalizeAgentMailInboxId(other.inboxId) === normalizeAgentMailInboxId(account.inboxId)
     ) {
       return other.accountId;
     }

src/channel/src/catch-up.test.ts

@@ -113,7 +113,7 @@ describe("AgentMail durable REST catch-up", () => {
       .mockResolvedValueOnce({
         count: 2,
         messages: [
-          message({ id: "message_1", timestamp: 1_100 }),
+          message({ id: "message_1", timestamp: 1_100, inboxId: "INBOX_1" }),
           message({ id: "sent_1", timestamp: 1_150, labels: ["sent"] }),
         ],
         nextPageToken: "page_2",
@@ -173,6 +173,97 @@ describe("AgentMail durable REST catch-up", () => {
     );
   });
 
+  it("admits malformed timestamps using local time without poisoning the cursor", 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);
+    const list = vi.fn(async () => ({ count: 1, messages: [malformed] }));
+    const session = await createAgentMailCatchUpSession({
+      account,
+      client: { inboxes: { messages: { list } } } as never,
+      store: store as never,
+      now: () => 1_000,
+    });
+    const receive = vi.fn(async () => undefined);
+
+    await session.run({ receive, abortSignal: new AbortController().signal });
+    expect(receive).toHaveBeenCalledWith(
+      expect.objectContaining({
+        messageId: "bad_timestamp",
+        receivedAt: 1_000,
+        arrivedAt: 1_000,
+      }),
+    );
+  });
+
+  it("does not advance the cursor beyond the sweep's fixed upper bound", 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);
+    const list = vi
+      .fn()
+      .mockResolvedValueOnce({ count: 1, messages: [malformed] })
+      .mockResolvedValueOnce({ count: 0, messages: [] });
+    let nowMs = 1_000_000;
+    const session = await createAgentMailCatchUpSession({
+      account,
+      client: { inboxes: { messages: { list } } } as never,
+      store: store as never,
+      now: () => nowMs,
+    });
+    nowMs = 2_000_000;
+    await session.run({
+      receive: async () => {
+        nowMs = 3_000_000;
+      },
+      abortSignal: new AbortController().signal,
+    });
+
+    await session.run({ receive: vi.fn(), abortSignal: new AbortController().signal });
+    expect(list).toHaveBeenLastCalledWith(
+      "inbox_1",
+      expect.objectContaining({
+        after: new Date(2_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS),
+      }),
+      expect.any(Object),
+    );
+  });
+
+  it("clamps a future message timestamp before advancing the high-water cursor", async () => {
+    const store = memoryStore<never>();
+    const nowMs = 1_000_000;
+    const list = vi
+      .fn()
+      .mockResolvedValueOnce({
+        count: 1,
+        messages: [message({ id: "future", timestamp: nowMs + 60 * 60_000 })],
+      })
+      .mockResolvedValueOnce({ count: 0, messages: [] });
+    const session = await createAgentMailCatchUpSession({
+      account,
+      client: { inboxes: { messages: { list } } } as never,
+      store: store as never,
+      now: () => nowMs,
+    });
+    const receive = vi.fn(async () => undefined);
+
+    await session.run({ receive, abortSignal: new AbortController().signal });
+    expect(list).toHaveBeenNthCalledWith(
+      1,
+      "inbox_1",
+      expect.objectContaining({ before: new Date(nowMs + 1) }),
+      expect.any(Object),
+    );
+    await session.run({ receive, abortSignal: new AbortController().signal });
+    expect(list).toHaveBeenLastCalledWith(
+      "inbox_1",
+      expect.objectContaining({
+        after: new Date(nowMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS),
+      }),
+      expect.any(Object),
+    );
+  });
+
   it("lists from the monitoring baseline on a deep sweep", async () => {
     const store = memoryStore<never>();
     const list = vi.fn(async () => ({

src/channel/src/catch-up.ts

@@ -7,7 +7,11 @@ import {
   AGENTMAIL_DURABLE_COMPLETED_TTL_MS,
   AgentMailIngressCapacityError,
 } from "./durable-receive.js";
-import { AGENTMAIL_RECEIVED_LABEL } from "./inbound.js";
+import {
+  AGENTMAIL_RECEIVED_LABEL,
+  isReceivedAgentMailMessage,
+  resolveReceivedAgentMailMessageTimestampMs,
+} from "./received-message.js";
 import { createBackoff, waitForRetry } from "./retry.js";
 import { getAgentMailRuntime } from "./runtime.js";
 import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js";
@@ -175,20 +179,12 @@ function normalizeCursor(value: unknown): AgentMailCatchUpCursor | null {
   };
 }
 
-function isReceivedMessage(message: AgentMail.MessageItem, inboxId: string): boolean {
-  return (
-    message.inboxId === inboxId &&
-    (Array.isArray(message.labels) ? message.labels : []).some(
-      (label) => String(label).toLocaleLowerCase("en-US") === AGENTMAIL_RECEIVED_LABEL,
-    )
-  );
-}
-
 async function persistCursor(params: {
   store: PluginStateKeyedStore<AgentMailCatchUpCursor>;
   key: string;
   baselineAtMs: number;
   highWaterAtMs: number;
+  upperBoundAtMs: number;
   established: boolean;
 }): Promise<void> {
   const update = params.store.update;
@@ -198,7 +194,13 @@ async function persistCursor(params: {
       return {
         version: CURSOR_VERSION,
         baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs,
-        highWaterAtMs: Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs),
+        highWaterAtMs: Math.max(
+          current?.baselineAtMs ?? params.baselineAtMs,
+          Math.min(
+            params.upperBoundAtMs,
+            Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs),
+          ),
+        ),
         established: current?.established === true || params.established,
       };
     });
@@ -208,7 +210,13 @@ async function persistCursor(params: {
   await params.store.register(params.key, {
     version: CURSOR_VERSION,
     baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs,
-    highWaterAtMs: Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs),
+    highWaterAtMs: Math.max(
+      current?.baselineAtMs ?? params.baselineAtMs,
+      Math.min(
+        params.upperBoundAtMs,
+        Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs),
+      ),
+    ),
     established: current?.established === true || params.established,
   });
 }
@@ -245,6 +253,9 @@ export async function createAgentMailCatchUpSession(params: {
       if (!storedCursor) {
         throw new Error("AgentMail WebSocket catch-up cursor is unavailable");
       }
+      const runAtMs = now();
+      const scanUpperBoundAtMs = runAtMs;
+      const effectiveHighWaterAtMs = Math.min(storedCursor.highWaterAtMs, runAtMs);
       // Never scan below the dedupe horizon on ANY sweep. Completed-message tombstones expire after
       // AGENTMAIL_DURABLE_COMPLETED_TTL_MS; once they do, a message still in scan range would be
       // re-admitted as a duplicate turn. The floor is applied to the normal high-water sweep too:
@@ -254,13 +265,13 @@ export async function createAgentMailCatchUpSession(params: {
       // The deep sweep lists from max(baseline, dedupeFloor): it recovers back-dated mail within the
       // dedupe window and since monitoring began, but deliberately does not scan before the baseline,
       // which would re-inject pre-monitoring history that has no tombstone protection.
-      const dedupeFloorMs = now() - AGENTMAIL_DURABLE_COMPLETED_TTL_MS;
+      const dedupeFloorMs = runAtMs - AGENTMAIL_DURABLE_COMPLETED_TTL_MS;
       const baseAfterMs =
         sinceBaseline || !storedCursor.established
           ? storedCursor.baselineAtMs
-          : storedCursor.highWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS;
+          : effectiveHighWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS;
       const afterMs = Math.max(0, baseAfterMs, dedupeFloorMs);
-      let highWaterAtMs = storedCursor.highWaterAtMs;
+      let highWaterAtMs = effectiveHighWaterAtMs;
       let pageCursor: string | undefined;
       let admitted = 0;
       do {
@@ -271,6 +282,10 @@ export async function createAgentMailCatchUpSession(params: {
             ...(pageCursor ? { pageToken: pageCursor } : {}),
             labels: [AGENTMAIL_RECEIVED_LABEL],
             after: new Date(afterMs),
+            // Do not repeatedly scan provider-clock-skewed future messages. Add one millisecond so
+            // an exclusive provider bound includes messages stamped exactly at runAtMs and so a
+            // fresh cursor never sends identical after/before values.
+            before: new Date(scanUpperBoundAtMs + 1),
             ascending: true,
             includeSpam: false,
             includeBlocked: false,
@@ -284,16 +299,21 @@ export async function createAgentMailCatchUpSession(params: {
           if (abortSignal.aborted) {
             return;
           }
-          if (!isReceivedMessage(message, params.account.inboxId)) {
+          if (!isReceivedAgentMailMessage(message, params.account.inboxId)) {
             continue;
           }
+          // Invalid provider time must not turn a missed live event into a permanent drop. Admit
+          // with local observation time; durable message-id dedupe keeps overlap scans safe.
+          const messageTimestampMs =
+            resolveReceivedAgentMailMessageTimestampMs(message, params.account.inboxId) ??
+            scanUpperBoundAtMs;
           try {
             await receive({
               accountId: params.account.accountId,
               inboxId: params.account.inboxId,
               messageId: message.messageId,
               transport: "rest",
-              receivedAt: message.timestamp.getTime(),
+              receivedAt: messageTimestampMs,
               arrivedAt: now(),
             });
           } catch (error) {
@@ -309,7 +329,10 @@ export async function createAgentMailCatchUpSession(params: {
             throw error;
           }
           admitted += 1;
-          highWaterAtMs = Math.max(highWaterAtMs, message.timestamp.getTime());
+          highWaterAtMs = Math.max(
+            highWaterAtMs,
+            Math.min(messageTimestampMs, scanUpperBoundAtMs),
+          );
           pageAdvanced = true;
         }
         if (pageAdvanced) {
@@ -320,6 +343,7 @@ export async function createAgentMailCatchUpSession(params: {
             key,
             baselineAtMs: storedCursor.baselineAtMs,
             highWaterAtMs,
+            upperBoundAtMs: scanUpperBoundAtMs,
             established: false,
           });
         }
@@ -334,6 +358,7 @@ export async function createAgentMailCatchUpSession(params: {
         key,
         baselineAtMs: storedCursor.baselineAtMs,
         highWaterAtMs,
+        upperBoundAtMs: scanUpperBoundAtMs,
         established: true,
       });
       if (admitted > 0) {