From af0d3ee87c37050659ef6892be529d33c737457a Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 03:02:52 +0000 Subject: [PATCH 1/4] Fix durable AgentMail ingress and plugin loading --- openclaw.plugin.json | 6 -- package-lock.json | 2 +- package.json | 5 +- scripts/build-manifest.mjs | 10 +-- scripts/validate-plugin.mjs | 2 + src/channel/src/accounts.test.ts | 4 +- src/channel/src/accounts.ts | 5 +- src/channel/src/catch-up.test.ts | 58 +++++++++++- src/channel/src/catch-up.ts | 51 +++++++---- src/channel/src/durable-receive.test.ts | 98 +++++++++++++++++++- src/channel/src/durable-receive.ts | 95 +++++++++++++++++--- src/channel/src/gateway.test.ts | 41 ++++++++- src/channel/src/gateway.ts | 81 ++++++++++------- src/channel/src/inbound.test.ts | 79 +++++++++++++++- src/channel/src/inbound.ts | 50 +++++++++-- src/channel/src/ingress.ts | 114 +++++++++++++++++++++--- src/channel/src/media.test.ts | 40 +++++++-- src/channel/src/media.ts | 85 ++++++++++++------ src/channel/src/received-message.ts | 39 ++++++++ src/channel/src/send.test.ts | 37 +++++++- src/channel/src/send.ts | 22 +++-- src/channel/src/webhook.test.ts | 7 +- src/channel/src/webhook.ts | 11 ++- src/channel/src/websocket.test.ts | 32 +++++-- src/channel/src/websocket.ts | 26 ++++-- src/index.ts | 34 +++++++ 26 files changed, 867 insertions(+), 167 deletions(-) create mode 100644 src/channel/src/received-message.ts create mode 100644 src/index.ts diff --git a/openclaw.plugin.json b/openclaw.plugin.json index 2975b49..8fcad10 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -32,12 +32,6 @@ "channels": [ "agentmail" ], - "channelEnvVars": { - "agentmail": [ - "AGENTMAIL_API_KEY", - "AGENTMAIL_WEBHOOK_SECRET" - ] - }, "channelConfigs": { "agentmail": { "schema": { diff --git a/package-lock.json b/package-lock.json index 749cb29..913da3c 100644 --- a/package-lock.json +++ b/package-lock.json @@ -20,7 +20,7 @@ "vitest": "^3.2.0" }, "engines": { - "node": ">=22.12.0" + "node": ">=22.22.3 <23 || >=24.15.0 <25 || >=25.9.0" }, "peerDependencies": { "openclaw": ">=2026.7.2-beta.3" diff --git a/package.json b/package.json index f8edb58..b3e7e8d 100644 --- a/package.json +++ b/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", diff --git a/scripts/build-manifest.mjs b/scripts/build-manifest.mjs index b8f1602..154f44a 100644 --- a/scripts/build-manifest.mjs +++ b/scripts/build-manifest.mjs @@ -1,8 +1,8 @@ // Generates openclaw.plugin.json for the combined AgentMail plugin (tool extension + channel // extension). The stock `openclaw plugins build` codegen only understands tool-plugin metadata, so // it cannot own a manifest that also declares a channel. This script derives the tool half from the -// compiled tool entry's metadata and merges the channel declarations (channels, channelEnvVars, -// channelConfigs) so both surfaces load from one manifest. +// compiled tool entry's metadata and merges the channel declarations (channels and channelConfigs) +// so both surfaces load from one manifest. // // Usage: // node scripts/build-manifest.mjs # write openclaw.plugin.json @@ -31,9 +31,9 @@ const manifest = { configSchema: toolMetadata.configSchema, activation: toolMetadata.activation ?? { onStartup: true }, channels: ["agentmail"], - channelEnvVars: { - agentmail: ["AGENTMAIL_API_KEY", "AGENTMAIL_WEBHOOK_SECRET"], - }, + // Deliberately omit channelEnvVars. AGENTMAIL_API_KEY also configures the tools, while the + // channel requires an inboxId; the host treats any declared non-empty env var as channel + // presence and would otherwise surface an unconfigured phantom channel. channelConfigs: { agentmail: { schema: AgentMailChannelConfigSchema.schema, diff --git a/scripts/validate-plugin.mjs b/scripts/validate-plugin.mjs index c3ed38f..4ed272f 100644 --- a/scripts/validate-plugin.mjs +++ b/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) { diff --git a/src/channel/src/accounts.test.ts b/src/channel/src/accounts.test.ts index c702878..9d48c88 100644 --- a/src/channel/src/accounts.test.ts +++ b/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", diff --git a/src/channel/src/accounts.ts b/src/channel/src/accounts.ts index 763153b..6ed5f1b 100644 --- a/src/channel/src/accounts.ts +++ b/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,7 @@ export function resolveAgentMailAccount( accountId: id, enabled: channel.enabled !== false && account?.enabled !== false, apiKey: apiVal, - inboxId: merged.inboxId?.trim() ?? "", + inboxId: normalizeAgentMailInboxId(merged.inboxId ?? ""), webhookSecret: hookVal, webhookPath: configuredPath || @@ -187,7 +188,7 @@ export function findConflictingAgentMailInboxOwner( if ( other.enabled && isAgentMailAccountConfigured(other) && - other.inboxId === account.inboxId + normalizeAgentMailInboxId(other.inboxId) === normalizeAgentMailInboxId(account.inboxId) ) { return other.accountId; } diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 3972bff..8f34c4d 100644 --- a/src/channel/src/catch-up.test.ts +++ b/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,62 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("skips malformed timestamps without poisoning the catch-up cursor", async () => { + const store = memoryStore(); + 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 expect( + session.run({ receive, abortSignal: new AbortController().signal }), + ).resolves.toBeUndefined(); + await expect( + session.run({ receive, abortSignal: new AbortController().signal }), + ).resolves.toBeUndefined(); + expect(receive).not.toHaveBeenCalled(); + expect(list).toHaveBeenLastCalledWith( + "inbox_1", + expect.objectContaining({ after: new Date(0) }), + expect.any(Object), + ); + }); + + it("clamps a future message timestamp before advancing the high-water cursor", async () => { + const store = memoryStore(); + 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 }); + 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(); const list = vi.fn(async () => ({ diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index 85ccea3..aa3f9ab 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -7,7 +7,10 @@ import { AGENTMAIL_DURABLE_COMPLETED_TTL_MS, AgentMailIngressCapacityError, } from "./durable-receive.js"; -import { AGENTMAIL_RECEIVED_LABEL } from "./inbound.js"; +import { + AGENTMAIL_RECEIVED_LABEL, + 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 +178,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; key: string; baselineAtMs: number; highWaterAtMs: number; + upperBoundAtMs: number; established: boolean; }): Promise { const update = params.store.update; @@ -198,7 +193,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 +209,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 +252,8 @@ export async function createAgentMailCatchUpSession(params: { if (!storedCursor) { throw new Error("AgentMail WebSocket catch-up cursor is unavailable"); } + const runAtMs = now(); + 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 +263,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 { @@ -284,7 +293,11 @@ export async function createAgentMailCatchUpSession(params: { if (abortSignal.aborted) { return; } - if (!isReceivedMessage(message, params.account.inboxId)) { + const messageTimestampMs = resolveReceivedAgentMailMessageTimestampMs( + message, + params.account.inboxId, + ); + if (messageTimestampMs === null) { continue; } try { @@ -293,7 +306,7 @@ export async function createAgentMailCatchUpSession(params: { inboxId: params.account.inboxId, messageId: message.messageId, transport: "rest", - receivedAt: message.timestamp.getTime(), + receivedAt: messageTimestampMs, arrivedAt: now(), }); } catch (error) { @@ -309,7 +322,7 @@ export async function createAgentMailCatchUpSession(params: { throw error; } admitted += 1; - highWaterAtMs = Math.max(highWaterAtMs, message.timestamp.getTime()); + highWaterAtMs = Math.max(highWaterAtMs, Math.min(messageTimestampMs, now())); pageAdvanced = true; } if (pageAdvanced) { @@ -320,6 +333,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, + upperBoundAtMs: now(), established: false, }); } @@ -334,6 +348,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, + upperBoundAtMs: now(), established: true, }); if (admitted > 0) { diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index 01a5081..2bce9cd 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -33,14 +33,20 @@ describe("AgentMail durable ingress", () => { }); it("rejects new ingress at capacity without evicting accepted pending work", async () => { + const id = createAgentMailDurableInboundId(record); const accept = vi.fn(async () => ({ kind: "accepted", duplicate: false, record: {} })); + const deletePending = vi.fn(async () => true); const journal = withAgentMailIngressCapacity( { accept, - // Already holding the single permitted pending row for a different message. - pending: vi.fn(async () => [{ id: "other", payload: record, attempts: 0 }]), + // The queue lookup atomically admitted the new row alongside the existing full-cap row. + pending: vi.fn(async () => [ + { id: "other", payload: record, attempts: 0 }, + { id, payload: record, attempts: 0 }, + ]), complete: vi.fn(), release: vi.fn(), + deletePending, } as never, 1, ); @@ -51,7 +57,27 @@ describe("AgentMail durable ingress", () => { dispatch: vi.fn(), }), ).rejects.toBeInstanceOf(AgentMailIngressCapacityError); - expect(accept).not.toHaveBeenCalled(); + expect(accept).toHaveBeenCalledOnce(); + expect(deletePending).toHaveBeenCalledWith(id); + }); + + it("accepts a completed duplicate at capacity without applying backpressure", async () => { + const accept = vi.fn(async () => ({ kind: "completed", duplicate: true, record: {} })); + const deletePending = vi.fn(); + const journal = withAgentMailIngressCapacity( + { + accept, + pending: vi.fn(async () => [{ id: "other", payload: record, attempts: 0 }]), + complete: vi.fn(), + release: vi.fn(), + deletePending, + } as never, + 1, + ); + await expect( + processAgentMailIngress({ journal: journal as never, record, dispatch: vi.fn() }), + ).resolves.toBe("duplicate"); + expect(deletePending).not.toHaveBeenCalled(); }); it("re-admits an already-pending duplicate even at capacity", async () => { @@ -313,6 +339,72 @@ describe("AgentMail durable ingress", () => { expect(complete).toHaveBeenCalledOnce(); }); + it("keeps a deferred turn pending until adoption or abandonment", async () => { + const complete = vi.fn(async () => undefined); + const release = vi.fn(async () => true); + let deferredLifecycle: + | { + onTurnAdopted: () => Promise; + onTurnDeferred: () => void; + onTurnAbandoned: () => Promise; + } + | undefined; + const dispatch = vi.fn(async (_record: AgentMailIngressRecord, lifecycle: NonNullable) => { + if (dispatch.mock.calls.length === 1) { + deferredLifecycle = lifecycle; + lifecycle.onTurnDeferred(); + return; + } + await lifecycle.onTurnAdopted(); + }); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce()); + expect(complete).not.toHaveBeenCalled(); + expect(release).not.toHaveBeenCalled(); + + await deferredLifecycle?.onTurnAbandoned(); + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(2)); + expect(release).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { + lastError: "deferred turn abandoned before adoption", + }); + expect(complete).toHaveBeenCalledOnce(); + }); + + it("fails a completion marker after its retry ceiling without redispatching", async () => { + 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", + }); + }); + it("does not replay a turn after adoption if later dispatch settlement fails", async () => { let state: "pending" | "completed" = "pending"; const release = vi.fn(async () => true); diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index 48dfb51..8be4e31 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -31,7 +31,12 @@ export function createAgentMailDurableInboundId(params: { // some published releases), so the wrapper stays portable across SDK versions. type AgentMailJournal = ReturnType< typeof createDurableInboundReceiveJournalFromQueue ->; +> & { + fail?: ( + id: string, + options: { reason: string; message?: string; failedAt?: number }, + ) => Promise; +}; /** * Wraps the store-backed journal with an admission cap. The published SDK's queue journal only @@ -55,15 +60,21 @@ export function withAgentMailIngressCapacity( ...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 && !pending.some((entry) => entry.id === id)) { + if (pendingEstimate > maxPendingEntries) { + await journal.deletePending(id); + pendingEstimate -= 1; throw new AgentMailIngressCapacityError(); } - } - const accepted = await journal.accept(id, payload, options); - if (accepted.kind === "accepted") { + } else { pendingEstimate += 1; } return accepted; @@ -88,14 +99,70 @@ export function createAgentMailDurableInboundReceiveJournal(params: { stateDir: runtime.state.resolveStateDir(), }, ); - const journal = createDurableInboundReceiveJournalFromQueue({ - queue, - retention: { - pendingTtlMs: AGENTMAIL_DURABLE_PENDING_TTL_MS, - completedTtlMs: AGENTMAIL_DURABLE_COMPLETED_TTL_MS, - failedTtlMs: AGENTMAIL_DURABLE_PENDING_TTL_MS, - failedMaxEntries: AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES, + const retention = { + pendingTtlMs: AGENTMAIL_DURABLE_PENDING_TTL_MS, + completedTtlMs: AGENTMAIL_DURABLE_COMPLETED_TTL_MS, + failedTtlMs: AGENTMAIL_DURABLE_PENDING_TTL_MS, + failedMaxEntries: AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES, + }; + const prune = async (protectId?: string) => { + await queue.prune({ + ...retention, + ...(protectId ? { protectIds: [protectId] } : {}), + }); + }; + // Keep failed tombstones terminal. The SDK facade currently projects queue `failed` results as + // pending records for compatibility, which would redispatch an already-produced turn after a + // completion-marker failure. This small facade maps them to the existing terminal `completed` + // journal result while retaining the failed record in the underlying queue for diagnostics. + const extendedJournal: AgentMailJournal = { + accept: async (id, payload, options) => { + await prune(); + const result = await queue.enqueue(id.trim(), payload, options); + await prune(id); + if (result.kind === "accepted") { + return { kind: "accepted", duplicate: false, record: result.record }; + } + if (result.kind === "pending" || result.kind === "claimed") { + return { kind: "pending", duplicate: true, record: result.record }; + } + if (result.kind === "completed") { + return { kind: "completed", duplicate: true, record: result.record }; + } + return { + kind: "completed", + duplicate: true, + record: { + id: result.record.id, + channelId: result.record.channelId, + accountId: result.record.accountId, + queueName: result.record.queueName, + completedAt: result.record.failedAt, + }, + }; }, - }); - return withAgentMailIngressCapacity(journal, AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES); + pending: async () => { + await prune(); + return await queue.listPending({ limit: "all" }); + }, + complete: async (id, options) => { + await queue.complete(id, options); + await prune(id); + }, + release: async (id, options) => { + const released = await queue.release(id, options); + await prune(id); + return released; + }, + deletePending: async (id) => await queue.delete(id), + fail: async (id, options) => { + const failed = await queue.fail(id, options); + await prune(id); + return failed; + }, + }; + return withAgentMailIngressCapacity( + extendedJournal, + AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES, + ); } diff --git a/src/channel/src/gateway.test.ts b/src/channel/src/gateway.test.ts index 87cbabd..8173ff6 100644 --- a/src/channel/src/gateway.test.ts +++ b/src/channel/src/gateway.test.ts @@ -12,6 +12,8 @@ const mocks = vi.hoisted(() => ({ catchUpRun: vi.fn(async () => undefined), catchUpRequest: vi.fn(), catchUpSettle: vi.fn(async () => undefined), + createCatchUpSession: vi.fn(async () => ({ run: vi.fn(async () => undefined) })), + registerError: false, webhookReceives: [] as Array<(record: unknown) => Promise>, })); const apiVal = "key"; @@ -33,6 +35,9 @@ vi.mock("openclaw/plugin-sdk/channel-outbound", () => ({ vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ registerPluginHttpRoute: ({ path }: { path: string }) => { + if (mocks.registerError) { + throw new Error("route registration failed"); + } const unregister = vi.fn(); mocks.routes.push({ path, unregister }); return unregister; @@ -44,7 +49,7 @@ vi.mock("./durable-receive.js", () => ({ })); vi.mock("./catch-up.js", () => ({ - createAgentMailCatchUpSession: vi.fn(async () => ({ run: mocks.catchUpRun })), + createAgentMailCatchUpSession: mocks.createCatchUpSession, createAgentMailCatchUpSupervisor: vi.fn(() => ({ request: mocks.catchUpRequest, requestDeep: mocks.catchUpRequest, @@ -213,6 +218,40 @@ describe("AgentMail gateway route ownership", () => { thirdAbort.abort(); await Promise.all([second, third]); }); + + it.each(["initialization", "registration"])( + "releases route ownership when webhook %s fails", + async (failure) => { + mocks.routes.length = 0; + mocks.registerError = failure === "registration"; + mocks.createCatchUpSession.mockReset(); + mocks.createCatchUpSession.mockResolvedValue({ run: mocks.catchUpRun }); + if (failure === "initialization") { + mocks.createCatchUpSession.mockRejectedValueOnce(new Error("state store unavailable")); + } + const path = `/webhooks/agentmail/failure-${failure}`; + await expect( + startAgentMailGatewayAccount({ + cfg: {}, + account: account("first-owner", path), + channelRuntime: {} as never, + abortSignal: new AbortController().signal, + }), + ).rejects.toThrow(); + + mocks.registerError = false; + const controller = new AbortController(); + const replacement = startAgentMailGatewayAccount({ + cfg: {}, + account: account("replacement-owner", path), + channelRuntime: {} as never, + abortSignal: controller.signal, + }); + await vi.waitFor(() => expect(mocks.routes.some((route) => route.path === path)).toBe(true)); + controller.abort(); + await replacement; + }, + ); }); describe("AgentMail security warnings", () => { diff --git a/src/channel/src/gateway.ts b/src/channel/src/gateway.ts index 1f42dad..3159093 100644 --- a/src/channel/src/gateway.ts +++ b/src/channel/src/gateway.ts @@ -89,7 +89,12 @@ export async function startAgentMailGatewayAccount(params: { }); const dispatch = async ( record: AgentMailIngressRecord, - lifecycle: { onTurnAdopted: () => Promise; abortSignal?: AbortSignal }, + lifecycle: { + onTurnAdopted: () => Promise; + onTurnDeferred: () => void; + onTurnAbandoned: () => Promise; + abortSignal?: AbortSignal; + }, ) => await dispatchAgentMailInboundEvent({ cfg: params.cfg, @@ -99,6 +104,8 @@ export async function startAgentMailGatewayAccount(params: { client, log: params.log, onTurnAdopted: lifecycle.onTurnAdopted, + onTurnDeferred: lifecycle.onTurnDeferred, + onTurnAbandoned: lifecycle.onTurnAbandoned, abortSignal: lifecycle.abortSignal, }); const receive = async (record: AgentMailIngressRecord) => { @@ -161,39 +168,49 @@ export async function startAgentMailGatewayAccount(params: { // Claim ownership synchronously here, before the first await, so the check-and-claim is atomic: // two concurrent startups for the same path can no longer both observe it as free. routeOwners.set(path, params.account.accountId); - const catchUpSession = await createAgentMailCatchUpSession({ - account: params.account, - client, - log: params.log, - }); - const catchUpSupervisor = createAgentMailCatchUpSupervisor({ - session: catchUpSession, - receive, - abortSignal: params.abortSignal, - log: params.log, - }); - const receiveWithRecovery = async (record: AgentMailIngressRecord) => { - try { - await receive(record); - } catch (error) { - // Provider retries remain useful, but REST recovery is the durable fallback if the provider - // exhausts them while local admission is full or temporarily unavailable. - catchUpSupervisor.request(); - throw error; - } - }; - const unregister = registerPluginHttpRoute({ - path, - auth: "plugin", - pluginId: "agentmail", - accountId: params.account.accountId, - handler: createAgentMailWebhookHandler({ + let catchUpSupervisor: ReturnType; + let unregister: (() => void) | undefined; + try { + const catchUpSession = await createAgentMailCatchUpSession({ account: params.account, - verifier, - receive: receiveWithRecovery, + client, log: params.log, - }), - }); + }); + catchUpSupervisor = createAgentMailCatchUpSupervisor({ + session: catchUpSession, + receive, + abortSignal: params.abortSignal, + log: params.log, + }); + const receiveWithRecovery = async (record: AgentMailIngressRecord) => { + try { + await receive(record); + } catch (error) { + // Provider retries remain useful, but REST recovery is the durable fallback if the provider + // exhausts them while local admission is full or temporarily unavailable. + catchUpSupervisor.request(); + throw error; + } + }; + unregister = registerPluginHttpRoute({ + path, + auth: "plugin", + pluginId: "agentmail", + accountId: params.account.accountId, + handler: createAgentMailWebhookHandler({ + account: params.account, + verifier, + receive: receiveWithRecovery, + log: params.log, + }), + }); + } catch (error) { + unregister?.(); + if (routeOwners.get(path) === params.account.accountId) { + routeOwners.delete(path); + } + throw error; + } const activeRoute = { path, unregister }; activeRoutes.set(params.account.accountId, activeRoute); // Ownership was already claimed synchronously above. diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts index 5654bba..6225081 100644 --- a/src/channel/src/inbound.test.ts +++ b/src/channel/src/inbound.test.ts @@ -243,7 +243,21 @@ describe("AgentMail REST-authoritative inbound", () => { conversationId: "inbox_1:thread:thread_1", }), ); - const delivery = turn?.delivery as { durable: () => Record }; + expect((turn?.ctxPayload as { reply?: unknown }).reply).toEqual({ + to: "message:message_1", + }); + const delivery = turn?.delivery as { + preparePayload: (payload: Record) => Record; + durable: () => Record; + }; + expect( + delivery.preparePayload({ + text: "reply", + replyToId: "message_1", + replyToTag: false, + replyToCurrent: true, + }), + ).toEqual({ text: "reply" }); expect(delivery.durable()).toMatchObject({ to: "message:message_1", replyToId: "message_1", @@ -284,6 +298,69 @@ describe("AgentMail REST-authoritative inbound", () => { expect(rm).toHaveBeenCalledWith("/tmp/a.bin", { force: true }); }); + it("retains deferred-turn attachments until the queued turn is abandoned", async () => { + rm.mockClear(); + loadAgentMailInboundAttachments.mockResolvedValueOnce({ + paths: ["/tmp/deferred.bin"], + types: ["application/octet-stream"], + }); + let lifecycle: + | { onDeferred: () => void; onAbandoned: () => Promise } + | undefined; + const onTurnDeferred = vi.fn(); + const onTurnAbandoned = vi.fn(async () => undefined); + await dispatchAgentMailInboundEvent({ + cfg: {}, + account, + record, + channelRuntime: { + routing: { resolveAgentRoute: () => ({ agentId: "agent-1" }) }, + inbound: { + buildContext: (ctx: Record) => ctx, + run: async ({ turnAdoptionLifecycle }: { turnAdoptionLifecycle: typeof lifecycle }) => { + lifecycle = turnAdoptionLifecycle; + turnAdoptionLifecycle?.onDeferred(); + }, + }, + session: { resolveStorePath: () => "/tmp/s.json", recordInboundSession: vi.fn() }, + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + } as never, + client: { inboxes: { messages: { get: vi.fn(async () => message()) } } } as never, + onTurnDeferred, + onTurnAbandoned, + }); + + expect(onTurnDeferred).toHaveBeenCalledOnce(); + expect(rm).not.toHaveBeenCalled(); + await lifecycle?.onAbandoned(); + expect(rm).toHaveBeenCalledWith("/tmp/deferred.bin", { force: true }); + expect(onTurnAbandoned).toHaveBeenCalledOnce(); + }); + + it("cleans up attachments when inbound handling resolves without adoption", async () => { + rm.mockClear(); + loadAgentMailInboundAttachments.mockResolvedValueOnce({ + paths: ["/tmp/ignored.bin"], + types: ["application/octet-stream"], + }); + await dispatchAgentMailInboundEvent({ + cfg: {}, + account, + record, + channelRuntime: { + routing: { resolveAgentRoute: () => ({ agentId: "agent-1" }) }, + inbound: { + buildContext: (ctx: Record) => ctx, + run: async () => undefined, + }, + session: { resolveStorePath: () => "/tmp/s.json", recordInboundSession: vi.fn() }, + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + } as never, + client: { inboxes: { messages: { get: vi.fn(async () => message()) } } } as never, + }); + expect(rm).toHaveBeenCalledWith("/tmp/ignored.bin", { force: true }); + }); + it("cleans up attachments when the adoption hook itself fails", async () => { rm.mockClear(); loadAgentMailInboundAttachments.mockResolvedValueOnce({ diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts index 08e7df9..a4be9d4 100644 --- a/src/channel/src/inbound.ts +++ b/src/channel/src/inbound.ts @@ -8,6 +8,11 @@ import type { AgentMailLog } from "./log.js"; import { createAgentMailClient } from "./client.js"; import { isAgentMailSenderAllowed, parseSingleFromMailbox } from "./mailbox.js"; import { AgentMailMediaPolicyError, loadAgentMailInboundAttachments } from "./media.js"; +import { + AGENTMAIL_RECEIVED_LABEL, + agentMailInboxIdsEqual, + resolveAgentMailTimestampMs, +} from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const CHANNEL_ID = "agentmail"; @@ -15,7 +20,6 @@ export const HYDRATION_NOT_FOUND_RETRY_WINDOW_MS = 5 * 60_000; // 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. -export const AGENTMAIL_RECEIVED_LABEL = "received"; const AGENTMAIL_REJECTED_LABELS = ["spam", "blocked", "unauthenticated"]; /** @@ -127,6 +131,8 @@ export async function dispatchAgentMailInboundEvent(params: { client?: AgentMailClient; log?: AgentMailLog; onTurnAdopted?: () => void | Promise; + onTurnDeferred?: () => void; + onTurnAbandoned?: () => void | Promise; abortSignal?: AbortSignal; now?: () => number; }): Promise { @@ -157,7 +163,7 @@ export async function dispatchAgentMailInboundEvent(params: { throw error; } if ( - message.inboxId !== params.account.inboxId || + !agentMailInboxIdsEqual(message.inboxId, params.account.inboxId) || message.messageId !== params.record.messageId || hasRejectedLabel(message) ) { @@ -249,6 +255,7 @@ export async function dispatchAgentMailInboundEvent(params: { }); let turnAdopted = false; + let turnDeferred = false; // Adoption fires when core has made recovery-relevant session/run state durable. The ingress row // is completed here (via params.onTurnAdopted), closing the crash window before agent tools run; // exclusive admission isolates the reply lane per turn, and the abort signal cancels a pre-adoption @@ -260,6 +267,16 @@ export async function dispatchAgentMailInboundEvent(params: { await params.onTurnAdopted?.(); turnAdopted = true; }, + onDeferred: () => { + turnDeferred = true; + params.onTurnDeferred?.(); + }, + onAbandoned: async () => { + if (!turnAdopted && inboundMedia.paths.length > 0) { + await Promise.allSettled(inboundMedia.paths.map((path) => rm(path, { force: true }))); + } + await params.onTurnAbandoned?.(); + }, ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), }; const runPromise = params.channelRuntime.inbound.run({ @@ -273,9 +290,8 @@ export async function dispatchAgentMailInboundEvent(params: { // Hydrated SDK objects may bypass Date validation; fall back to the durable record's arrival // time rather than throwing on an invalid timestamp. timestamp: - raw.timestamp instanceof Date && !Number.isNaN(raw.timestamp.getTime()) - ? raw.timestamp.getTime() - : (params.record.arrivedAt ?? params.record.receivedAt), + resolveAgentMailTimestampMs(raw.timestamp) ?? + (params.record.arrivedAt ?? params.record.receivedAt), rawText: body, textForAgent: body, textForCommands: body, @@ -300,7 +316,11 @@ export async function dispatchAgentMailInboundEvent(params: { routeSessionKey: sessionKey, dispatchSessionKey: sessionKey, }, - reply: { to: target, replyToId: message.messageId }, + // Keep the triggering id on the durable delivery contract below, not on the payload + // context. Core treats a payload-level replyToId as explicitly authored even when it was + // copied from this inbound context; the AgentMail sender intentionally accepts only the + // implicit, host-owned reply binding for the active turn. + reply: { to: target }, message: { rawBody: input.rawText, commandBody: input.textForCommands, @@ -333,6 +353,17 @@ export async function dispatchAgentMailInboundEvent(params: { dispatchReplyWithBufferedBlockDispatcher: params.channelRuntime.reply.dispatchReplyWithBufferedBlockDispatcher, delivery: { + // Core auto-threads final payloads from MessageSid, which makes the payload-level id + // look explicitly authored to the outbound adapter. AgentMail accepts only the + // implicit id owned by the durable delivery contract below, so remove those derived + // payload directives immediately before delivery. + preparePayload: (payload) => { + const prepared = { ...payload }; + delete prepared.replyToId; + delete prepared.replyToTag; + delete prepared.replyToCurrent; + return prepared; + }, durable: () => ({ to: target, replyToId: message.messageId, @@ -361,7 +392,7 @@ export async function dispatchAgentMailInboundEvent(params: { try { await runPromise; } catch (error) { - if (!turnAdopted && inboundMedia.paths.length > 0) { + if (!turnAdopted && !turnDeferred && inboundMedia.paths.length > 0) { // The turn never adopted, so core did not take ownership of these freshly-saved attachment // files. Remove them so a durable retry (which re-downloads a clean set) does not leak one // copy per attempt. @@ -369,4 +400,9 @@ export async function dispatchAgentMailInboundEvent(params: { } throw error; } + if (!turnAdopted && !turnDeferred && inboundMedia.paths.length > 0) { + // A normal return without adoption did not transfer ownership of freshly persisted files. + // Deferred turns retain them until the lifecycle later adopts or abandons the queued turn. + await Promise.allSettled(inboundMedia.paths.map((path) => rm(path, { force: true }))); + } } diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 8d374ed..3e00f45 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -8,7 +8,12 @@ import type { AgentMailIngressRecord } from "./types.js"; // Re-exported for transports (websocket) that catch capacity backpressure by class. export { AgentMailIngressCapacityError }; -type AgentMailJournal = ReturnType; +type AgentMailJournal = ReturnType & { + fail?: ( + id: string, + options: { reason: string; message?: string; failedAt?: number }, + ) => Promise; +}; type DispatchParams = { journal: AgentMailJournal; @@ -24,15 +29,43 @@ type DispatchParams = { export type AgentMailIngressDispatch = ( record: AgentMailIngressRecord, - lifecycle: { onTurnAdopted: () => Promise; abortSignal?: AbortSignal }, + lifecycle: { + onTurnAdopted: () => Promise; + onTurnDeferred: () => void; + onTurnAbandoned: () => Promise; + abortSignal?: AbortSignal; + }, ) => Promise; // Ceiling on pre-adoption dispatch attempts for a single message. A deterministically failing // (poison) message would otherwise retry until its ~30-day TTL, and ~450 such rows would fill the -// pending queue and reject all new mail. On exhaustion the row is dropped (completed) so it stops -// occupying an admission slot. The journal exposes no distinct `fail` op, so completion is the -// terminal marker. Comfortably above the legitimate transient-retry budget the tests exercise. +// pending queue and reject all new mail. On exhaustion the row receives a terminal marker so it +// stops occupying an admission slot. Comfortably above the legitimate transient-retry budget the +// tests exercise. const AGENTMAIL_MAX_DISPATCH_ATTEMPTS = 50; +const AGENTMAIL_MAX_COMPLETION_ATTEMPTS = 50; + +type DeferredOutcome = "adopted" | "abandoned" | "aborted"; + +async function waitForDeferredOutcome( + outcome: Promise>, + abortSignal?: AbortSignal, +): Promise { + if (!abortSignal) { + return await outcome; + } + if (abortSignal.aborted) { + return "aborted"; + } + return await new Promise((resolve) => { + const onAbort = () => resolve("aborted"); + abortSignal.addEventListener("abort", onAbort, { once: true }); + void outcome.then((value) => { + abortSignal.removeEventListener("abort", onAbort); + resolve(value); + }); + }); +} type ActiveDispatch = { task: Promise; @@ -150,6 +183,11 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro while (!params.abortSignal?.aborted) { if (!params.dispatchCompleted) { let turnAdopted = false; + let turnDeferred = false; + let settleDeferred!: (outcome: Exclude) => void; + const deferredOutcome = new Promise>((resolve) => { + settleDeferred = resolve; + }); const onTurnAdopted = async () => { if (turnAdopted) { return; @@ -160,12 +198,50 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro await params.journal.complete(params.id); turnAdopted = true; params.dispatchCompleted = true; + settleDeferred("adopted"); + }; + const onTurnDeferred = () => { + turnDeferred = true; + }; + const onTurnAbandoned = async () => { + if (!turnAdopted) { + settleDeferred("abandoned"); + } }; try { - await params.dispatch(params.record, { onTurnAdopted, abortSignal: params.abortSignal }); + await params.dispatch(params.record, { + onTurnAdopted, + onTurnDeferred, + onTurnAbandoned, + abortSignal: params.abortSignal, + }); if (turnAdopted) { return true; } + if (turnDeferred) { + const outcome = await waitForDeferredOutcome(deferredOutcome, params.abortSignal); + if (outcome === "adopted") { + return true; + } + if (outcome === "aborted") { + return false; + } + attempts += 1; + const lastError = "deferred turn abandoned before adoption"; + const released = await params.journal.release(params.id, { lastError }); + if (!released) { + return true; + } + if ( + !(await waitForRetry( + params.abortSignal, + (params.retryDelay ?? retryDelayMs)(attempts), + )) + ) { + return false; + } + continue; + } params.dispatchCompleted = true; } catch (error) { if (turnAdopted) { @@ -175,14 +251,20 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } attempts += 1; if (attempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { - // Poison message: it has failed deterministically past the retry ceiling. Drop it - // (complete removes it from the pending set) so it stops occupying an admission slot and - // blocking new mail. Surface it so an operator can investigate the underlying failure. + // Poison message: it has failed deterministically past the retry ceiling. Mark it + // terminal so it stops occupying an admission slot and blocking new mail. params.log?.error?.( `AgentMail dropping message ${params.record.messageId} after ${attempts} failed dispatch attempts: ${errorText(error)}`, ); try { - await params.journal.complete(params.id); + if (params.journal.fail) { + await params.journal.fail(params.id, { + reason: "dispatch-attempts-exhausted", + message: errorText(error), + }); + } else { + await params.journal.complete(params.id); + } } catch { // Best effort: TTL pruning still reclaims the row if the terminal marker cannot persist. } @@ -232,6 +314,18 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro return true; } catch { attempts += 1; + if (attempts >= AGENTMAIL_MAX_COMPLETION_ATTEMPTS) { + params.log?.error?.( + `AgentMail failed to persist completion for message ${params.record.messageId} after ${attempts} attempts; marking the ingress row failed`, + ); + if (params.journal.fail) { + return await params.journal.fail(params.id, { + reason: "completion-marker-failed", + message: "AgentMail could not persist the completion marker", + }); + } + return false; + } // Dispatch already produced the agent turn. Keep the row pending and retry only the // idempotent completion marker; releasing and redispatching would duplicate the reply. const shouldRetry = await waitForRetry( diff --git a/src/channel/src/media.test.ts b/src/channel/src/media.test.ts index 691e5aa..d481104 100644 --- a/src/channel/src/media.test.ts +++ b/src/channel/src/media.test.ts @@ -66,8 +66,16 @@ describe("AgentMail inbound attachments", () => { expect(saveMediaBuffer).not.toHaveBeenCalled(); }); - it("skips inline and CID parts", async () => { - const getAttachment = vi.fn(); + it("skips explicit inline parts but retains attachment parts with a Content-ID", async () => { + const getAttachment = vi.fn(async () => ({ + downloadUrl: "https://download.example/cid", + filename: "cid.png", + })); + loadWebMediaRaw.mockResolvedValueOnce({ + buffer: Buffer.from("x"), + contentType: "image/png", + }); + saveMediaBuffer.mockResolvedValueOnce({ path: "/tmp/cid.png", contentType: "image/png" }); await expect( loadAgentMailInboundAttachments({ client: { inboxes: { messages: { getAttachment } } } as never, @@ -75,14 +83,36 @@ describe("AgentMail inbound attachments", () => { messageId: "message_1", attachments: [ { attachmentId: "inline", size: 1, contentDisposition: "inline" }, - { attachmentId: "cid", size: 1, contentId: "image@cid" }, + { + attachmentId: "cid", + size: 1, + contentId: "image@cid", + contentDisposition: "attachment", + }, ], maxBytes: 100, }), - ).resolves.toEqual({ paths: [], types: [] }); - expect(getAttachment).not.toHaveBeenCalled(); + ).resolves.toEqual({ paths: ["/tmp/cid.png"], types: ["image/png"] }); + expect(getAttachment).toHaveBeenCalledOnce(); }); + it.each([undefined, Number.NaN, -1, 1.5])( + "rejects malformed declared attachment size %s before downloading", + async (size) => { + const getAttachment = vi.fn(); + await expect( + loadAgentMailInboundAttachments({ + client: { inboxes: { messages: { getAttachment } } } as never, + inboxId: "inbox_1", + messageId: "message_1", + attachments: [{ attachmentId: "bad-size", size } as never], + maxBytes: 100, + }), + ).rejects.toThrow("invalid declared size"); + expect(getAttachment).not.toHaveBeenCalled(); + }, + ); + it("classifies static attachment size violations as terminal policy rejections", async () => { await expect( loadAgentMailInboundAttachments({ diff --git a/src/channel/src/media.ts b/src/channel/src/media.ts index ec3bd9b..95b6809 100644 --- a/src/channel/src/media.ts +++ b/src/channel/src/media.ts @@ -18,7 +18,29 @@ export type AgentMailInboundMedia = { export class AgentMailMediaPolicyError extends Error {} function isAcceptedAttachment(attachment: AgentMail.Attachment): boolean { - return attachment.contentDisposition !== "inline" && !attachment.contentId; + // Content-ID is not sufficient evidence that a part is inline: ordinary attachment-disposition + // parts may also carry one. Honor the explicit disposition and retain all other files. + return attachment.contentDisposition?.toLocaleLowerCase("en-US") !== "inline"; +} + +function addAttachmentBytesToBudget(params: { + currentBytes: number; + attachmentBytes: number; + maxBytes: number; + errorMessage: string; +}): number { + const nextBytes = params.currentBytes + params.attachmentBytes; + if (!Number.isSafeInteger(nextBytes) || nextBytes > params.maxBytes) { + throw new AgentMailMediaPolicyError(params.errorMessage); + } + return nextBytes; +} + +function rethrowMediaFetchAsPolicy(error: unknown, message: string): never { + if (error instanceof MediaFetchError && error.code === "max_bytes") { + throw new AgentMailMediaPolicyError(message, { cause: error }); + } + throw error; } export async function loadAgentMailInboundAttachments(params: { @@ -31,17 +53,25 @@ export async function loadAgentMailInboundAttachments(params: { const accepted = params.attachments.filter(isAcceptedAttachment); let declaredBytes = 0; for (const attachment of accepted) { - if (attachment.size > params.maxBytes) { - throw new AgentMailMediaPolicyError( - "AgentMail attachment exceeds the configured per-file media limit", - ); + const declaredSize: unknown = attachment.size; + if ( + typeof declaredSize !== "number" || + !Number.isSafeInteger(declaredSize) || + declaredSize < 0 + ) { + throw new AgentMailMediaPolicyError("AgentMail attachment has an invalid declared size"); } - declaredBytes += attachment.size; - if (declaredBytes > params.maxBytes) { + if (declaredSize > params.maxBytes) { throw new AgentMailMediaPolicyError( - "AgentMail attachments exceed the configured aggregate media limit", + "AgentMail attachment exceeds the configured per-file media limit", ); } + declaredBytes = addAttachmentBytesToBudget({ + currentBytes: declaredBytes, + attachmentBytes: declaredSize, + maxBytes: params.maxBytes, + errorMessage: "AgentMail attachments exceed the configured aggregate media limit", + }); } // Download every part before persisting any of them so a failed signed URL never dispatches a @@ -68,15 +98,17 @@ export async function loadAgentMailInboundAttachments(params: { try { loaded = await loadWebMediaRaw(metadata.downloadUrl, { maxBytes: remaining }); } catch (error) { - if (error instanceof MediaFetchError && error.code === "max_bytes") { - throw new AgentMailMediaPolicyError( - "AgentMail attachments exceed the configured aggregate media limit", - { cause: error }, - ); - } - throw error; + rethrowMediaFetchAsPolicy( + error, + "AgentMail attachments exceed the configured aggregate media limit", + ); } - actualBytes += loaded.buffer.byteLength; + actualBytes = addAttachmentBytesToBudget({ + currentBytes: actualBytes, + attachmentBytes: loaded.buffer.byteLength, + maxBytes: params.maxBytes, + errorMessage: "AgentMail attachments exceed the configured aggregate media limit", + }); downloaded.push({ buffer: loaded.buffer, contentType: @@ -138,22 +170,17 @@ export async function loadAgentMailOutboundAttachments(params: { mediaReadFile: params.mediaReadFile, }); } catch (error) { - if (error instanceof MediaFetchError && error.code === "max_bytes") { - throw new AgentMailMediaPolicyError( - "AgentMail outbound attachments exceed the configured aggregate media limit", - { cause: error }, - ); - } - throw error; - } - totalBytes += loaded.buffer.byteLength; - if (totalBytes > params.maxBytes) { - // Defensive backstop: the per-fetch budget above already rejects oversize files, but a lenient - // loader that returns more than requested must still not push the reply past the aggregate. - throw new AgentMailMediaPolicyError( + rethrowMediaFetchAsPolicy( + error, "AgentMail outbound attachments exceed the configured aggregate media limit", ); } + totalBytes = addAttachmentBytesToBudget({ + currentBytes: totalBytes, + attachmentBytes: loaded.buffer.byteLength, + maxBytes: params.maxBytes, + errorMessage: "AgentMail outbound attachments exceed the configured aggregate media limit", + }); // Convert to base64 immediately and let the source buffer go out of scope; only the encoded // representation is retained for the reply payload. attachments.push({ diff --git a/src/channel/src/received-message.ts b/src/channel/src/received-message.ts new file mode 100644 index 0000000..ea4bdf8 --- /dev/null +++ b/src/channel/src/received-message.ts @@ -0,0 +1,39 @@ +import type { AgentMail } from "agentmail"; + +export const AGENTMAIL_RECEIVED_LABEL = "received"; + +export function normalizeAgentMailInboxId(value: string): string { + return value.trim().toLocaleLowerCase("en-US"); +} + +export function agentMailInboxIdsEqual(left: string, right: string): boolean { + return normalizeAgentMailInboxId(left) === normalizeAgentMailInboxId(right); +} + +export function resolveAgentMailTimestampMs(value: unknown): number | null { + const timestampMs = + value instanceof Date + ? value.getTime() + : typeof value === "string" || typeof value === "number" + ? new Date(value).getTime() + : Number.NaN; + return Number.isFinite(timestampMs) && timestampMs >= 0 ? timestampMs : null; +} + +export function resolveReceivedAgentMailMessageTimestampMs( + message: AgentMail.MessageItem | AgentMail.Message, + inboxId: string, +): number | null { + if (!agentMailInboxIdsEqual(message.inboxId, inboxId)) { + return null; + } + const labels = Array.isArray(message.labels) ? message.labels : []; + if ( + !labels.some( + (label) => String(label).toLocaleLowerCase("en-US") === AGENTMAIL_RECEIVED_LABEL, + ) + ) { + return null; + } + return resolveAgentMailTimestampMs(message.timestamp); +} diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts index 071c5c2..5c851db 100644 --- a/src/channel/src/send.test.ts +++ b/src/channel/src/send.test.ts @@ -452,7 +452,7 @@ describe("AgentMail reply-only outbound", () => { expect(reply).not.toHaveBeenCalled(); }); - it("fails fast (non-retryable) when the triggering message was deleted (404)", async () => { + it("retries a recent triggering-message 404 as a provider projection race", async () => { reply.mockClear(); const deletedClient = () => ({ @@ -484,10 +484,43 @@ describe("AgentMail reply-only outbound", () => { } as never, { client: deletedClient(), now: () => now }, ); - expect(result).toMatchObject({ status: "unresolved", retryable: false }); + 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 = { + inboxes: { + messages: { + get: vi.fn(async () => { + throw new AgentMailError({ message: "gone", statusCode: 404 }); + }), + reply, + }, + }, + } as never; + 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", + enqueuedAt: 0, + retryCount: 2, + effectiveReplyToId: "msg_1", + payloads: [{ text: "Hello" }], + } as never, + { client, now: () => now }, + ); + expect(result).toMatchObject({ status: "unresolved", retryable: false }); + }); + it("refuses recovery when the persisted reply target differs", async () => { const now = 10_000; const result = await reconcileAgentMailUnknownSend( diff --git a/src/channel/src/send.ts b/src/channel/src/send.ts index 015af7b..9599781 100644 --- a/src/channel/src/send.ts +++ b/src/channel/src/send.ts @@ -12,6 +12,8 @@ import { createAgentMailClient } from "./client.js"; import { sha256Hex } from "./digest.js"; import { isAgentMailSenderAllowed, parseSingleFromMailbox } from "./mailbox.js"; import { AgentMailMediaPolicyError, loadAgentMailOutboundAttachments } from "./media.js"; +import { agentMailInboxIdsEqual } from "./received-message.js"; +import { HYDRATION_NOT_FOUND_RETRY_WINDOW_MS } from "./inbound.js"; // A media-size policy violation is genuinely terminal: the same payload will always exceed the // limit, so retrying the reply is pointless. @@ -33,8 +35,12 @@ function isHostLocalMediaFailure(error: unknown): boolean { // A 404 while re-hydrating the triggering message means it was deleted; the reply can never be sent, // so fail fast instead of consuming the bounded retry budget before dead-lettering. -function isDeletedTriggerFailure(error: unknown): boolean { - return error instanceof AgentMailError && error.statusCode === 404; +function isDeletedTriggerFailure(error: unknown, recoveryAgeMs: number): boolean { + return ( + error instanceof AgentMailError && + error.statusCode === 404 && + recoveryAgeMs >= HYDRATION_NOT_FOUND_RETRY_WINDOW_MS + ); } // Transient failures during reconciliation (provider 5xx/throttling or network) resolve on their @@ -156,7 +162,7 @@ async function sendBoundAgentMailReply( // inbound, preventing an allowlisted sender from redirecting the agent reply. const triggeringMessage = await client.inboxes.messages.get(account.inboxId, triggeringMessageId); if ( - triggeringMessage.inboxId !== account.inboxId || + !agentMailInboxIdsEqual(triggeringMessage.inboxId, account.inboxId) || triggeringMessage.messageId !== triggeringMessageId ) { throw new Error("AgentMail reply target did not hydrate to the configured inbox and message."); @@ -306,13 +312,17 @@ export async function reconcileAgentMailUnknownSend( if (isTerminalMediaPolicyFailure(error)) { return { status: "unresolved", error: error.message, retryable: false }; } - if (isDeletedTriggerFailure(error)) { - // The triggering message is gone; retrying cannot succeed. Fail fast. + if (isDeletedTriggerFailure(error, recoveryAgeMs)) { + // A 404 that persists beyond the provider projection window is treated as deletion. return { status: "unresolved", error: errorText(error), retryable: false }; } // Host-local media access (dropped recovery handles) and transient hydration/provider failures // both resolve on a later attempt, so keep the queue retrying rather than throwing out. - if (isHostLocalMediaFailure(error) || isTransientReplyFailure(error)) { + if ( + isHostLocalMediaFailure(error) || + isTransientReplyFailure(error) || + (error instanceof AgentMailError && error.statusCode === 404) + ) { return { status: "unresolved", error: errorText(error), retryable: true }; } throw error; diff --git a/src/channel/src/webhook.test.ts b/src/channel/src/webhook.test.ts index e752655..85a6087 100644 --- a/src/channel/src/webhook.test.ts +++ b/src/channel/src/webhook.test.ts @@ -59,7 +59,11 @@ describe("AgentMail webhook", () => { type: "event", event_type: "message.received", event_id: "event_1", - message: { inbox_id: "inbox_1", message_id: "message_1" }, + message: { + inbox_id: "INBOX_1", + message_id: "message_1", + timestamp: "2026-07-15T12:34:56.000Z", + }, }); const res = response(); await createAgentMailWebhookHandler({ account: account(), verifier, receive })( @@ -72,6 +76,7 @@ describe("AgentMail webhook", () => { inboxId: "inbox_1", messageId: "message_1", transport: "webhook", + receivedAt: Date.parse("2026-07-15T12:34:56.000Z"), }), ); }); diff --git a/src/channel/src/webhook.ts b/src/channel/src/webhook.ts index a7e169a..03d82cc 100644 --- a/src/channel/src/webhook.ts +++ b/src/channel/src/webhook.ts @@ -5,6 +5,7 @@ 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 type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const MAX_WEBHOOK_BODY_BYTES = 1024 * 1024; @@ -24,6 +25,7 @@ function respond(res: ServerResponse, status: number, body = ""): true { function parseVerifiedEvent(payload: unknown): { inboxId: string; messageId: string; + receivedAt?: number; } | null { if (!payload || typeof payload !== "object") { return null; @@ -40,7 +42,8 @@ function parseVerifiedEvent(payload: unknown): { if (!inboxId || !messageId) { return null; } - return { inboxId, messageId }; + const receivedAt = resolveAgentMailTimestampMs(mail.timestamp); + return { inboxId, messageId, ...(receivedAt === null ? {} : { receivedAt }) }; } /** @@ -106,7 +109,7 @@ export function createAgentMailWebhookHandler(params: { params.log?.warn?.("AgentMail webhook ignored a malformed signed received event"); return respond(res, 200); } - if (event.inboxId !== params.account.inboxId) { + if (!agentMailInboxIdsEqual(event.inboxId, params.account.inboxId)) { // The signature is valid but this route cannot ever own the inbox. Acknowledge permanently // so provider retries cannot amplify a routing/configuration error. params.log?.warn?.("AgentMail webhook ignored an event for the wrong inbox"); @@ -116,10 +119,10 @@ export function createAgentMailWebhookHandler(params: { const nowMs = Date.now(); await params.receive({ accountId: params.account.accountId, - inboxId: event.inboxId, + inboxId: params.account.inboxId, messageId: event.messageId, transport: "webhook", - receivedAt: nowMs, + receivedAt: event.receivedAt ?? nowMs, arrivedAt: nowMs, }); return respond(res, 200); diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts index ff75d2a..ccd9e8a 100644 --- a/src/channel/src/websocket.test.ts +++ b/src/channel/src/websocket.test.ts @@ -74,8 +74,9 @@ describe("AgentMail WebSocket ingress", () => { eventType: "message.received", eventId: "event_1", message: { - inboxId: "inbox_1", + inboxId: "INBOX_1", messageId: "message_1", + labels: ["received"], timestamp: new Date(1_234), }, }); @@ -132,6 +133,7 @@ describe("AgentMail WebSocket ingress", () => { message: { inboxId: "inbox_1", messageId: "message_retry", + labels: ["received"], timestamp: new Date(1_234), }, }); @@ -162,13 +164,23 @@ describe("AgentMail WebSocket ingress", () => { handlers.get("message")?.({ type: "event", eventType: "message.received", - message: { inboxId: "inbox_1", messageId: "message_1", timestamp: new Date(1_234) }, + message: { + inboxId: "inbox_1", + messageId: "message_1", + labels: ["received"], + timestamp: new Date(1_234), + }, }); await vi.waitFor(() => expect(receive).toHaveBeenCalledOnce()); handlers.get("message")?.({ type: "event", eventType: "message.received", - message: { inboxId: "inbox_1", messageId: "message_2", timestamp: new Date(1_235) }, + message: { + inboxId: "inbox_1", + messageId: "message_2", + labels: ["received"], + timestamp: new Date(1_235), + }, }); await vi.waitFor(() => expect(catchUpRun).toHaveBeenCalledOnce()); expect(receive).toHaveBeenCalledOnce(); @@ -198,12 +210,22 @@ describe("AgentMail WebSocket ingress", () => { handlers.get("message")?.({ type: "event", eventType: "message.received", - message: { inboxId: "inbox_1", messageId: "message_full", timestamp: new Date(1_234) }, + message: { + inboxId: "inbox_1", + messageId: "message_full", + labels: ["received"], + timestamp: new Date(1_234), + }, }); handlers.get("message")?.({ type: "event", eventType: "message.received", - message: { inboxId: "inbox_1", messageId: "message_next", timestamp: new Date(1_235) }, + message: { + inboxId: "inbox_1", + messageId: "message_next", + labels: ["received"], + timestamp: new Date(1_235), + }, }); await vi.waitFor(() => expect(receive).toHaveBeenCalledTimes(2)); diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts index 890f127..c41b233 100644 --- a/src/channel/src/websocket.ts +++ b/src/channel/src/websocket.ts @@ -9,6 +9,10 @@ import { type AgentMailLog, errorText } from "./log.js"; import { createAgentMailClient } from "./client.js"; import { AgentMailIngressCapacityError } from "./ingress.js"; import { createBackoff, waitForRetry } from "./retry.js"; +import { + agentMailInboxIdsEqual, + resolveReceivedAgentMailMessageTimestampMs, +} from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const AGENTMAIL_WEBSOCKET_LIVE_QUEUE_MAX = 32; @@ -25,16 +29,13 @@ function isReceivedEvent(value: unknown): value is AgentMail.MessageReceivedEven if (event.type !== "event" || event.eventType !== "message.received") { return false; } - // Fully validate the message shape here so downstream field access (inboxId, messageId, - // timestamp.getTime()) cannot throw on a malformed frame. A malformed live event is ignored; REST - // catch-up still recovers the message from the provider. + // Validate the stable identifier shape here. The shared received-message predicate below owns + // inbox, label, and timestamp validation for both WebSocket and REST recovery. const message = event.message as Partial | undefined; return Boolean( message && typeof message.inboxId === "string" && - typeof message.messageId === "string" && - message.timestamp instanceof Date && - !Number.isNaN(message.timestamp.getTime()), + typeof message.messageId === "string", ); } @@ -163,10 +164,17 @@ export async function startAgentMailWebSocket(params: { if (!isReceivedEvent(event)) { return; } - if (event.message.inboxId !== params.account.inboxId) { + if (!agentMailInboxIdsEqual(event.message.inboxId, params.account.inboxId)) { params.log?.warn?.("AgentMail WebSocket ignored an event for the wrong inbox"); return; } + const messageTimestampMs = resolveReceivedAgentMailMessageTimestampMs( + event.message, + params.account.inboxId, + ); + if (messageTimestampMs === null) { + return; + } if (queuedMessageIds.has(event.message.messageId)) { return; } @@ -180,10 +188,10 @@ export async function startAgentMailWebSocket(params: { queuedMessageIds.add(event.message.messageId); liveQueue.push({ accountId: params.account.accountId, - inboxId: event.message.inboxId, + inboxId: params.account.inboxId, messageId: event.message.messageId, transport: "websocket", - receivedAt: event.message.timestamp.getTime(), + receivedAt: messageTimestampMs, arrivedAt: Date.now(), }); runLiveWorker(); diff --git a/src/index.ts b/src/index.ts new file mode 100644 index 0000000..be77551 --- /dev/null +++ b/src/index.ts @@ -0,0 +1,34 @@ +import { + definePluginEntry, + type OpenClawPluginDefinition, +} from "openclaw/plugin-sdk/plugin-entry"; +import { agentMailPlugin } from "./channel/channel-plugin-api.js"; +import { setAgentMailRuntime } from "./channel/api.js"; +import toolEntry from "./tools/index.js"; + +// One runtime entry must own both surfaces. Multiple `openclaw.extensions` entries are a plugin +// pack and receive distinct runtime identities; listing the tool and channel entries separately +// caused the shared manifest id to resolve to the first (tool-only) runtime at gateway startup. +const entry: OpenClawPluginDefinition = definePluginEntry({ + id: "agentmail", + name: "AgentMail", + description: + "AgentMail for OpenClaw: email tools plus a durable, allowlisted, reply-only email channel.", + configSchema: toolEntry.configSchema, + register(api) { + if (api.registrationMode === "cli-metadata") { + return; + } + if (api.registrationMode === "tool-discovery") { + toolEntry.register(api); + return; + } + api.registerChannel({ plugin: agentMailPlugin as never }); + setAgentMailRuntime(api.runtime); + if (api.registrationMode === "full" || api.registrationMode === "discovery") { + toolEntry.register(api); + } + }, +}); + +export default entry; From 0a2192cbde43d06dd0fc05059aa222ac94fbed6a Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 04:06:07 +0000 Subject: [PATCH 2/4] Address AgentMail durability review feedback --- README.md | 11 +-- src/channel/src/accounts.test.ts | 2 +- src/channel/src/accounts.ts | 5 +- src/channel/src/catch-up.test.ts | 20 ++--- src/channel/src/catch-up.ts | 11 +-- src/channel/src/durable-receive.test.ts | 36 +++++++++ src/channel/src/gateway.test.ts | 103 +++++++++++++++++++++++- src/channel/src/gateway.ts | 58 +++++++++---- src/channel/src/ingress.ts | 19 +++++ src/channel/src/media.test.ts | 3 + src/channel/src/media.ts | 8 +- src/channel/src/received-message.ts | 21 +++-- src/channel/src/send.test.ts | 62 ++++++++++++++ src/channel/src/send.ts | 20 +++-- src/channel/src/websocket.test.ts | 6 +- src/channel/src/websocket.ts | 26 +++--- src/tools/client.ts | 15 +++- src/tools/index.test.ts | 39 +++++++-- src/tools/index.ts | 18 ++--- 19 files changed, 394 insertions(+), 89 deletions(-) diff --git a/README.md b/README.md index de905cf..4b542c7 100644 --- a/README.md +++ b/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`: diff --git a/src/channel/src/accounts.test.ts b/src/channel/src/accounts.test.ts index 9d48c88..d66705f 100644 --- a/src/channel/src/accounts.test.ts +++ b/src/channel/src/accounts.test.ts @@ -34,7 +34,7 @@ describe("AgentMail account config", () => { expect(account).toMatchObject({ accountId: "default", apiKey: apiVal, - inboxId: "agent@agentmail.to", + inboxId: "Agent@AgentMail.TO", webhookSecret: hookVal, webhookPath: "/webhooks/agentmail", dmPolicy: "allowlist", diff --git a/src/channel/src/accounts.ts b/src/channel/src/accounts.ts index 6ed5f1b..a473c91 100644 --- a/src/channel/src/accounts.ts +++ b/src/channel/src/accounts.ts @@ -148,7 +148,10 @@ export function resolveAgentMailAccount( accountId: id, enabled: channel.enabled !== false && account?.enabled !== false, apiKey: apiVal, - inboxId: normalizeAgentMailInboxId(merged.inboxId ?? ""), + // 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 || diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 8f34c4d..35d715f 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -173,7 +173,7 @@ describe("AgentMail durable REST catch-up", () => { ); }); - it("skips malformed timestamps without poisoning the catch-up cursor", async () => { + it("admits malformed timestamps using local time without poisoning the cursor", async () => { const store = memoryStore(); const malformed = message({ id: "bad_timestamp", timestamp: 1_100 }) as AgentMail.MessageItem; (malformed as { timestamp: unknown }).timestamp = new Date(Number.NaN); @@ -186,17 +186,13 @@ describe("AgentMail durable REST catch-up", () => { }); const receive = vi.fn(async () => undefined); - await expect( - session.run({ receive, abortSignal: new AbortController().signal }), - ).resolves.toBeUndefined(); - await expect( - session.run({ receive, abortSignal: new AbortController().signal }), - ).resolves.toBeUndefined(); - expect(receive).not.toHaveBeenCalled(); - expect(list).toHaveBeenLastCalledWith( - "inbox_1", - expect.objectContaining({ after: new Date(0) }), - expect.any(Object), + await session.run({ receive, abortSignal: new AbortController().signal }); + expect(receive).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "bad_timestamp", + receivedAt: 1_000, + arrivedAt: 1_000, + }), ); }); diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index aa3f9ab..84cb012 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -9,6 +9,7 @@ import { } from "./durable-receive.js"; import { AGENTMAIL_RECEIVED_LABEL, + isReceivedAgentMailMessage, resolveReceivedAgentMailMessageTimestampMs, } from "./received-message.js"; import { createBackoff, waitForRetry } from "./retry.js"; @@ -293,13 +294,13 @@ export async function createAgentMailCatchUpSession(params: { if (abortSignal.aborted) { return; } - const messageTimestampMs = resolveReceivedAgentMailMessageTimestampMs( - message, - params.account.inboxId, - ); - if (messageTimestampMs === null) { + 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) ?? now(); try { await receive({ accountId: params.account.accountId, diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index 2bce9cd..dc145d0 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -379,6 +379,42 @@ describe("AgentMail durable ingress", () => { expect(complete).toHaveBeenCalledOnce(); }); + it("fails repeatedly abandoned deferred turns at the dispatch ceiling", async () => { + const release = vi.fn(async () => true); + const fail = vi.fn(async () => true); + const dispatch = vi.fn( + async ( + _record: AgentMailIngressRecord, + lifecycle: { + onTurnDeferred: () => void; + onTurnAbandoned: () => Promise; + }, + ) => { + lifecycle.onTurnDeferred(); + await lifecycle.onTurnAbandoned(); + }, + ); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete: vi.fn(), + release, + fail, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + + await vi.waitFor(() => expect(fail).toHaveBeenCalledOnce()); + expect(dispatch).toHaveBeenCalledTimes(50); + expect(release).toHaveBeenCalledTimes(49); + expect(fail).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { + reason: "dispatch-attempts-exhausted", + message: "deferred turn abandoned before adoption", + }); + }); + it("fails a completion marker after its retry ceiling without redispatching", async () => { const dispatch = vi.fn(async () => undefined); const complete = vi.fn(async () => { diff --git a/src/channel/src/gateway.test.ts b/src/channel/src/gateway.test.ts index 8173ff6..1ac5887 100644 --- a/src/channel/src/gateway.test.ts +++ b/src/channel/src/gateway.test.ts @@ -6,7 +6,11 @@ import { import type { ResolvedAgentMailAccount } from "./types.js"; const mocks = vi.hoisted(() => ({ - routes: [] as Array<{ path: string; unregister: ReturnType }>, + routes: [] as Array<{ + path: string; + replaceExisting?: boolean; + unregister: ReturnType; + }>, startWebSocket: vi.fn(async () => undefined), processIngress: vi.fn(async () => "accepted"), catchUpRun: vi.fn(async () => undefined), @@ -34,12 +38,18 @@ vi.mock("openclaw/plugin-sdk/channel-outbound", () => ({ })); vi.mock("openclaw/plugin-sdk/webhook-ingress", () => ({ - registerPluginHttpRoute: ({ path }: { path: string }) => { + registerPluginHttpRoute: ({ + path, + replaceExisting, + }: { + path: string; + replaceExisting?: boolean; + }) => { if (mocks.registerError) { throw new Error("route registration failed"); } const unregister = vi.fn(); - mocks.routes.push({ path, unregister }); + mocks.routes.push({ path, replaceExisting, unregister }); return unregister; }, })); @@ -201,6 +211,7 @@ describe("AgentMail gateway route ownership", () => { abortSignal: secondAbort.signal, }); await vi.waitFor(() => expect(mocks.routes).toHaveLength(2)); + expect(mocks.routes[1]?.replaceExisting).toBe(true); expect(mocks.routes[0]?.unregister).toHaveBeenCalledOnce(); firstAbort.abort(); @@ -252,6 +263,92 @@ describe("AgentMail gateway route ownership", () => { await replacement; }, ); + + it.each(["initialization", "registration"])( + "keeps the predecessor route active when replacement %s fails", + async (failure) => { + mocks.routes.length = 0; + mocks.registerError = false; + mocks.createCatchUpSession.mockReset(); + mocks.createCatchUpSession.mockResolvedValue({ run: mocks.catchUpRun }); + const controller = new AbortController(); + const running = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/stable"), + channelRuntime: {} as never, + abortSignal: controller.signal, + }); + await vi.waitFor(() => expect(mocks.routes).toHaveLength(1)); + + if (failure === "initialization") { + mocks.createCatchUpSession.mockRejectedValueOnce(new Error("state store unavailable")); + } else { + mocks.registerError = true; + } + await expect( + startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/replacement"), + channelRuntime: {} as never, + abortSignal: new AbortController().signal, + }), + ).rejects.toThrow(); + mocks.registerError = false; + expect(mocks.routes[0]?.unregister).not.toHaveBeenCalled(); + + controller.abort(); + await running; + expect(mocks.routes[0]?.unregister).toHaveBeenCalledOnce(); + }, + ); + + it("does not let an older startup failure release a newer route claim", async () => { + mocks.routes.length = 0; + mocks.registerError = false; + mocks.createCatchUpSession.mockReset(); + let rejectOlder!: (error: Error) => void; + mocks.createCatchUpSession + .mockImplementationOnce( + async () => + await new Promise((_resolve, reject) => { + rejectOlder = reject; + }), + ) + .mockResolvedValueOnce({ run: mocks.catchUpRun }); + + const olderAbort = new AbortController(); + const newerAbort = new AbortController(); + const older = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/race"), + channelRuntime: {} as never, + abortSignal: olderAbort.signal, + }); + const olderResult = expect(older).rejects.toThrow("older startup failed"); + await vi.waitFor(() => expect(mocks.createCatchUpSession).toHaveBeenCalledTimes(1)); + const newer = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/race"), + channelRuntime: {} as never, + abortSignal: newerAbort.signal, + }); + await vi.waitFor(() => expect(mocks.routes).toHaveLength(1)); + + rejectOlder(new Error("older startup failed")); + await olderResult; + await expect( + startAgentMailGatewayAccount({ + cfg: {}, + account: account("billing", "/webhooks/agentmail/race"), + channelRuntime: {} as never, + abortSignal: new AbortController().signal, + }), + ).rejects.toThrow("already registered by account support"); + + olderAbort.abort(); + newerAbort.abort(); + await newer; + }); }); describe("AgentMail security warnings", () => { diff --git a/src/channel/src/gateway.ts b/src/channel/src/gateway.ts index 3159093..fe72ca6 100644 --- a/src/channel/src/gateway.ts +++ b/src/channel/src/gateway.ts @@ -20,10 +20,11 @@ import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.j import { createAgentMailWebhookHandler, createAgentMailWebhookVerifier } from "./webhook.js"; import { startAgentMailWebSocket } from "./websocket.js"; -type ActiveRoute = { path: string; unregister: () => void }; +type RouteClaim = { accountId: string; generation: symbol; valid: boolean }; +type ActiveRoute = { path: string; unregister: () => void; claim: RouteClaim }; const activeRoutes = new Map(); -const routeOwners = new Map(); +const routeOwners = new Map(); // Single source of truth for AgentMail sender-authorization warnings, shared by gateway startup // diagnostics and the channel security surface so the two never drift. @@ -153,21 +154,21 @@ export async function startAgentMailGatewayAccount(params: { ? params.account.webhookPath : `/${params.account.webhookPath}`; const owner = routeOwners.get(path); - if (owner && owner !== params.account.accountId) { + if (owner && owner.accountId !== params.account.accountId) { throw new Error( - `AgentMail webhook path ${path} is already registered by account ${owner}; configure a distinct webhookPath.`, + `AgentMail webhook path ${path} is already registered by account ${owner.accountId}; configure a distinct webhookPath.`, ); } const previousRoute = activeRoutes.get(params.account.accountId); - if (previousRoute) { - previousRoute.unregister(); - if (routeOwners.get(previousRoute.path) === params.account.accountId) { - routeOwners.delete(previousRoute.path); - } - } + const claim: RouteClaim = { + accountId: params.account.accountId, + generation: Symbol(params.account.accountId), + valid: true, + }; // Claim ownership synchronously here, before the first await, so the check-and-claim is atomic: - // two concurrent startups for the same path can no longer both observe it as free. - routeOwners.set(path, params.account.accountId); + // two concurrent startups for the same path can no longer both observe it as free. Keep the + // predecessor registered until the replacement is fully initialized and registered. + routeOwners.set(path, claim); let catchUpSupervisor: ReturnType; let unregister: (() => void) | undefined; try { @@ -176,6 +177,12 @@ export async function startAgentMailGatewayAccount(params: { client, log: params.log, }); + // A newer overlapping startup claimed this path while initialization awaited. Its generation + // owns registration; park this stale invocation until its lifecycle is cancelled. + if (routeOwners.get(path) !== claim) { + claim.valid = false; + return await waitUntilAbort(params.abortSignal); + } catchUpSupervisor = createAgentMailCatchUpSupervisor({ session: catchUpSession, receive, @@ -203,17 +210,33 @@ export async function startAgentMailGatewayAccount(params: { receive: receiveWithRecovery, log: params.log, }), + // Replacing first is safe: OpenClaw removes the predecessor entry atomically, and its stale + // unregister handle becomes a no-op. This avoids a route outage if replacement setup fails. + replaceExisting: true, }); } catch (error) { + claim.valid = false; unregister?.(); - if (routeOwners.get(path) === params.account.accountId) { - routeOwners.delete(path); + // A stale startup must never release a newer startup's claim. Restore a still-valid predecessor + // on same-path rollback; otherwise release only this exact generation. + if (routeOwners.get(path) === claim) { + if (owner?.valid) { + routeOwners.set(path, owner); + } else { + routeOwners.delete(path); + } } throw error; } - const activeRoute = { path, unregister }; + const activeRoute = { path, unregister, claim }; activeRoutes.set(params.account.accountId, activeRoute); - // Ownership was already claimed synchronously above. + if (previousRoute && previousRoute !== activeRoute) { + previousRoute.claim.valid = false; + previousRoute.unregister(); + if (routeOwners.get(previousRoute.path) === previousRoute.claim) { + routeOwners.delete(previousRoute.path); + } + } params.log?.info?.( `Registered AgentMail webhook route ${path} for account ${params.account.accountId}`, ); @@ -228,9 +251,10 @@ export async function startAgentMailGatewayAccount(params: { await waitUntilAbort(params.abortSignal, () => { // A replaced account invocation can abort later; it must not delete the newer registration. if (activeRoutes.get(params.account.accountId) === activeRoute) { + claim.valid = false; unregister(); activeRoutes.delete(params.account.accountId); - if (routeOwners.get(path) === params.account.accountId) { + if (routeOwners.get(path) === claim) { routeOwners.delete(path); } } diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 3e00f45..116cbab 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -228,6 +228,25 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } attempts += 1; const lastError = "deferred turn abandoned before adoption"; + if (attempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { + params.log?.error?.( + `AgentMail dropping message ${params.record.messageId} after ${attempts} abandoned deferred turns`, + ); + try { + if (params.journal.fail) { + await params.journal.fail(params.id, { + reason: "dispatch-attempts-exhausted", + message: lastError, + }); + } else { + await params.journal.complete(params.id); + } + } catch { + // Best effort: TTL pruning still reclaims the row if the terminal marker cannot + // persist. + } + return true; + } const released = await params.journal.release(params.id, { lastError }); if (!released) { return true; diff --git a/src/channel/src/media.test.ts b/src/channel/src/media.test.ts index d481104..a68595e 100644 --- a/src/channel/src/media.test.ts +++ b/src/channel/src/media.test.ts @@ -83,6 +83,9 @@ describe("AgentMail inbound attachments", () => { messageId: "message_1", attachments: [ { attachmentId: "inline", size: 1, contentDisposition: "inline" }, + // Bare Content-ID parts are embedded HTML media and must not consume the attachment + // budget. If accepted, this declared size would reject the whole set. + { attachmentId: "embedded-cid", size: 1_000, contentId: "logo@cid" }, { attachmentId: "cid", size: 1, diff --git a/src/channel/src/media.ts b/src/channel/src/media.ts index 95b6809..df86ae5 100644 --- a/src/channel/src/media.ts +++ b/src/channel/src/media.ts @@ -18,9 +18,11 @@ export type AgentMailInboundMedia = { export class AgentMailMediaPolicyError extends Error {} function isAcceptedAttachment(attachment: AgentMail.Attachment): boolean { - // Content-ID is not sufficient evidence that a part is inline: ordinary attachment-disposition - // parts may also carry one. Honor the explicit disposition and retain all other files. - return attachment.contentDisposition?.toLocaleLowerCase("en-US") !== "inline"; + const disposition = attachment.contentDisposition?.toLocaleLowerCase("en-US"); + // An explicit attachment disposition wins even when the part also has a Content-ID. A bare + // Content-ID is how embedded HTML images are commonly represented, so exclude those from the + // user's attachment set and aggregate budget. + return disposition === "attachment" || (disposition !== "inline" && !attachment.contentId); } function addAttachmentBytesToBudget(params: { diff --git a/src/channel/src/received-message.ts b/src/channel/src/received-message.ts index ea4bdf8..c69314f 100644 --- a/src/channel/src/received-message.ts +++ b/src/channel/src/received-message.ts @@ -20,19 +20,24 @@ export function resolveAgentMailTimestampMs(value: unknown): number | null { return Number.isFinite(timestampMs) && timestampMs >= 0 ? timestampMs : null; } -export function resolveReceivedAgentMailMessageTimestampMs( +export function isReceivedAgentMailMessage( message: AgentMail.MessageItem | AgentMail.Message, inboxId: string, -): number | null { +): boolean { if (!agentMailInboxIdsEqual(message.inboxId, inboxId)) { - return null; + return false; } const labels = Array.isArray(message.labels) ? message.labels : []; - if ( - !labels.some( - (label) => String(label).toLocaleLowerCase("en-US") === AGENTMAIL_RECEIVED_LABEL, - ) - ) { + return labels.some( + (label) => String(label).toLocaleLowerCase("en-US") === AGENTMAIL_RECEIVED_LABEL, + ); +} + +export function resolveReceivedAgentMailMessageTimestampMs( + message: AgentMail.MessageItem | AgentMail.Message, + inboxId: string, +): number | null { + if (!isReceivedAgentMailMessage(message, inboxId)) { return null; } return resolveAgentMailTimestampMs(message.timestamp); diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts index 5c851db..5deb2b7 100644 --- a/src/channel/src/send.test.ts +++ b/src/channel/src/send.test.ts @@ -416,6 +416,68 @@ describe("AgentMail reply-only outbound", () => { expect(reply).not.toHaveBeenCalled(); }); + it("filters empty rendered media before applying the no-content verdict", async () => { + reply.mockClear(); + const now = 10_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", + enqueuedAt: now - 1_000, + retryCount: 1, + effectiveReplyToId: "msg_1", + payloads: [{ text: " " }], + renderedBatchPlan: { + payloadCount: 1, + textCount: 0, + mediaCount: 1, + voiceCount: 0, + presentationCount: 0, + interactiveCount: 0, + channelDataCount: 0, + items: [{ index: 0, kinds: ["media"], text: " ", mediaUrls: [""] }], + }, + } as never, + { client: client(), now: () => now }, + ); + + expect(result).toEqual({ status: "not_sent" }); + expect(reply).not.toHaveBeenCalled(); + }); + + it("returns a terminal structured verdict for a malformed recovery target", async () => { + const now = 10_000; + const result = await reconcileAgentMailUnknownSend( + { + cfg: {}, + queueId: "queue_1", + channel: "agentmail", + to: "person@example.com", + accountId: "default", + enqueuedAt: now - 1_000, + retryCount: 1, + effectiveReplyToId: "msg_1", + payloads: [{ text: "Hello" }], + } as never, + { client: client(), now: () => now }, + ); + + expect(result).toEqual({ + status: "unresolved", + error: + "AgentMail target must be message:; new threads and recipients are not supported.", + retryable: false, + }); + expect(reply).not.toHaveBeenCalled(); + }); + it("maps a transient hydration failure to a retryable verdict", async () => { reply.mockClear(); const transientClient = () => diff --git a/src/channel/src/send.ts b/src/channel/src/send.ts index 9599781..ad9d426 100644 --- a/src/channel/src/send.ts +++ b/src/channel/src/send.ts @@ -255,7 +255,16 @@ export async function reconcileAgentMailUnknownSend( return { status: "not_sent" }; } const rendered = ctx.renderedBatchPlan?.items[0]; - const triggeringMessageId = parseAgentMailMessageTarget(ctx.to); + const normalizedTarget = normalizeAgentMailTarget(ctx.to); + if (!normalizedTarget) { + return { + status: "unresolved", + error: + "AgentMail target must be message:; new threads and recipients are not supported.", + retryable: false, + }; + } + const triggeringMessageId = normalizedTarget.slice(TARGET_PREFIX.length); if (ctx.effectiveReplyToId !== triggeringMessageId) { return { status: "unresolved", @@ -274,11 +283,12 @@ export async function reconcileAgentMailUnknownSend( } // A rendered plan is authoritative even when its media list is empty: capability filtering may // intentionally have removed media that still appears on the original queued payload. - const mediaUrls = rendered + const mediaUrls = (rendered ? [...rendered.mediaUrls] - : [payload.mediaUrl, ...(payload.mediaUrls ?? [])].filter((value): value is string => - Boolean(value), - ); + : [payload.mediaUrl, ...(payload.mediaUrls ?? [])] + ).filter( + (value): value is string => typeof value === "string" && value.trim().length > 0, + ); const text = rendered?.text ?? payload.text ?? ""; const recoveredPayload = { ...payload, text }; delete recoveredPayload.mediaUrl; diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts index ccd9e8a..6d9ef30 100644 --- a/src/channel/src/websocket.test.ts +++ b/src/channel/src/websocket.test.ts @@ -250,9 +250,11 @@ describe("AgentMail WebSocket ingress", () => { await running; }); - it("reports SDK errors and schedules authoritative catch-up", async () => { + it("reconnects after an SDK error even when no close event follows", async () => { handlers.clear(); catchUpRun.mockClear(); + connect.mockClear(); + close.mockClear(); const error = vi.fn(); const controller = new AbortController(); const running = startAgentMailWebSocket({ @@ -260,6 +262,7 @@ describe("AgentMail WebSocket ingress", () => { abortSignal: controller.signal, receive: vi.fn(async () => undefined), catchUpSession: { run: catchUpRun }, + reconnectDelayMs: () => 0, log: { error }, }); await vi.waitFor(() => expect(handlers.has("error")).toBe(true)); @@ -267,6 +270,7 @@ describe("AgentMail WebSocket ingress", () => { handlers.get("error")?.(new Error("frame parse failed")); await vi.waitFor(() => expect(catchUpRun).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); expect(error).toHaveBeenCalledWith( "AgentMail WebSocket error for account default: frame parse failed", ); diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts index c41b233..b5f2b18 100644 --- a/src/channel/src/websocket.ts +++ b/src/channel/src/websocket.ts @@ -1,4 +1,5 @@ import type { AgentMail, AgentMailClient } from "agentmail"; +import { waitUntilAbort } from "openclaw/plugin-sdk/channel-outbound"; import { createAgentMailCatchUpSession, createAgentMailCatchUpSupervisor, @@ -204,13 +205,7 @@ export async function startAgentMailWebSocket(params: { // // One shared abort promise for the whole loop: abort is terminal, so attaching a fresh listener // per reconnect would leak closures on the long-lived signal (listener-limit warnings under churn). - const aborted = new Promise((resolve) => { - if (params.abortSignal.aborted) { - resolve(); - return; - } - params.abortSignal.addEventListener("abort", () => resolve(), { once: true }); - }); + const aborted = waitUntilAbort(params.abortSignal); const connectionLoop = (async () => { let reconnectAttempt = 0; while (!params.abortSignal.aborted) { @@ -254,14 +249,20 @@ export async function startAgentMailWebSocket(params: { catchUpSupervisor.request(); }; const closed = new Promise((resolve) => { + let settled = false; + const settleClosed = () => { + if (settled) { + return; + } + settled = true; + subscribedForCurrentConnection = false; + resolve(); + }; socket.on("open", () => { reconnectAttempt = 0; subscribe(); }); - socket.on("close", () => { - subscribedForCurrentConnection = false; - resolve(); - }); + socket.on("close", settleClosed); socket.on("error", (error) => { params.log?.error?.( `AgentMail WebSocket error for account ${params.account.accountId}: ${errorText(error)}`, @@ -269,6 +270,9 @@ export async function startAgentMailWebSocket(params: { // Parsing/transport errors may not close the socket. Recover authoritative events even // when the socket stays connected and emits no close. catchUpSupervisor.request(); + // Fatal socket errors do not always emit a later close. Treat either event as terminal + // for this connection so the outer loop recreates it. + settleClosed(); }); socket.on("message", handleMessage); }); diff --git a/src/tools/client.ts b/src/tools/client.ts index 53c08fe..9d0e079 100644 --- a/src/tools/client.ts +++ b/src/tools/client.ts @@ -1,5 +1,7 @@ import { AgentMailClient } from "agentmail"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { Type, type Static } from "typebox"; +import { resolveAgentMailAccount } from "../channel/src/accounts.js"; export const agentMailConfigSchema = Type.Object( { @@ -29,12 +31,19 @@ export const agentMailConfigSchema = Type.Object( export type AgentMailConfig = Static; -export function createAgentMailClient(config: AgentMailConfig): AgentMailClient { - const apiKey = process.env.AGENTMAIL_API_KEY?.trim(); +export function createAgentMailClient( + config: AgentMailConfig, + hostConfig?: OpenClawConfig, +): AgentMailClient { + // The channel and tools share one plugin. Prefer the resolved channel credential so a secret + // entered through channel configuration works for both, while preserving tools-only env setup. + const apiKey = + (hostConfig ? resolveAgentMailAccount(hostConfig).apiKey : "") || + process.env.AGENTMAIL_API_KEY?.trim(); if (!apiKey) { throw new Error( - "AgentMail is not configured. Set the AGENTMAIL_API_KEY environment variable and restart OpenClaw.", + "AgentMail is not configured. Configure channels.agentmail.apiKey or set AGENTMAIL_API_KEY and restart OpenClaw.", ); } diff --git a/src/tools/index.test.ts b/src/tools/index.test.ts index da3b432..68ed4e6 100644 --- a/src/tools/index.test.ts +++ b/src/tools/index.test.ts @@ -45,7 +45,10 @@ type RegisteredTool = { ) => Promise; }; -function registerTools(pluginConfig: Record = {}): RegisteredTool[] { +function registerTools( + pluginConfig: Record = {}, + hostConfig: Record = {}, +): RegisteredTool[] { const registered: RegisteredTool[] = []; const register = entry.register as unknown as (api: { pluginConfig: Record; @@ -54,16 +57,21 @@ function registerTools(pluginConfig: Record = {}): RegisteredTo register({ pluginConfig, + config: hostConfig, registerTool(tool) { registered.push(tool); }, - }); + } as never); return registered; } -function findTool(name: string, config?: Record): RegisteredTool { - const found = registerTools(config).find((tool) => tool.name === name); +function findTool( + name: string, + config?: Record, + hostConfig?: Record, +): RegisteredTool { + const found = registerTools(config, hostConfig).find((tool) => tool.name === name); if (!found) { throw new Error(`Tool not registered: ${name}`); } @@ -117,6 +125,27 @@ describe("agentmail", () => { ); }); + it("uses the channel-configured API key when the environment is unset", async () => { + vi.stubEnv("AGENTMAIL_API_KEY", " "); + sdk.inboxes.list.mockResolvedValue({ count: 0, inboxes: [] }); + const tool = findTool( + "agentmail_list_inboxes", + {}, + { + channels: { + agentmail: { + apiKey: "am_channel", + inboxId: "Agent@AgentMail.TO", + }, + }, + }, + ); + + await tool.execute("call-channel-key", {}); + + expect(sdk.constructor).toHaveBeenCalledWith({ apiKey: "am_channel" }); + }); + it("sends a message with idempotency and cancellation options", async () => { sdk.messages.send.mockResolvedValue({ messageId: "msg_1", threadId: "thr_1" }); const signal = new AbortController().signal; @@ -170,7 +199,7 @@ describe("agentmail", () => { const tool = findTool("agentmail_list_inboxes"); await expect(tool.execute("call-4", {})).rejects.toThrow( - "Set the AGENTMAIL_API_KEY environment variable", + "Configure channels.agentmail.apiKey or set AGENTMAIL_API_KEY", ); expect(sdk.constructor).not.toHaveBeenCalled(); }); diff --git a/src/tools/index.ts b/src/tools/index.ts index 1a6c807..958ed1c 100644 --- a/src/tools/index.ts +++ b/src/tools/index.ts @@ -97,7 +97,7 @@ export default defineToolPlugin({ ), }, { additionalProperties: false }), execute: async (params, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.list(params, requestOptions(context.signal)); }, }), @@ -123,7 +123,7 @@ export default defineToolPlugin({ { additionalProperties: false }, ), execute: async (params, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.create(params, requestOptions(context.signal)); }, }), @@ -170,7 +170,7 @@ export default defineToolPlugin({ { additionalProperties: false }, ), execute: async ({ inboxId, before, after, ...params }, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.list( inboxId, { @@ -201,7 +201,7 @@ export default defineToolPlugin({ { additionalProperties: false }, ), execute: async ({ inboxId, query, before, after, ...params }, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.search( inboxId, { @@ -221,7 +221,7 @@ export default defineToolPlugin({ "Get one complete AgentMail message. Prefer extractedText or extractedHtml when processing a reply without quoted history.", parameters: Type.Object(messageLocationFields, { additionalProperties: false }), execute: async ({ inboxId, messageId }, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.get( inboxId, messageId, @@ -257,7 +257,7 @@ export default defineToolPlugin({ { additionalProperties: false }, ), execute: async ({ inboxId, idempotencyKey, ...message }, config, context) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.send(inboxId, message, { ...requestOptions(context.signal), ...(idempotencyKey ? { idempotencyKey } : {}), @@ -293,7 +293,7 @@ export default defineToolPlugin({ config, context, ) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.reply(inboxId, messageId, message, { ...requestOptions(context.signal), ...(idempotencyKey ? { idempotencyKey } : {}), @@ -331,7 +331,7 @@ export default defineToolPlugin({ config, context, ) => { - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.forward(inboxId, messageId, message, { ...requestOptions(context.signal), ...(idempotencyKey ? { idempotencyKey } : {}), @@ -364,7 +364,7 @@ export default defineToolPlugin({ throw new Error("Provide addLabels or removeLabels."); } - const client = createAgentMailClient(config); + const client = createAgentMailClient(config, context.api.config); return client.inboxes.messages.update( inboxId, messageId, From 961a67f912255521358d4eaaa744a52a8357a44d Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 04:39:41 +0000 Subject: [PATCH 3/4] Harden deferred ingress and route startup --- src/channel/src/catch-up.test.ts | 6 + src/channel/src/catch-up.ts | 4 + src/channel/src/durable-receive.test.ts | 181 ++++++++++++++++++++- src/channel/src/durable-receive.ts | 24 ++- src/channel/src/gateway.test.ts | 55 ++++++- src/channel/src/gateway.ts | 178 +++++++++++--------- src/channel/src/inbound.test.ts | 37 +++++ src/channel/src/inbound.ts | 60 +++++-- src/channel/src/ingress.ts | 205 ++++++++++++++++-------- src/tools/client.ts | 20 ++- src/tools/index.test.ts | 25 +++ 11 files changed, 625 insertions(+), 170 deletions(-) diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 35d715f..f5d458b 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -215,6 +215,12 @@ describe("AgentMail durable REST catch-up", () => { 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", diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index 84cb012..cfeb1e9 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -281,6 +281,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(runAtMs + 1), ascending: true, includeSpam: false, includeBlocked: false, diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index dc145d0..ea90c54 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -61,6 +61,35 @@ describe("AgentMail durable ingress", () => { expect(deletePending).toHaveBeenCalledWith(id); }); + it("preserves capacity backpressure when overflow deletion fails", async () => { + const id = createAgentMailDurableInboundId(record); + const fail = vi.fn(async () => true); + const journal = withAgentMailIngressCapacity( + { + accept: vi.fn(async () => ({ kind: "accepted", duplicate: false, record: {} })), + pending: vi.fn(async () => [ + { id: "other", payload: record, attempts: 0 }, + { id, payload: record, attempts: 0 }, + ]), + complete: vi.fn(), + release: vi.fn(), + deletePending: vi.fn(async () => { + throw new Error("queue delete unavailable"); + }), + fail, + } as never, + 1, + ); + + await expect( + processAgentMailIngress({ journal: journal as never, record, dispatch: vi.fn() }), + ).rejects.toBeInstanceOf(AgentMailIngressCapacityError); + expect(fail).toHaveBeenCalledWith(id, { + reason: "capacity-overflow", + message: "AgentMail rejected this ingress row because capacity was full", + }); + }); + it("accepts a completed duplicate at capacity without applying backpressure", async () => { const accept = vi.fn(async () => ({ kind: "completed", duplicate: true, record: {} })); const deletePending = vi.fn(); @@ -379,6 +408,123 @@ describe("AgentMail durable ingress", () => { expect(complete).toHaveBeenCalledOnce(); }); + it("does not redispatch when a queued deferred dispatch later rejects", async () => { + const complete = vi.fn(async () => undefined); + const release = vi.fn(async () => true); + let deferredLifecycle: + | { + onTurnDeferred: () => void; + onTurnAbandoned: () => Promise; + onTurnAdopted: () => Promise; + } + | undefined; + const dispatch = vi.fn( + async ( + _record: AgentMailIngressRecord, + lifecycle: NonNullable, + ) => { + if (dispatch.mock.calls.length === 1) { + deferredLifecycle = lifecycle; + lifecycle.onTurnDeferred(); + throw new Error("post-enqueue session write failed"); + } + await lifecycle.onTurnAdopted(); + }, + ); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledOnce()); + expect(release).not.toHaveBeenCalled(); + + await deferredLifecycle?.onTurnAbandoned(); + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(2)); + expect(release).toHaveBeenCalledOnce(); + expect(complete).toHaveBeenCalledOnce(); + }); + + it("retries a deferred adoption marker inside the lifecycle callback", async () => { + const complete = vi + .fn<() => Promise>() + .mockRejectedValueOnce(new Error("database busy")) + .mockResolvedValueOnce(undefined); + const release = vi.fn(async () => true); + let deferredLifecycle: + | { + onTurnDeferred: () => void; + onTurnAdopted: () => Promise; + } + | undefined; + const dispatch = vi.fn( + async ( + _record: AgentMailIngressRecord, + lifecycle: NonNullable, + ) => { + deferredLifecycle = lifecycle; + lifecycle.onTurnDeferred(); + }, + ); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + await vi.waitFor(() => expect(deferredLifecycle).toBeDefined()); + + await deferredLifecycle?.onTurnAdopted(); + await vi.waitFor(() => expect(complete).toHaveBeenCalledTimes(2)); + expect(dispatch).toHaveBeenCalledOnce(); + expect(release).not.toHaveBeenCalled(); + }); + + it("retries a turn abandoned before a deferred notification", async () => { + const complete = vi.fn(async () => undefined); + const release = vi.fn(async () => true); + const dispatch = vi.fn( + async ( + _record: AgentMailIngressRecord, + lifecycle: { + onTurnAbandoned: () => Promise; + onTurnAdopted: () => Promise; + }, + ) => { + if (dispatch.mock.calls.length === 1) { + await lifecycle.onTurnAbandoned(); + return; + } + await lifecycle.onTurnAdopted(); + }, + ); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + + await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(2)); + expect(release).toHaveBeenCalledWith(createAgentMailDurableInboundId(record), { + lastError: "deferred turn abandoned before adoption", + }); + expect(complete).toHaveBeenCalledOnce(); + }); + it("fails repeatedly abandoned deferred turns at the dispatch ceiling", async () => { const release = vi.fn(async () => true); const fail = vi.fn(async () => true); @@ -499,10 +645,43 @@ describe("AgentMail durable ingress", () => { retryDelayMs: () => 0, }); await vi.waitFor(() => expect(agentStarted).toHaveBeenCalledOnce()); - expect(dispatch).toHaveBeenCalledTimes(2); + expect(dispatch).toHaveBeenCalledOnce(); expect(complete).toHaveBeenCalledTimes(2); }); + it("gives completion markers a full retry budget after dispatch retries", async () => { + let dispatchFailures = 0; + const dispatch = vi.fn(async () => { + if (dispatchFailures < 49) { + dispatchFailures += 1; + throw new Error("temporary dispatch failure"); + } + }); + let completionFailures = 0; + const complete = vi.fn(async () => { + if (completionFailures < 49) { + completionFailures += 1; + throw new Error("temporary completion failure"); + } + }); + const fail = vi.fn(async () => true); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete, + release: vi.fn(async () => true), + fail, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + }); + + await vi.waitFor(() => expect(complete).toHaveBeenCalledTimes(50)); + expect(dispatch).toHaveBeenCalledTimes(50); + expect(fail).not.toHaveBeenCalled(); + }); + it("retries only the marker after an irrevocable active-turn adoption", async () => { const complete = vi .fn<() => Promise>() diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index 8be4e31..c23a5ee 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -70,8 +70,28 @@ export function withAgentMailIngressCapacity( const pending = await journal.pending(); pendingEstimate = pending.length; if (pendingEstimate > maxPendingEntries) { - await journal.deletePending(id); - pendingEstimate -= 1; + let removed = false; + try { + removed = await journal.deletePending(id); + } catch { + // Preserve the transport-facing capacity contract even if queue deletion is + // temporarily unavailable. A terminal marker is the best fallback for preventing + // this rejected admission from replaying as ordinary pending work. + } + if (!removed && journal.fail) { + try { + removed = await journal.fail(id, { + reason: "capacity-overflow", + message: "AgentMail rejected this ingress row because capacity was full", + }); + } catch { + // The next admission re-scans source-of-truth pending state. Keep returning the + // capacity error class so WebSocket/webhook recovery behavior remains correct. + } + } + if (removed) { + pendingEstimate -= 1; + } throw new AgentMailIngressCapacityError(); } } else { diff --git a/src/channel/src/gateway.test.ts b/src/channel/src/gateway.test.ts index 1ac5887..c0453ef 100644 --- a/src/channel/src/gateway.test.ts +++ b/src/channel/src/gateway.test.ts @@ -302,7 +302,7 @@ describe("AgentMail gateway route ownership", () => { }, ); - it("does not let an older startup failure release a newer route claim", async () => { + it("lets a newer startup proceed after an older concurrent startup fails", async () => { mocks.routes.length = 0; mocks.registerError = false; mocks.createCatchUpSession.mockReset(); @@ -324,7 +324,7 @@ describe("AgentMail gateway route ownership", () => { channelRuntime: {} as never, abortSignal: olderAbort.signal, }); - const olderResult = expect(older).rejects.toThrow("older startup failed"); + const olderResult = older.catch((error: unknown) => error); await vi.waitFor(() => expect(mocks.createCatchUpSession).toHaveBeenCalledTimes(1)); const newer = startAgentMailGatewayAccount({ cfg: {}, @@ -332,10 +332,15 @@ describe("AgentMail gateway route ownership", () => { channelRuntime: {} as never, abortSignal: newerAbort.signal, }); - await vi.waitFor(() => expect(mocks.routes).toHaveLength(1)); + // Per-account startup is serialized: the replacement cannot overtake the unresolved older + // initialization and leave both invocations parked. + expect(mocks.routes).toHaveLength(0); rejectOlder(new Error("older startup failed")); - await olderResult; + await expect(olderResult).resolves.toEqual( + expect.objectContaining({ message: "older startup failed" }), + ); + await vi.waitFor(() => expect(mocks.routes).toHaveLength(1)); await expect( startAgentMailGatewayAccount({ cfg: {}, @@ -349,6 +354,48 @@ describe("AgentMail gateway route ownership", () => { newerAbort.abort(); await newer; }); + + it("serializes different-path replacements without leaking the older route", async () => { + mocks.routes.length = 0; + mocks.registerError = false; + mocks.createCatchUpSession.mockReset(); + let resolveOlder!: (session: { run: typeof mocks.catchUpRun }) => void; + mocks.createCatchUpSession + .mockImplementationOnce( + async () => + await new Promise<{ run: typeof mocks.catchUpRun }>((resolve) => { + resolveOlder = resolve; + }), + ) + .mockResolvedValueOnce({ run: mocks.catchUpRun }); + + const olderAbort = new AbortController(); + const newerAbort = new AbortController(); + const older = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/older"), + channelRuntime: {} as never, + abortSignal: olderAbort.signal, + }); + await vi.waitFor(() => expect(mocks.createCatchUpSession).toHaveBeenCalledTimes(1)); + const newer = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/newer"), + channelRuntime: {} as never, + abortSignal: newerAbort.signal, + }); + expect(mocks.createCatchUpSession).toHaveBeenCalledTimes(1); + + resolveOlder({ run: mocks.catchUpRun }); + await vi.waitFor(() => expect(mocks.routes).toHaveLength(2)); + expect(mocks.routes[0]?.unregister).toHaveBeenCalledOnce(); + + olderAbort.abort(); + newerAbort.abort(); + await Promise.all([older, newer]); + expect(mocks.routes[0]?.unregister).toHaveBeenCalledOnce(); + expect(mocks.routes[1]?.unregister).toHaveBeenCalledOnce(); + }); }); describe("AgentMail security warnings", () => { diff --git a/src/channel/src/gateway.ts b/src/channel/src/gateway.ts index fe72ca6..29ddc8d 100644 --- a/src/channel/src/gateway.ts +++ b/src/channel/src/gateway.ts @@ -20,11 +20,36 @@ import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.j import { createAgentMailWebhookHandler, createAgentMailWebhookVerifier } from "./webhook.js"; import { startAgentMailWebSocket } from "./websocket.js"; -type RouteClaim = { accountId: string; generation: symbol; valid: boolean }; +type RouteClaim = { accountId: string; generation: symbol }; type ActiveRoute = { path: string; unregister: () => void; claim: RouteClaim }; const activeRoutes = new Map(); const routeOwners = new Map(); +const routeStartupTails = new Map>(); + +async function withSerializedRouteStartup( + accountId: string, + task: () => Promise, +): Promise { + const predecessor = routeStartupTails.get(accountId) ?? Promise.resolve(); + let release!: () => void; + const hold = new Promise((resolve) => { + release = resolve; + }); + const tail = predecessor.catch(() => undefined).then(() => hold); + routeStartupTails.set(accountId, tail); + await predecessor.catch(() => undefined); + try { + return await task(); + } finally { + release(); + void tail.finally(() => { + if (routeStartupTails.get(accountId) === tail) { + routeStartupTails.delete(accountId); + } + }); + } +} // Single source of truth for AgentMail sender-authorization warnings, shared by gateway startup // diagnostics and the channel security surface so the two never drift. @@ -153,90 +178,86 @@ export async function startAgentMailGatewayAccount(params: { const path = params.account.webhookPath.startsWith("/") ? params.account.webhookPath : `/${params.account.webhookPath}`; - const owner = routeOwners.get(path); - if (owner && owner.accountId !== params.account.accountId) { - throw new Error( - `AgentMail webhook path ${path} is already registered by account ${owner.accountId}; configure a distinct webhookPath.`, - ); - } - const previousRoute = activeRoutes.get(params.account.accountId); - const claim: RouteClaim = { - accountId: params.account.accountId, - generation: Symbol(params.account.accountId), - valid: true, - }; - // Claim ownership synchronously here, before the first await, so the check-and-claim is atomic: - // two concurrent startups for the same path can no longer both observe it as free. Keep the - // predecessor registered until the replacement is fully initialized and registered. - routeOwners.set(path, claim); - let catchUpSupervisor: ReturnType; - let unregister: (() => void) | undefined; - try { - const catchUpSession = await createAgentMailCatchUpSession({ - account: params.account, - client, - log: params.log, - }); - // A newer overlapping startup claimed this path while initialization awaited. Its generation - // owns registration; park this stale invocation until its lifecycle is cancelled. - if (routeOwners.get(path) !== claim) { - claim.valid = false; - return await waitUntilAbort(params.abortSignal); + const setup = await withSerializedRouteStartup(params.account.accountId, async () => { + if (params.abortSignal.aborted) { + return null; } - catchUpSupervisor = createAgentMailCatchUpSupervisor({ - session: catchUpSession, - receive, - abortSignal: params.abortSignal, - log: params.log, - }); - const receiveWithRecovery = async (record: AgentMailIngressRecord) => { - try { - await receive(record); - } catch (error) { - // Provider retries remain useful, but REST recovery is the durable fallback if the provider - // exhausts them while local admission is full or temporarily unavailable. - catchUpSupervisor.request(); - throw error; - } - }; - unregister = registerPluginHttpRoute({ - path, - auth: "plugin", - pluginId: "agentmail", + const owner = routeOwners.get(path); + if (owner && owner.accountId !== params.account.accountId) { + throw new Error( + `AgentMail webhook path ${path} is already registered by account ${owner.accountId}; configure a distinct webhookPath.`, + ); + } + const previousRoute = activeRoutes.get(params.account.accountId); + const claim: RouteClaim = { accountId: params.account.accountId, - handler: createAgentMailWebhookHandler({ + generation: Symbol(params.account.accountId), + }; + // Claim before initialization so a different account cannot race onto this path. Same-account + // startup is serialized above, so rollback always has one well-defined predecessor. + routeOwners.set(path, claim); + let unregister: (() => void) | undefined; + try { + const catchUpSession = await createAgentMailCatchUpSession({ account: params.account, - verifier, - receive: receiveWithRecovery, + client, log: params.log, - }), - // Replacing first is safe: OpenClaw removes the predecessor entry atomically, and its stale - // unregister handle becomes a no-op. This avoids a route outage if replacement setup fails. - replaceExisting: true, - }); - } catch (error) { - claim.valid = false; - unregister?.(); - // A stale startup must never release a newer startup's claim. Restore a still-valid predecessor - // on same-path rollback; otherwise release only this exact generation. - if (routeOwners.get(path) === claim) { - if (owner?.valid) { - routeOwners.set(path, owner); - } else { - routeOwners.delete(path); + }); + const catchUpSupervisor = createAgentMailCatchUpSupervisor({ + session: catchUpSession, + receive, + abortSignal: params.abortSignal, + log: params.log, + }); + const receiveWithRecovery = async (record: AgentMailIngressRecord) => { + try { + await receive(record); + } catch (error) { + // Provider retries remain useful, but REST recovery is the durable fallback if the + // provider exhausts them while local admission is temporarily unavailable. + catchUpSupervisor.request(); + throw error; + } + }; + unregister = registerPluginHttpRoute({ + path, + auth: "plugin", + pluginId: "agentmail", + accountId: params.account.accountId, + handler: createAgentMailWebhookHandler({ + account: params.account, + verifier, + receive: receiveWithRecovery, + log: params.log, + }), + // OpenClaw atomically replaces the predecessor and makes its stale unregister a no-op. + replaceExisting: true, + }); + const activeRoute = { path, unregister, claim }; + activeRoutes.set(params.account.accountId, activeRoute); + if (previousRoute) { + previousRoute.unregister(); + if (routeOwners.get(previousRoute.path) === previousRoute.claim) { + routeOwners.delete(previousRoute.path); + } } + return { activeRoute, catchUpSupervisor, claim, unregister }; + } catch (error) { + unregister?.(); + if (routeOwners.get(path) === claim) { + if (previousRoute?.path === path) { + routeOwners.set(path, previousRoute.claim); + } else { + routeOwners.delete(path); + } + } + throw error; } - throw error; - } - const activeRoute = { path, unregister, claim }; - activeRoutes.set(params.account.accountId, activeRoute); - if (previousRoute && previousRoute !== activeRoute) { - previousRoute.claim.valid = false; - previousRoute.unregister(); - if (routeOwners.get(previousRoute.path) === previousRoute.claim) { - routeOwners.delete(previousRoute.path); - } + }); + if (!setup) { + return; } + const { activeRoute, catchUpSupervisor, claim, unregister } = setup; params.log?.info?.( `Registered AgentMail webhook route ${path} for account ${params.account.accountId}`, ); @@ -251,7 +272,6 @@ export async function startAgentMailGatewayAccount(params: { await waitUntilAbort(params.abortSignal, () => { // A replaced account invocation can abort later; it must not delete the newer registration. if (activeRoutes.get(params.account.accountId) === activeRoute) { - claim.valid = false; unregister(); activeRoutes.delete(params.account.accountId); if (routeOwners.get(path) === claim) { diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts index 6225081..690a038 100644 --- a/src/channel/src/inbound.test.ts +++ b/src/channel/src/inbound.test.ts @@ -337,6 +337,43 @@ describe("AgentMail REST-authoritative inbound", () => { expect(onTurnAbandoned).toHaveBeenCalledOnce(); }); + it("cleans up deferred-turn attachments when account shutdown aborts the queued turn", async () => { + rm.mockClear(); + loadAgentMailInboundAttachments.mockResolvedValueOnce({ + paths: ["/tmp/deferred-abort.bin"], + types: ["application/octet-stream"], + }); + const controller = new AbortController(); + await dispatchAgentMailInboundEvent({ + cfg: {}, + account, + record, + channelRuntime: { + routing: { resolveAgentRoute: () => ({ agentId: "agent-1" }) }, + inbound: { + buildContext: (ctx: Record) => ctx, + run: async ({ + turnAdoptionLifecycle, + }: { + turnAdoptionLifecycle: { onDeferred: () => void }; + }) => { + turnAdoptionLifecycle.onDeferred(); + }, + }, + session: { resolveStorePath: () => "/tmp/s.json", recordInboundSession: vi.fn() }, + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + } as never, + client: { inboxes: { messages: { get: vi.fn(async () => message()) } } } as never, + abortSignal: controller.signal, + }); + expect(rm).not.toHaveBeenCalled(); + + controller.abort(); + await vi.waitFor(() => + expect(rm).toHaveBeenCalledWith("/tmp/deferred-abort.bin", { force: true }), + ); + }); + it("cleans up attachments when inbound handling resolves without adoption", async () => { rm.mockClear(); loadAgentMailInboundAttachments.mockResolvedValueOnce({ diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts index a4be9d4..839432e 100644 --- a/src/channel/src/inbound.ts +++ b/src/channel/src/inbound.ts @@ -256,24 +256,60 @@ export async function dispatchAgentMailInboundEvent(params: { let turnAdopted = false; let turnDeferred = false; + let turnAdoptionObserved = false; + let mediaCleanup: Promise | undefined; + const cleanupInboundMedia = async () => { + if (inboundMedia.paths.length === 0) { + return; + } + mediaCleanup ??= Promise.allSettled( + inboundMedia.paths.map((path) => rm(path, { force: true })), + ).then(() => undefined); + await mediaCleanup; + }; + let detachAbortCleanup = () => {}; + if (params.abortSignal) { + const onAbort = () => { + if (!turnAdoptionObserved) { + void cleanupInboundMedia(); + } + }; + params.abortSignal.addEventListener("abort", onAbort, { once: true }); + detachAbortCleanup = () => params.abortSignal?.removeEventListener("abort", onAbort); + if (params.abortSignal.aborted) { + onAbort(); + } + } // Adoption fires when core has made recovery-relevant session/run state durable. The ingress row // is completed here (via params.onTurnAdopted), closing the crash window before agent tools run; - // exclusive admission isolates the reply lane per turn, and the abort signal cancels a pre-adoption - // turn on shutdown. The local `turnAdopted` flag is set only AFTER the hook resolves so a failed - // journal.complete still triggers media cleanup below. + // exclusive admission isolates the reply lane per turn, and the abort signal cancels a + // pre-adoption turn on shutdown. const turnAdoptionLifecycle = { admission: "exclusive" as const, onAdopted: async () => { - await params.onTurnAdopted?.(); - turnAdopted = true; + turnAdoptionObserved = true; + try { + await params.onTurnAdopted?.(); + turnAdopted = true; + detachAbortCleanup(); + } catch (error) { + // Core has not started a fresh turn when its adoption observer rejects. Restore local media + // ownership so the normal failure/abort path can remove the files before a durable retry. + turnAdoptionObserved = false; + if (params.abortSignal?.aborted) { + await cleanupInboundMedia(); + } + throw error; + } }, onDeferred: () => { turnDeferred = true; params.onTurnDeferred?.(); }, onAbandoned: async () => { - if (!turnAdopted && inboundMedia.paths.length > 0) { - await Promise.allSettled(inboundMedia.paths.map((path) => rm(path, { force: true }))); + detachAbortCleanup(); + if (!turnAdoptionObserved) { + await cleanupInboundMedia(); } await params.onTurnAbandoned?.(); }, @@ -392,17 +428,19 @@ export async function dispatchAgentMailInboundEvent(params: { try { await runPromise; } catch (error) { - if (!turnAdopted && !turnDeferred && inboundMedia.paths.length > 0) { + if (!turnAdoptionObserved && !turnDeferred) { // The turn never adopted, so core did not take ownership of these freshly-saved attachment // files. Remove them so a durable retry (which re-downloads a clean set) does not leak one // copy per attempt. - await Promise.allSettled(inboundMedia.paths.map((path) => rm(path, { force: true }))); + detachAbortCleanup(); + await cleanupInboundMedia(); } throw error; } - if (!turnAdopted && !turnDeferred && inboundMedia.paths.length > 0) { + if (!turnAdopted && !turnDeferred) { // A normal return without adoption did not transfer ownership of freshly persisted files. // Deferred turns retain them until the lifecycle later adopts or abandons the queued turn. - await Promise.allSettled(inboundMedia.paths.map((path) => rm(path, { force: true }))); + detachAbortCleanup(); + await cleanupInboundMedia(); } } diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 116cbab..178ddcc 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -45,7 +45,7 @@ export type AgentMailIngressDispatch = ( const AGENTMAIL_MAX_DISPATCH_ATTEMPTS = 50; const AGENTMAIL_MAX_COMPLETION_ATTEMPTS = 50; -type DeferredOutcome = "adopted" | "abandoned" | "aborted"; +type DeferredOutcome = "adopted" | "abandoned" | "completion-failed" | "aborted"; async function waitForDeferredOutcome( outcome: Promise>, @@ -179,35 +179,90 @@ function scheduleAgentMailIngressDispatch(params: DispatchParams): void { } async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Promise { - let attempts = params.initialAttempts; + let dispatchAttempts = params.initialAttempts; + let completionAttempts = 0; while (!params.abortSignal?.aborted) { if (!params.dispatchCompleted) { let turnAdopted = false; let turnDeferred = false; + let turnAbandoned = false; + let turnAdoptionObserved = false; + let adoptionTask: Promise | undefined; let settleDeferred!: (outcome: Exclude) => void; const deferredOutcome = new Promise>((resolve) => { settleDeferred = resolve; }); const onTurnAdopted = async () => { - if (turnAdopted) { - return; + if (adoptionTask) { + return await adoptionTask; } - // Core persists restart-recovery delivery state before this callback. A fresh turn does - // not begin if it rejects; an already-committed active steer falls through to the - // marker-only retry below. Completing here closes the normal crash window before tools. - await params.journal.complete(params.id); - turnAdopted = true; - params.dispatchCompleted = true; - settleDeferred("adopted"); + // Core waits for this observer before starting a fresh turn. Retry the marker inside the + // observer so a transient store failure neither wedges a deferred callback nor causes the + // outer dispatcher to enqueue a second turn. + turnAdoptionObserved = true; + adoptionTask = (async () => { + while (!params.abortSignal?.aborted) { + try { + await params.journal.complete(params.id); + params.dispatchCompleted = true; + turnAdopted = true; + settleDeferred("adopted"); + return; + } catch (error) { + completionAttempts += 1; + if (completionAttempts >= AGENTMAIL_MAX_COMPLETION_ATTEMPTS) { + params.log?.error?.( + `AgentMail failed to persist adoption for message ${params.record.messageId} after ${completionAttempts} attempts`, + ); + try { + if ( + params.journal.fail && + (await params.journal.fail(params.id, { + reason: "completion-marker-failed", + message: "AgentMail could not persist the adoption completion marker", + })) + ) { + params.dispatchCompleted = true; + turnAdopted = true; + settleDeferred("adopted"); + return; + } + } catch { + // Leave the row pending for restart recovery when neither terminal marker can + // persist. + } + settleDeferred("completion-failed"); + throw error; + } + if ( + !(await waitForRetry( + params.abortSignal, + (params.retryDelay ?? retryDelayMs)(completionAttempts), + )) + ) { + settleDeferred("completion-failed"); + throw error; + } + } + } + settleDeferred("completion-failed"); + throw new Error("AgentMail adoption completion was aborted"); + })(); + return await adoptionTask; }; const onTurnDeferred = () => { turnDeferred = true; }; const onTurnAbandoned = async () => { - if (!turnAdopted) { + // Abandonment can precede the deferred notification on rejection paths. Record it + // independently so a dispatch return cannot accidentally complete and drop the row. + if (!turnAdoptionObserved) { + turnAbandoned = true; settleDeferred("abandoned"); } }; + let dispatchFailed = false; + let dispatchError: unknown; try { await params.dispatch(params.record, { onTurnAdopted, @@ -215,71 +270,80 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro onTurnAbandoned, abortSignal: params.abortSignal, }); - if (turnAdopted) { + } catch (error) { + dispatchFailed = true; + dispatchError = error; + } + + if (turnAdopted) { + return true; + } + if (turnDeferred || turnAdoptionObserved) { + // Once queued, a later dispatch rejection says nothing about ownership of that queued turn. + // Wait for core to adopt or abandon it instead of releasing the row and double-dispatching. + const outcome = await waitForDeferredOutcome(deferredOutcome, params.abortSignal); + if (outcome === "adopted") { return true; } - if (turnDeferred) { - const outcome = await waitForDeferredOutcome(deferredOutcome, params.abortSignal); - if (outcome === "adopted") { - return true; - } - if (outcome === "aborted") { - return false; - } - attempts += 1; - const lastError = "deferred turn abandoned before adoption"; - if (attempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { - params.log?.error?.( - `AgentMail dropping message ${params.record.messageId} after ${attempts} abandoned deferred turns`, - ); - try { - if (params.journal.fail) { - await params.journal.fail(params.id, { - reason: "dispatch-attempts-exhausted", - message: lastError, - }); - } else { - await params.journal.complete(params.id); - } - } catch { - // Best effort: TTL pruning still reclaims the row if the terminal marker cannot - // persist. + if (outcome === "aborted") { + return false; + } + if (outcome === "completion-failed") { + return false; + } + turnAbandoned = true; + } + + if (turnAbandoned) { + dispatchAttempts += 1; + const lastError = "deferred turn abandoned before adoption"; + if (dispatchAttempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { + params.log?.error?.( + `AgentMail dropping message ${params.record.messageId} after ${dispatchAttempts} abandoned deferred turns`, + ); + try { + if (params.journal.fail) { + await params.journal.fail(params.id, { + reason: "dispatch-attempts-exhausted", + message: lastError, + }); + } else { + await params.journal.complete(params.id); } - return true; - } - const released = await params.journal.release(params.id, { lastError }); - if (!released) { - return true; - } - if ( - !(await waitForRetry( - params.abortSignal, - (params.retryDelay ?? retryDelayMs)(attempts), - )) - ) { - return false; + } catch { + // Best effort: TTL pruning still reclaims the row if the terminal marker cannot + // persist. } - continue; + return true; } - params.dispatchCompleted = true; - } catch (error) { - if (turnAdopted) { - // The adopted turn is now owned by core's restart-recovery machinery. Releasing the - // ingress row would replay agent tools even if the later turn failed. + const released = await params.journal.release(params.id, { lastError }); + if (!released) { return true; } - attempts += 1; - if (attempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { + if ( + !(await waitForRetry( + params.abortSignal, + (params.retryDelay ?? retryDelayMs)(dispatchAttempts), + )) + ) { + return false; + } + continue; + } + + if (dispatchFailed) { + dispatchAttempts += 1; + if (dispatchAttempts >= AGENTMAIL_MAX_DISPATCH_ATTEMPTS) { // Poison message: it has failed deterministically past the retry ceiling. Mark it // terminal so it stops occupying an admission slot and blocking new mail. params.log?.error?.( - `AgentMail dropping message ${params.record.messageId} after ${attempts} failed dispatch attempts: ${errorText(error)}`, + `AgentMail dropping message ${params.record.messageId} after ${dispatchAttempts} failed dispatch attempts: ${errorText(dispatchError)}`, ); try { if (params.journal.fail) { await params.journal.fail(params.id, { reason: "dispatch-attempts-exhausted", - message: errorText(error), + message: errorText(dispatchError), }); } else { await params.journal.complete(params.id); @@ -289,7 +353,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } return true; } - const lastError = errorText(error); + const lastError = errorText(dispatchError); while (!params.abortSignal?.aborted) { try { const released = await params.journal.release(params.id, { lastError }); @@ -303,7 +367,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro if ( !(await waitForRetry( params.abortSignal, - (params.retryDelay ?? retryDelayMs)(attempts), + (params.retryDelay ?? retryDelayMs)(dispatchAttempts), )) ) { return false; @@ -316,9 +380,9 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro const shouldRetry = await waitForRetry( params.abortSignal, nextDispatchDelayMs({ - error, + error: dispatchError, record: params.record, - attempts, + attempts: dispatchAttempts, retryDelay: params.retryDelay ?? retryDelayMs, }), ); @@ -327,15 +391,16 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } continue; } + params.dispatchCompleted = true; } try { await params.journal.complete(params.id); return true; } catch { - attempts += 1; - if (attempts >= AGENTMAIL_MAX_COMPLETION_ATTEMPTS) { + completionAttempts += 1; + if (completionAttempts >= AGENTMAIL_MAX_COMPLETION_ATTEMPTS) { params.log?.error?.( - `AgentMail failed to persist completion for message ${params.record.messageId} after ${attempts} attempts; marking the ingress row failed`, + `AgentMail failed to persist completion for message ${params.record.messageId} after ${completionAttempts} attempts; marking the ingress row failed`, ); if (params.journal.fail) { return await params.journal.fail(params.id, { @@ -349,7 +414,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro // idempotent completion marker; releasing and redispatching would duplicate the reply. const shouldRetry = await waitForRetry( params.abortSignal, - (params.retryDelay ?? retryDelayMs)(attempts), + (params.retryDelay ?? retryDelayMs)(completionAttempts), ); if (!shouldRetry) { return false; diff --git a/src/tools/client.ts b/src/tools/client.ts index 9d0e079..d3851bc 100644 --- a/src/tools/client.ts +++ b/src/tools/client.ts @@ -37,11 +37,25 @@ export function createAgentMailClient( ): AgentMailClient { // The channel and tools share one plugin. Prefer the resolved channel credential so a secret // entered through channel configuration works for both, while preserving tools-only env setup. - const apiKey = - (hostConfig ? resolveAgentMailAccount(hostConfig).apiKey : "") || - process.env.AGENTMAIL_API_KEY?.trim(); + const envApiKey = process.env.AGENTMAIL_API_KEY?.trim(); + let channelApiKey = ""; + let channelResolutionError: unknown; + if (hostConfig) { + try { + channelApiKey = resolveAgentMailAccount(hostConfig).apiKey; + } catch (error) { + // Tool execution can receive the unresolved persisted config outside a gateway secret + // snapshot. Keep the environment fallback reachable; if it is absent, preserve the precise + // unresolved-secret diagnostic instead of replacing it with a generic configuration error. + channelResolutionError = error; + } + } + const apiKey = channelApiKey || envApiKey; if (!apiKey) { + if (channelResolutionError) { + throw channelResolutionError; + } throw new Error( "AgentMail is not configured. Configure channels.agentmail.apiKey or set AGENTMAIL_API_KEY and restart OpenClaw.", ); diff --git a/src/tools/index.test.ts b/src/tools/index.test.ts index 68ed4e6..a77295c 100644 --- a/src/tools/index.test.ts +++ b/src/tools/index.test.ts @@ -146,6 +146,31 @@ describe("agentmail", () => { expect(sdk.constructor).toHaveBeenCalledWith({ apiKey: "am_channel" }); }); + it("falls back to the environment when channel secret resolution is unavailable", async () => { + vi.stubEnv("AGENTMAIL_API_KEY", "am_env_fallback"); + sdk.inboxes.list.mockResolvedValue({ count: 0, inboxes: [] }); + const tool = findTool( + "agentmail_list_inboxes", + {}, + { + channels: { + agentmail: { + apiKey: { + source: "env", + provider: "agentmail", + id: "UNRESOLVED_AGENTMAIL_API_KEY", + }, + inboxId: "agent@agentmail.to", + }, + }, + }, + ); + + await tool.execute("call-secret-fallback", {}); + + expect(sdk.constructor).toHaveBeenCalledWith({ apiKey: "am_env_fallback" }); + }); + it("sends a message with idempotency and cancellation options", async () => { sdk.messages.send.mockResolvedValue({ messageId: "msg_1", threadId: "thr_1" }); const signal = new AbortController().signal; From aace84b1a421263139ac619b789a9791b2f1df42 Mon Sep 17 00:00:00 2001 From: Haakam Aujla Date: Thu, 23 Jul 2026 05:13:18 +0000 Subject: [PATCH 4/4] Keep capacity overflow retryable --- src/channel/src/catch-up.test.ts | 33 ++++++++++++++++++++++++ src/channel/src/catch-up.ts | 15 +++++++---- src/channel/src/durable-receive.test.ts | 27 +++++++++++++++----- src/channel/src/durable-receive.ts | 25 +++++------------- src/channel/src/websocket.test.ts | 34 +++++++++++++++++++++++++ src/channel/src/websocket.ts | 11 ++++---- 6 files changed, 109 insertions(+), 36 deletions(-) diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index f5d458b..73ef9a3 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -196,6 +196,39 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("does not advance the cursor beyond the sweep's fixed upper bound", async () => { + const store = memoryStore(); + 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(); const nowMs = 1_000_000; diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index cfeb1e9..4a78fc1 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -254,6 +254,7 @@ export async function createAgentMailCatchUpSession(params: { 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 @@ -284,7 +285,7 @@ export async function createAgentMailCatchUpSession(params: { // 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(runAtMs + 1), + before: new Date(scanUpperBoundAtMs + 1), ascending: true, includeSpam: false, includeBlocked: false, @@ -304,7 +305,8 @@ export async function createAgentMailCatchUpSession(params: { // 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) ?? now(); + resolveReceivedAgentMailMessageTimestampMs(message, params.account.inboxId) ?? + scanUpperBoundAtMs; try { await receive({ accountId: params.account.accountId, @@ -327,7 +329,10 @@ export async function createAgentMailCatchUpSession(params: { throw error; } admitted += 1; - highWaterAtMs = Math.max(highWaterAtMs, Math.min(messageTimestampMs, now())); + highWaterAtMs = Math.max( + highWaterAtMs, + Math.min(messageTimestampMs, scanUpperBoundAtMs), + ); pageAdvanced = true; } if (pageAdvanced) { @@ -338,7 +343,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, - upperBoundAtMs: now(), + upperBoundAtMs: scanUpperBoundAtMs, established: false, }); } @@ -353,7 +358,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, - upperBoundAtMs: now(), + upperBoundAtMs: scanUpperBoundAtMs, established: true, }); if (admitted > 0) { diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index ea90c54..2452ff7 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -61,17 +61,26 @@ describe("AgentMail durable ingress", () => { expect(deletePending).toHaveBeenCalledWith(id); }); - it("preserves capacity backpressure when overflow deletion fails", async () => { + it("keeps an overflow row retryable when rollback deletion fails", async () => { const id = createAgentMailDurableInboundId(record); const fail = vi.fn(async () => true); + const accept = vi + .fn() + .mockResolvedValueOnce({ kind: "accepted", duplicate: false, record: { id } }) + .mockResolvedValueOnce({ + kind: "pending", + duplicate: true, + record: { id, payload: record, attempts: 0 }, + }); + const complete = vi.fn(async () => undefined); const journal = withAgentMailIngressCapacity( { - accept: vi.fn(async () => ({ kind: "accepted", duplicate: false, record: {} })), + accept, pending: vi.fn(async () => [ { id: "other", payload: record, attempts: 0 }, { id, payload: record, attempts: 0 }, ]), - complete: vi.fn(), + complete, release: vi.fn(), deletePending: vi.fn(async () => { throw new Error("queue delete unavailable"); @@ -84,10 +93,14 @@ describe("AgentMail durable ingress", () => { await expect( processAgentMailIngress({ journal: journal as never, record, dispatch: vi.fn() }), ).rejects.toBeInstanceOf(AgentMailIngressCapacityError); - expect(fail).toHaveBeenCalledWith(id, { - reason: "capacity-overflow", - message: "AgentMail rejected this ingress row because capacity was full", - }); + + const dispatch = vi.fn(async () => undefined); + await expect( + processAgentMailIngress({ journal: journal as never, record, dispatch }), + ).resolves.toBe("accepted"); + await vi.waitFor(() => expect(complete).toHaveBeenCalledWith(id)); + expect(dispatch).toHaveBeenCalledOnce(); + expect(fail).not.toHaveBeenCalled(); }); it("accepts a completed duplicate at capacity without applying backpressure", async () => { diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index c23a5ee..b5539c9 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -70,27 +70,14 @@ export function withAgentMailIngressCapacity( const pending = await journal.pending(); pendingEstimate = pending.length; if (pendingEstimate > maxPendingEntries) { - let removed = false; try { - removed = await journal.deletePending(id); - } catch { - // Preserve the transport-facing capacity contract even if queue deletion is - // temporarily unavailable. A terminal marker is the best fallback for preventing - // this rejected admission from replaying as ordinary pending work. - } - if (!removed && journal.fail) { - try { - removed = await journal.fail(id, { - reason: "capacity-overflow", - message: "AgentMail rejected this ingress row because capacity was full", - }); - } catch { - // The next admission re-scans source-of-truth pending state. Keep returning the - // capacity error class so WebSocket/webhook recovery behavior remains correct. + if (await journal.deletePending(id)) { + pendingEstimate -= 1; } - } - if (removed) { - 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. } throw new AgentMailIngressCapacityError(); } diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts index 6d9ef30..f541179 100644 --- a/src/channel/src/websocket.test.ts +++ b/src/channel/src/websocket.test.ts @@ -112,6 +112,40 @@ describe("AgentMail WebSocket ingress", () => { expect(waitForOpen).not.toHaveBeenCalled(); }); + it("durably admits message.received frames before the received label projects", async () => { + handlers.clear(); + const receive = vi.fn(async () => undefined); + const controller = new AbortController(); + const running = startAgentMailWebSocket({ + account, + abortSignal: controller.signal, + receive, + catchUpSession: { run: catchUpRun }, + }); + await vi.waitFor(() => expect(handlers.has("message")).toBe(true)); + handlers.get("message")?.({ + type: "event", + eventType: "message.received", + message: { + inboxId: "inbox_1", + messageId: "message_label_pending", + labels: [], + timestamp: new Date(1_234), + }, + }); + + await vi.waitFor(() => expect(receive).toHaveBeenCalledOnce()); + expect(receive).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "message_label_pending", + transport: "websocket", + receivedAt: 1_234, + }), + ); + controller.abort(); + await running; + }); + it("retries until a WebSocket event is durably admitted", async () => { handlers.clear(); const receive = vi diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts index b5f2b18..44ae86e 100644 --- a/src/channel/src/websocket.ts +++ b/src/channel/src/websocket.ts @@ -12,7 +12,7 @@ import { AgentMailIngressCapacityError } from "./ingress.js"; import { createBackoff, waitForRetry } from "./retry.js"; import { agentMailInboxIdsEqual, - resolveReceivedAgentMailMessageTimestampMs, + resolveAgentMailTimestampMs, } from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; @@ -169,11 +169,12 @@ export async function startAgentMailWebSocket(params: { params.log?.warn?.("AgentMail WebSocket ignored an event for the wrong inbox"); return; } - const messageTimestampMs = resolveReceivedAgentMailMessageTimestampMs( - event.message, - params.account.inboxId, - ); + // 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 messageTimestampMs = resolveAgentMailTimestampMs(event.message.timestamp); if (messageTimestampMs === null) { + params.log?.warn?.("AgentMail WebSocket received an event with an invalid timestamp"); + catchUpSupervisor.request(); return; } if (queuedMessageIds.has(event.message.messageId)) {