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/package-lock.json b/package-lock.json index 749cb29..59ce3d3 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,11 +1,11 @@ { - "name": "@openclaw/agentmail", + "name": "@agentmail/agentmail", "version": "0.1.0", "lockfileVersion": 3, "requires": true, "packages": { "": { - "name": "@openclaw/agentmail", + "name": "@agentmail/agentmail", "version": "0.1.0", "license": "MIT", "dependencies": { @@ -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 0c9013b..2147a06 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/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..d66705f 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..a473c91 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,10 @@ export function resolveAgentMailAccount( accountId: id, enabled: channel.enabled !== false && account?.enabled !== false, apiKey: apiVal, - inboxId: merged.inboxId?.trim() ?? "", + // Preserve configured spelling as the durable identity. Queue namespaces, catch-up cursor keys, + // and conversation ids include this value, so lowercasing it during resolution would strand + // persisted state after an upgrade. Normalize provider values only where ids are compared. + inboxId: (merged.inboxId ?? "").trim(), webhookSecret: hookVal, webhookPath: configuredPath || @@ -187,7 +191,7 @@ export function findConflictingAgentMailInboxOwner( if ( other.enabled && isAgentMailAccountConfigured(other) && - other.inboxId === account.inboxId + normalizeAgentMailInboxId(other.inboxId) === normalizeAgentMailInboxId(account.inboxId) ) { return other.accountId; } diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 3972bff..73ef9a3 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,97 @@ describe("AgentMail durable REST catch-up", () => { ); }); + 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); + const list = vi.fn(async () => ({ count: 1, messages: [malformed] })); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store: store as never, + now: () => 1_000, + }); + const receive = vi.fn(async () => undefined); + + await session.run({ receive, abortSignal: new AbortController().signal }); + expect(receive).toHaveBeenCalledWith( + expect.objectContaining({ + messageId: "bad_timestamp", + receivedAt: 1_000, + arrivedAt: 1_000, + }), + ); + }); + + it("does not advance the cursor beyond the sweep's fixed upper bound", async () => { + const store = memoryStore(); + 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; + const list = vi + .fn() + .mockResolvedValueOnce({ + count: 1, + messages: [message({ id: "future", timestamp: nowMs + 60 * 60_000 })], + }) + .mockResolvedValueOnce({ count: 0, messages: [] }); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store: store as never, + now: () => nowMs, + }); + const receive = vi.fn(async () => undefined); + + await session.run({ receive, abortSignal: new AbortController().signal }); + expect(list).toHaveBeenNthCalledWith( + 1, + "inbox_1", + expect.objectContaining({ before: new Date(nowMs + 1) }), + expect.any(Object), + ); + await session.run({ receive, abortSignal: new AbortController().signal }); + expect(list).toHaveBeenLastCalledWith( + "inbox_1", + expect.objectContaining({ + after: new Date(nowMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), + }), + expect.any(Object), + ); + }); + it("lists from the monitoring baseline on a deep sweep", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index 85ccea3..4a78fc1 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -7,7 +7,11 @@ import { AGENTMAIL_DURABLE_COMPLETED_TTL_MS, AgentMailIngressCapacityError, } from "./durable-receive.js"; -import { AGENTMAIL_RECEIVED_LABEL } from "./inbound.js"; +import { + AGENTMAIL_RECEIVED_LABEL, + isReceivedAgentMailMessage, + resolveReceivedAgentMailMessageTimestampMs, +} from "./received-message.js"; import { createBackoff, waitForRetry } from "./retry.js"; import { getAgentMailRuntime } from "./runtime.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; @@ -175,20 +179,12 @@ function normalizeCursor(value: unknown): AgentMailCatchUpCursor | null { }; } -function isReceivedMessage(message: AgentMail.MessageItem, inboxId: string): boolean { - return ( - message.inboxId === inboxId && - (Array.isArray(message.labels) ? message.labels : []).some( - (label) => String(label).toLocaleLowerCase("en-US") === AGENTMAIL_RECEIVED_LABEL, - ) - ); -} - async function persistCursor(params: { store: PluginStateKeyedStore; key: string; baselineAtMs: number; highWaterAtMs: number; + upperBoundAtMs: number; established: boolean; }): Promise { const update = params.store.update; @@ -198,7 +194,13 @@ async function persistCursor(params: { return { version: CURSOR_VERSION, baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), + highWaterAtMs: Math.max( + current?.baselineAtMs ?? params.baselineAtMs, + Math.min( + params.upperBoundAtMs, + Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), + ), + ), established: current?.established === true || params.established, }; }); @@ -208,7 +210,13 @@ async function persistCursor(params: { await params.store.register(params.key, { version: CURSOR_VERSION, baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), + highWaterAtMs: Math.max( + current?.baselineAtMs ?? params.baselineAtMs, + Math.min( + params.upperBoundAtMs, + Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), + ), + ), established: current?.established === true || params.established, }); } @@ -245,6 +253,9 @@ export async function createAgentMailCatchUpSession(params: { if (!storedCursor) { throw new Error("AgentMail WebSocket catch-up cursor is unavailable"); } + const runAtMs = now(); + const scanUpperBoundAtMs = runAtMs; + const effectiveHighWaterAtMs = Math.min(storedCursor.highWaterAtMs, runAtMs); // Never scan below the dedupe horizon on ANY sweep. Completed-message tombstones expire after // AGENTMAIL_DURABLE_COMPLETED_TTL_MS; once they do, a message still in scan range would be // re-admitted as a duplicate turn. The floor is applied to the normal high-water sweep too: @@ -254,13 +265,13 @@ export async function createAgentMailCatchUpSession(params: { // The deep sweep lists from max(baseline, dedupeFloor): it recovers back-dated mail within the // dedupe window and since monitoring began, but deliberately does not scan before the baseline, // which would re-inject pre-monitoring history that has no tombstone protection. - const dedupeFloorMs = now() - AGENTMAIL_DURABLE_COMPLETED_TTL_MS; + const dedupeFloorMs = runAtMs - AGENTMAIL_DURABLE_COMPLETED_TTL_MS; const baseAfterMs = sinceBaseline || !storedCursor.established ? storedCursor.baselineAtMs - : storedCursor.highWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS; + : effectiveHighWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS; const afterMs = Math.max(0, baseAfterMs, dedupeFloorMs); - let highWaterAtMs = storedCursor.highWaterAtMs; + let highWaterAtMs = effectiveHighWaterAtMs; let pageCursor: string | undefined; let admitted = 0; do { @@ -271,6 +282,10 @@ export async function createAgentMailCatchUpSession(params: { ...(pageCursor ? { pageToken: pageCursor } : {}), labels: [AGENTMAIL_RECEIVED_LABEL], after: new Date(afterMs), + // Do not repeatedly scan provider-clock-skewed future messages. Add one millisecond so + // an exclusive provider bound includes messages stamped exactly at runAtMs and so a + // fresh cursor never sends identical after/before values. + before: new Date(scanUpperBoundAtMs + 1), ascending: true, includeSpam: false, includeBlocked: false, @@ -284,16 +299,21 @@ export async function createAgentMailCatchUpSession(params: { if (abortSignal.aborted) { return; } - if (!isReceivedMessage(message, params.account.inboxId)) { + if (!isReceivedAgentMailMessage(message, params.account.inboxId)) { continue; } + // Invalid provider time must not turn a missed live event into a permanent drop. Admit + // with local observation time; durable message-id dedupe keeps overlap scans safe. + const messageTimestampMs = + resolveReceivedAgentMailMessageTimestampMs(message, params.account.inboxId) ?? + scanUpperBoundAtMs; try { await receive({ accountId: params.account.accountId, inboxId: params.account.inboxId, messageId: message.messageId, transport: "rest", - receivedAt: message.timestamp.getTime(), + receivedAt: messageTimestampMs, arrivedAt: now(), }); } catch (error) { @@ -309,7 +329,10 @@ export async function createAgentMailCatchUpSession(params: { throw error; } admitted += 1; - highWaterAtMs = Math.max(highWaterAtMs, message.timestamp.getTime()); + highWaterAtMs = Math.max( + highWaterAtMs, + Math.min(messageTimestampMs, scanUpperBoundAtMs), + ); pageAdvanced = true; } if (pageAdvanced) { @@ -320,6 +343,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, + upperBoundAtMs: scanUpperBoundAtMs, established: false, }); } @@ -334,6 +358,7 @@ export async function createAgentMailCatchUpSession(params: { key, baselineAtMs: storedCursor.baselineAtMs, highWaterAtMs, + upperBoundAtMs: scanUpperBoundAtMs, established: true, }); if (admitted > 0) { diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index 01a5081..2452ff7 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,69 @@ describe("AgentMail durable ingress", () => { dispatch: vi.fn(), }), ).rejects.toBeInstanceOf(AgentMailIngressCapacityError); - expect(accept).not.toHaveBeenCalled(); + expect(accept).toHaveBeenCalledOnce(); + expect(deletePending).toHaveBeenCalledWith(id); + }); + + 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, + pending: vi.fn(async () => [ + { id: "other", payload: record, attempts: 0 }, + { id, payload: record, attempts: 0 }, + ]), + complete, + 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); + + 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 () => { + 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 +381,225 @@ 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("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); + 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 () => { + 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); @@ -371,10 +658,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 48dfb51..b5539c9 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,28 @@ 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) { + try { + if (await journal.deletePending(id)) { + 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(); } - } - const accepted = await journal.accept(id, payload, options); - if (accepted.kind === "accepted") { + } else { pendingEstimate += 1; } return accepted; @@ -88,14 +106,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..c0453ef 100644 --- a/src/channel/src/gateway.test.ts +++ b/src/channel/src/gateway.test.ts @@ -6,12 +6,18 @@ 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), 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"; @@ -32,9 +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; }, })); @@ -44,7 +59,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, @@ -196,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(); @@ -213,6 +229,173 @@ 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; + }, + ); + + 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("lets a newer startup proceed after an older concurrent startup fails", 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 = older.catch((error: unknown) => error); + await vi.waitFor(() => expect(mocks.createCatchUpSession).toHaveBeenCalledTimes(1)); + const newer = startAgentMailGatewayAccount({ + cfg: {}, + account: account("support", "/webhooks/agentmail/race"), + channelRuntime: {} as never, + abortSignal: newerAbort.signal, + }); + // 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 expect(olderResult).resolves.toEqual( + expect.objectContaining({ message: "older startup failed" }), + ); + await vi.waitFor(() => expect(mocks.routes).toHaveLength(1)); + 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; + }); + + 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 1f42dad..29ddc8d 100644 --- a/src/channel/src/gateway.ts +++ b/src/channel/src/gateway.ts @@ -20,10 +20,36 @@ 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 }; +type ActiveRoute = { path: string; unregister: () => void; claim: RouteClaim }; const activeRoutes = new Map(); -const routeOwners = 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. @@ -89,7 +115,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 +130,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) => { @@ -145,58 +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 !== params.account.accountId) { - throw new Error( - `AgentMail webhook path ${path} is already registered by account ${owner}; 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 setup = await withSerializedRouteStartup(params.account.accountId, async () => { + if (params.abortSignal.aborted) { + return null; } - } - // 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) => { + 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), + }; + // 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 { - await receive(record); + 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 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) { - // 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(); + unregister?.(); + if (routeOwners.get(path) === claim) { + if (previousRoute?.path === path) { + routeOwners.set(path, previousRoute.claim); + } else { + routeOwners.delete(path); + } + } throw error; } - }; - const unregister = registerPluginHttpRoute({ - path, - auth: "plugin", - pluginId: "agentmail", - accountId: params.account.accountId, - handler: createAgentMailWebhookHandler({ - account: params.account, - verifier, - receive: receiveWithRecovery, - log: params.log, - }), }); - const activeRoute = { path, unregister }; - activeRoutes.set(params.account.accountId, activeRoute); - // Ownership was already claimed synchronously above. + if (!setup) { + return; + } + const { activeRoute, catchUpSupervisor, claim, unregister } = setup; params.log?.info?.( `Registered AgentMail webhook route ${path} for account ${params.account.accountId}`, ); @@ -213,7 +274,7 @@ export async function startAgentMailGatewayAccount(params: { if (activeRoutes.get(params.account.accountId) === activeRoute) { 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/inbound.test.ts b/src/channel/src/inbound.test.ts index 5654bba..690a038 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,106 @@ 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 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({ + 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..839432e 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,16 +255,63 @@ 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 () => { + detachAbortCleanup(); + if (!turnAdoptionObserved) { + await cleanupInboundMedia(); + } + await params.onTurnAbandoned?.(); }, ...(params.abortSignal ? { abortSignal: params.abortSignal } : {}), }; @@ -273,9 +326,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 +352,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 +389,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,12 +428,19 @@ export async function dispatchAgentMailInboundEvent(params: { try { await runPromise; } catch (error) { - if (!turnAdopted && 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) { + // 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. + detachAbortCleanup(); + await cleanupInboundMedia(); + } } diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 8d374ed..178ddcc 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" | "completion-failed" | "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; @@ -146,49 +179,181 @@ 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 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 () => { + // 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"); } - // 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; }; + let dispatchFailed = false; + let dispatchError: unknown; try { - await params.dispatch(params.record, { onTurnAdopted, abortSignal: params.abortSignal }); - if (turnAdopted) { + await params.dispatch(params.record, { + onTurnAdopted, + onTurnDeferred, + onTurnAbandoned, + abortSignal: params.abortSignal, + }); + } 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; } - 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. + 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); + } + } 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; } - 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. + 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 { - await params.journal.complete(params.id); + if (params.journal.fail) { + await params.journal.fail(params.id, { + reason: "dispatch-attempts-exhausted", + message: errorText(dispatchError), + }); + } 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 lastError = errorText(error); + const lastError = errorText(dispatchError); while (!params.abortSignal?.aborted) { try { const released = await params.journal.release(params.id, { lastError }); @@ -202,7 +367,7 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro if ( !(await waitForRetry( params.abortSignal, - (params.retryDelay ?? retryDelayMs)(attempts), + (params.retryDelay ?? retryDelayMs)(dispatchAttempts), )) ) { return false; @@ -215,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, }), ); @@ -226,17 +391,30 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } continue; } + params.dispatchCompleted = true; } try { await params.journal.complete(params.id); return true; } catch { - attempts += 1; + completionAttempts += 1; + if (completionAttempts >= AGENTMAIL_MAX_COMPLETION_ATTEMPTS) { + params.log?.error?.( + `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, { + 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( params.abortSignal, - (params.retryDelay ?? retryDelayMs)(attempts), + (params.retryDelay ?? retryDelayMs)(completionAttempts), ); if (!shouldRetry) { return false; diff --git a/src/channel/src/media.test.ts b/src/channel/src/media.test.ts index 691e5aa..a68595e 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,39 @@ describe("AgentMail inbound attachments", () => { messageId: "message_1", attachments: [ { attachmentId: "inline", size: 1, contentDisposition: "inline" }, - { attachmentId: "cid", size: 1, contentId: "image@cid" }, + // 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, + 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..df86ae5 100644 --- a/src/channel/src/media.ts +++ b/src/channel/src/media.ts @@ -18,7 +18,31 @@ export type AgentMailInboundMedia = { export class AgentMailMediaPolicyError extends Error {} function isAcceptedAttachment(attachment: AgentMail.Attachment): boolean { - return attachment.contentDisposition !== "inline" && !attachment.contentId; + 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: { + 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 +55,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 +100,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 +172,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..c69314f --- /dev/null +++ b/src/channel/src/received-message.ts @@ -0,0 +1,44 @@ +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 isReceivedAgentMailMessage( + message: AgentMail.MessageItem | AgentMail.Message, + inboxId: string, +): boolean { + if (!agentMailInboxIdsEqual(message.inboxId, inboxId)) { + return false; + } + const labels = Array.isArray(message.labels) ? message.labels : []; + 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 071c5c2..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 = () => @@ -452,7 +514,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 +546,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..ad9d426 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."); @@ -249,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", @@ -268,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; @@ -306,13 +322,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..f541179 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), }, }); @@ -111,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 @@ -132,6 +167,7 @@ describe("AgentMail WebSocket ingress", () => { message: { inboxId: "inbox_1", messageId: "message_retry", + labels: ["received"], timestamp: new Date(1_234), }, }); @@ -162,13 +198,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 +244,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)); @@ -228,9 +284,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({ @@ -238,6 +296,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)); @@ -245,6 +304,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 890f127..44ae86e 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, @@ -9,6 +10,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, + resolveAgentMailTimestampMs, +} from "./received-message.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const AGENTMAIL_WEBSOCKET_LIVE_QUEUE_MAX = 32; @@ -25,16 +30,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 +165,18 @@ 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; } + // 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)) { return; } @@ -180,10 +190,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(); @@ -196,13 +206,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) { @@ -246,14 +250,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)}`, @@ -261,6 +271,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/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; diff --git a/src/tools/client.ts b/src/tools/client.ts index 53c08fe..d3851bc 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,33 @@ 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 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. 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..a77295c 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,52 @@ 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("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; @@ -170,7 +224,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,