diff --git a/.gitignore b/.gitignore index e8e9ed6..50b7882 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ node_modules/ dist/ +vendor/ coverage/ *.tgz .env diff --git a/README.md b/README.md index 4b542c7..3cd41f2 100644 --- a/README.md +++ b/README.md @@ -2,15 +2,20 @@ Give an OpenClaw agent an email address with [AgentMail](https://www.agentmail.to/). This package ships **two capabilities in one plugin**: -- **Email tools** — the agent can create inboxes and read, search, send, reply, forward, and label email on demand. +- **A CLI-backed AgentMail skill** — the agent uses the official AgentMail CLI bundled with the + plugin. It can discover and use new AgentMail API resources without waiting for this plugin to + add another fixed tool schema. - **An email channel** — a durable, allowlisted, **reply-only** email channel. Inbound email drives agent turns; the agent replies within the AgentMail thread. Ingress is committed durably before acknowledgement, senders are authorized against a default-deny allowlist, and replies stay bound to the triggering message (`replyAll: false`, no proactive threads, no arbitrary recipients). ## Requirements -- Node.js 22.22.3+, 24.15+, or 25.9+ +- Node.js 22.22.3–22.x, 24.15.0–24.x, or 25.9.0+ - OpenClaw 2026.7.2 (beta) or newer - An AgentMail API key from the [AgentMail console](https://console.agentmail.to/) +The published plugin includes the official AgentMail CLI for supported macOS, Linux, and Windows +architectures. A separate global CLI installation is not required. + ## Install Build and install a local checkout: @@ -26,26 +31,14 @@ For development, use `openclaw plugins install --link .` so OpenClaw loads this ## Configure -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. +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. -OpenClaw can scope the secrets to this plugin in `~/.openclaw/openclaw.json`: +For a managed Gateway, put them in `~/.openclaw/.env` so the channel and agent turns inherit the +same credentials: -```json5 -{ - plugins: { - entries: { - agentmail: { - enabled: true, - env: { - AGENTMAIL_API_KEY: "am_...", - AGENTMAIL_WEBHOOK_SECRET: "whsec_...", // optional; enables webhook ingress - }, - }, - }, - }, -} +```dotenv +AGENTMAIL_API_KEY=am_... +AGENTMAIL_WEBHOOK_SECRET=whsec_... ``` Keep keys out of source control. Restart the Gateway after installing or changing configuration: @@ -55,13 +48,15 @@ openclaw gateway restart openclaw plugins inspect agentmail --runtime ``` -### Tool config (optional SDK settings) +### CLI config -> **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. +> **Credentials:** the bundled CLI authenticates 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 CLI-backed skill unavailable. -Optional AgentMail SDK settings for the **tools** belong under `plugins.entries.agentmail.config`: +An optional API base URL override for the bundled CLI belongs under +`plugins.entries.agentmail.config`: ```json5 { @@ -69,9 +64,7 @@ Optional AgentMail SDK settings for the **tools** belong under `plugins.entries. entries: { agentmail: { config: { - timeoutSeconds: 30, - maxRetries: 2, - // baseUrl: "https://api.agentmail.to/v0", + baseUrl: "https://api.agentmail.to/v0", }, }, }, @@ -79,6 +72,17 @@ Optional AgentMail SDK settings for the **tools** belong under `plugins.entries. } ``` +The previous `timeoutSeconds` and `maxRetries` tool settings remain accepted so existing +configurations continue to load, but the bundled CLI does not use them. +For credential safety, command arguments cannot override `--api-key`, `--base-url`, or +`--environment`; only inherited credentials and the operator-controlled plugin setting above can +select the AgentMail identity and API endpoint. The passthrough also removes inherited +endpoint-selector and standard proxy environment variables (`HTTP_PROXY`, `HTTPS_PROXY`, +`ALL_PROXY`, and `NO_PROXY`, including lowercase forms). +If a command value must literally begin with `--base-url` or `--environment`, use the CLI's +`--option=value` form (for example, `--subject=--base-url-is-restricted`). The endpoint guard fails +closed for unrecognized separate-value options. + ### Channel config The **channel** is configured under `channels.agentmail` (single inbox) or `channels.agentmail.accounts.` (multiple): @@ -104,33 +108,47 @@ Security defaults worth knowing: - `dmPolicy: "open"` requires `allowFrom` to include `"*"`. - Every reply re-hydrates the triggering message and re-authorizes its `From`, so an untrusted `Reply-To` cannot redirect delivery. -## Tools +## CLI-backed skill + +The plugin registers a passthrough command: -| Tool | Purpose | -| --- | --- | -| `agentmail_list_inboxes` | List available inboxes | -| `agentmail_create_inbox` | Create an inbox, with optional idempotent `clientId` | -| `agentmail_list_messages` | List and filter messages in an inbox | -| `agentmail_search_messages` | Full-text search messages | -| `agentmail_get_message` | Retrieve a complete message | -| `agentmail_send_message` | Send text/HTML email with optional attachments | -| `agentmail_reply_to_message` | Reply or reply-all in an existing thread | -| `agentmail_forward_message` | Forward an existing message | -| `agentmail_update_message_labels` | Add or remove labels, including `read`/`unread` | +```bash +openclaw agentmail -- --help +openclaw agentmail -- --format json inboxes list +openclaw agentmail -- --format json inboxes:messages send \ + --inbox-id agent@agentmail.to \ + --to person@example.com \ + --subject "Hello" \ + --text "Hello from OpenClaw" +``` -Send, reply, and forward accept an optional `idempotencyKey` to make retries safe. Attachments can use Base64-encoded `content` or a public `url`. +Keep the `--` separator so OpenClaw forwards all following flags to AgentMail. The included skill +uses JSON output, consults CLI help instead of guessing flags, and covers inboxes, messages, +threads, drafts, webhooks, domains, pods, API keys, and future CLI resources. + +The CLI skill runs on the OpenClaw host because that is where its executable and credentials are +installed. Sandboxed agents need permission to execute this host command. ## Develop ```bash npm install npm run build # tsc -npm run plugin:build # build + regenerate openclaw.plugin.json +npm run cli:prepare # fetch + verify the current platform's pinned AgentMail CLI +npm run plugin:build # build + prepare CLI + regenerate openclaw.plugin.json npm run plugin:check # fail if the manifest is stale npm test # vitest ``` -`plugin:build` compiles TypeScript and regenerates `openclaw.plugin.json` (tool metadata + channel declarations) via `scripts/build-manifest.mjs`. Commit manifest changes whenever tool metadata or the channel config schema changes. +`plugin:build` compiles TypeScript, downloads the pinned CLI release for the current platform, +verifies its SHA-256 checksum, and regenerates `openclaw.plugin.json`. `npm pack` prepares every +supported CLI target so installation never runs lifecycle scripts or downloads executables. Do not +publish with lifecycle scripts disabled (`--ignore-scripts`), because `prepack` is what assembles +and validates the complete eight-platform vendor tree. + +CLI release version, archive checksums, and extracted-executable checksums live in +`src/cli/agentmail-cli-release.json`. Update that file when intentionally adopting a new AgentMail +CLI release. ## License diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000..0c283d2 --- /dev/null +++ b/THIRD_PARTY_NOTICES.md @@ -0,0 +1,11 @@ +# Third-party notices + +Published packages of this plugin include platform-specific executables from the official +[AgentMail CLI](https://github.com/agentmail-to/agentmail-cli). + +- Component: AgentMail CLI +- Version: 0.7.14 +- License: Apache License 2.0 + +The complete AgentMail CLI license is included in published packages at +`vendor/agentmail/LICENSE`. diff --git a/openclaw.plugin.json b/openclaw.plugin.json index 2975b49..feed78c 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -1,25 +1,25 @@ { "id": "agentmail", "name": "AgentMail", - "description": "AgentMail for OpenClaw: email tools plus a durable, allowlisted, reply-only email channel.", + "description": "AgentMail for OpenClaw: a CLI-backed skill plus a durable, allowlisted, reply-only email channel.", "version": "0.1.0", "configSchema": { "type": "object", "properties": { "baseUrl": { "type": "string", - "description": "Optional AgentMail API base URL override.", + "description": "Optional AgentMail API base URL override for the bundled CLI.", "format": "uri" }, "timeoutSeconds": { "type": "integer", - "description": "Maximum time to wait for an AgentMail API request.", + "description": "Deprecated tool setting retained for upgrade compatibility; the bundled CLI ignores it.", "minimum": 1, "maximum": 300 }, "maxRetries": { "type": "integer", - "description": "Number of times the AgentMail SDK retries a request.", + "description": "Deprecated tool setting retained for upgrade compatibility; the bundled CLI ignores it.", "minimum": 0, "maximum": 10 } @@ -32,6 +32,9 @@ "channels": [ "agentmail" ], + "skills": [ + "./skills" + ], "channelEnvVars": { "agentmail": [ "AGENTMAIL_API_KEY", @@ -236,7 +239,7 @@ }, "mediaMaxMb": { "type": "number", - "exclusiveMinimum": 0, + "minimum": 9.5367431640625e-7, "maximum": 100 }, "accounts": { @@ -439,7 +442,7 @@ }, "mediaMaxMb": { "type": "number", - "exclusiveMinimum": 0, + "minimum": 9.5367431640625e-7, "maximum": 100 } }, @@ -453,7 +456,7 @@ "additionalProperties": false }, "label": "AgentMail", - "description": "AgentMail for OpenClaw: email tools plus a durable, allowlisted, reply-only email channel.", + "description": "AgentMail for OpenClaw: a CLI-backed skill plus a durable, allowlisted, reply-only email channel.", "uiHints": { "": { "label": "AgentMail", @@ -487,18 +490,5 @@ } } } - }, - "contracts": { - "tools": [ - "agentmail_list_inboxes", - "agentmail_create_inbox", - "agentmail_list_messages", - "agentmail_search_messages", - "agentmail_get_message", - "agentmail_send_message", - "agentmail_reply_to_message", - "agentmail_forward_message", - "agentmail_update_message_labels" - ] } } diff --git a/package-lock.json b/package-lock.json index 59ce3d3..ce242d4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,11 +10,10 @@ "license": "MIT", "dependencies": { "agentmail": "^0.5.16", - "svix": "^1.96.1", - "typebox": "^1.1.38", - "zod": "^4.4.3" + "svix": "^1.96.1" }, "devDependencies": { + "fflate": "^0.8.3", "openclaw": "2026.7.2-beta.3", "typescript": "^5.9.0", "vitest": "^3.2.0" @@ -1190,6 +1189,13 @@ } } }, + "node_modules/fflate": { + "version": "0.8.3", + "resolved": "https://registry.npmjs.org/fflate/-/fflate-0.8.3.tgz", + "integrity": "sha512-tbZNuJrLwGUp3zshBtdy4W+ORxZuIh8a5ilyIEQDC5rY1f3U20JMry0Ll3WBzU58EZKsEuJFXhb5gwv8CsPvgA==", + "dev": true, + "license": "MIT" + }, "node_modules/fsevents": { "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", @@ -5709,12 +5715,6 @@ "node": ">=14.0.0" } }, - "node_modules/typebox": { - "version": "1.3.6", - "resolved": "https://registry.npmjs.org/typebox/-/typebox-1.3.6.tgz", - "integrity": "sha512-Sc8RA0NCMEFmApHNU9ZMzqcpQj46She44J8ffpLM/bdhLNUZKq7DJumcLcsFx1gRmDfQPgCgOmFFJ7rcnfWNyA==", - "license": "MIT" - }, "node_modules/typescript": { "version": "5.9.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", @@ -5937,15 +5937,6 @@ "optional": true } } - }, - "node_modules/zod": { - "version": "4.4.3", - "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz", - "integrity": "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ==", - "license": "MIT", - "funding": { - "url": "https://github.com/sponsors/colinhacks" - } } } } diff --git a/package.json b/package.json index 2147a06..656f5c0 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "@agentmail/agentmail", "version": "0.1.0", - "description": "AgentMail for OpenClaw: email tools plus a durable, allowlisted, reply-only email channel.", + "description": "AgentMail for OpenClaw: a CLI-backed skill plus a durable, allowlisted, reply-only email channel.", "type": "module", "private": false, "license": "MIT", @@ -22,27 +22,30 @@ }, "scripts": { "build": "tsc -p tsconfig.json", - "plugin:build": "npm run build && node scripts/build-manifest.mjs", + "cli:prepare": "node scripts/prepare-agentmail-cli.mjs", + "plugin:build": "npm run build && npm run cli:prepare && node scripts/build-manifest.mjs", "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", + "plugin:validate": "npm run build && npm run cli:prepare && node scripts/build-manifest.mjs --check && node scripts/validate-plugin.mjs", + "prepack": "npm run cli:prepare -- --all && npm run plugin:validate", "test": "vitest run --config ./vitest.config.ts" }, "files": [ "dist", + "skills", + "vendor", "openclaw.plugin.json", - "README.md" + "README.md", + "THIRD_PARTY_NOTICES.md" ], "peerDependencies": { "openclaw": ">=2026.7.2-beta.3" }, "dependencies": { "agentmail": "^0.5.16", - "svix": "^1.96.1", - "typebox": "^1.1.38", - "zod": "^4.4.3" + "svix": "^1.96.1" }, "devDependencies": { + "fflate": "^0.8.3", "openclaw": "2026.7.2-beta.3", "typescript": "^5.9.0", "vitest": "^3.2.0" diff --git a/scripts/build-manifest.mjs b/scripts/build-manifest.mjs index f1bf775..279c104 100644 --- a/scripts/build-manifest.mjs +++ b/scripts/build-manifest.mjs @@ -1,36 +1,31 @@ -// Generates openclaw.plugin.json for the combined AgentMail plugin (tool extension + channel -// extension). The stock `openclaw plugins build` codegen only understands tool-plugin metadata, so -// it cannot own a manifest that also declares a channel. This script derives the tool half from the -// compiled tool entry's metadata and merges the channel declarations (channels, channelEnvVars, -// channelConfigs) so both surfaces load from one manifest. +// Generates openclaw.plugin.json for the combined AgentMail plugin (CLI-backed skill + channel +// extension). The stock `openclaw plugins build` codegen only understands tool-plugin metadata, +// while this plugin intentionally exposes the evolving AgentMail API through its bundled CLI. +// This script owns the manifest so the CLI config, plugin skill, and channel declarations remain +// one installable unit. // // Usage: // node scripts/build-manifest.mjs # write openclaw.plugin.json // node scripts/build-manifest.mjs --check # fail if the on-disk manifest is out of date import { readFileSync, writeFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; -import { getToolPluginMetadata } from "openclaw/plugin-sdk/tool-plugin"; -import toolEntry from "../dist/tools/index.js"; +import { agentMailCliConfigJsonSchema } from "../dist/cli/config.js"; import { AgentMailChannelConfigSchema } from "../dist/channel/src/config-schema.js"; const MANIFEST_PATH = fileURLToPath(new URL("../openclaw.plugin.json", import.meta.url)); -const DESCRIPTION = - "AgentMail for OpenClaw: email tools plus a durable, allowlisted, reply-only email channel."; - -const toolMetadata = getToolPluginMetadata(toolEntry); -if (!toolMetadata) { - throw new Error("Tool plugin metadata is unavailable from ./dist/tools/index.js"); -} +const packageJson = JSON.parse(readFileSync(new URL("../package.json", import.meta.url), "utf8")); +const DESCRIPTION = packageJson.description; const manifest = { - id: toolMetadata.id, - name: toolMetadata.name, + id: "agentmail", + name: "AgentMail", description: DESCRIPTION, - version: JSON.parse(readFileSync(new URL("../package.json", import.meta.url), "utf8")).version, - configSchema: toolMetadata.configSchema, - activation: toolMetadata.activation ?? { onStartup: true }, + version: packageJson.version, + configSchema: agentMailCliConfigJsonSchema, + activation: { onStartup: true }, channels: ["agentmail"], + skills: ["./skills"], // The Plugin Inspector emits a `channel-env-vars` deprecation warning whenever this field is // present (it is unconditional — a setup entry does NOT suppress it). We keep it anyway: on our // supported OpenClaw range (peerDependency floor >=2026.7.2-beta.3, currently the newest published @@ -51,9 +46,6 @@ const manifest = { uiHints: AgentMailChannelConfigSchema.uiHints, }, }, - contracts: { - tools: toolMetadata.tools.map((tool) => tool.name), - }, }; const serialized = `${JSON.stringify(manifest, null, 2)}\n`; diff --git a/scripts/prepare-agentmail-cli.mjs b/scripts/prepare-agentmail-cli.mjs new file mode 100644 index 0000000..85bf868 --- /dev/null +++ b/scripts/prepare-agentmail-cli.mjs @@ -0,0 +1,301 @@ +import { spawnSync } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import { unzipSync } from "fflate"; +import { + chmodSync, + copyFileSync, + cpSync, + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + renameSync, + rmSync, + statSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { basename, dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = fileURLToPath(new URL("..", import.meta.url)); +const release = JSON.parse( + readFileSync(new URL("../src/cli/agentmail-cli-release.json", import.meta.url), "utf8"), +); +const vendorRoot = join(root, ...release.vendorDirectory.split("/")); +const vendorParent = dirname(vendorRoot); +const versionPath = join(vendorRoot, "VERSION"); +const preparationLock = `${vendorRoot}.prepare-lock`; +const allTargets = Object.keys(release.assets).sort(); +const INCOMPLETE_LOCK_GRACE_MS = 30_000; + +function processIsRunning(pid) { + if (!Number.isSafeInteger(pid) || pid <= 0) { + return false; + } + try { + process.kill(pid, 0); + return true; + } catch (error) { + return error?.code !== "ESRCH"; + } +} + +function acquirePreparationLock() { + const owner = { + pid: process.pid, + token: randomUUID(), + createdAt: Date.now(), + }; + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + // `wx` makes ownership acquisition atomic. The token prevents this process from deleting a + // replacement lock during cleanup. + writeFileSync(preparationLock, `${JSON.stringify(owner)}\n`, { + encoding: "utf8", + flag: "wx", + mode: 0o600, + }); + return owner.token; + } catch (error) { + if (error?.code !== "EEXIST") { + throw error; + } + } + + let existing; + try { + existing = JSON.parse(readFileSync(preparationLock, "utf8")); + } catch { + // A writer can be between exclusive creation and its synchronous write. Do not steal a + // fresh unreadable lock; an interrupted/incomplete marker becomes recoverable after grace. + let ageMs; + try { + ageMs = Date.now() - statSync(preparationLock).mtimeMs; + } catch (error) { + if (error?.code === "ENOENT") { + continue; + } + throw error; + } + if (ageMs < INCOMPLETE_LOCK_GRACE_MS) { + throw new Error( + `Another AgentMail CLI preparation is already using ${preparationLock}.`, + ); + } + } + if (processIsRunning(existing?.pid)) { + throw new Error( + `Another AgentMail CLI preparation (pid ${existing.pid}) is already using ${preparationLock}.`, + ); + } + + // Rename the exact stale marker out of the lock path before deletion. A competing recovery can + // then win acquisition without either process deleting the other's new lock. + const staleLock = `${preparationLock}.stale-${process.pid}-${randomUUID()}`; + try { + renameSync(preparationLock, staleLock); + rmSync(staleLock, { recursive: true, force: true }); + } catch (error) { + if (error?.code !== "ENOENT") { + throw error; + } + } + } + throw new Error(`Could not acquire AgentMail CLI preparation lock ${preparationLock}.`); +} + +function releasePreparationLock(token) { + try { + const owner = JSON.parse(readFileSync(preparationLock, "utf8")); + if (owner?.pid === process.pid && owner?.token === token) { + rmSync(preparationLock, { force: true }); + } + } catch { + // Best effort. A missing or replaced lock belongs to no work this process may clean up. + } +} + +function currentTarget() { + const key = `${process.platform}-${process.arch}`; + if (!release.assets[key]) { + throw new Error(`AgentMail CLI ${release.version} has no packaged target for ${key}.`); + } + return key; +} + +function sha256(path) { + return createHash("sha256").update(readFileSync(path)).digest("hex"); +} + +function preparedTargetIsValid(target, destinationRoot) { + const metadata = release.assets[target]; + const executable = join(destinationRoot, target, metadata.executableName); + if ( + !existsSync(executable) || + typeof metadata.executableSha256 !== "string" || + sha256(executable) !== metadata.executableSha256 + ) { + return false; + } + return target.startsWith("win32-") || (statSync(executable).mode & 0o111) !== 0; +} + +async function download(url, destination) { + const response = await fetch(url, { redirect: "follow" }); + if (!response.ok || !response.body) { + throw new Error(`Failed to download ${url}: HTTP ${response.status}`); + } + writeFileSync(destination, Buffer.from(await response.arrayBuffer())); +} + +function extract(archive, destination, executableName) { + mkdirSync(destination, { recursive: true }); + if (!archive.endsWith(".tar.gz")) { + const entries = unzipSync(new Uint8Array(readFileSync(archive))); + const executable = entries[executableName]; + if (!executable) { + throw new Error(`${basename(archive)} did not contain ${executableName} at its root.`); + } + writeFileSync(join(destination, executableName), executable); + return; + } + const result = spawnSync("tar", ["-xzf", archive, "-C", destination], { + stdio: "inherit", + }); + if (result.error) { + throw result.error; + } + if (result.status !== 0) { + throw new Error(`Failed to extract ${basename(archive)} (exit ${result.status}).`); + } +} + +async function prepareTarget(target, temporaryRoot, destinationRoot) { + const metadata = release.assets[target]; + const executableName = metadata.executableName; + const destination = join(destinationRoot, target, executableName); + if (preparedTargetIsValid(target, destinationRoot)) { + console.log(`AgentMail CLI ${release.version} already prepared for ${target}.`); + return; + } + // A stale version marker must not bless a corrupt, wrong, or non-executable cached binary. + // Remove only this staged target; the live vendor tree remains untouched until the full swap. + rmSync(destination, { force: true }); + + const archivePath = join(temporaryRoot, metadata.archive); + const url = + `https://github.com/${release.repository}/releases/download/` + + `v${release.version}/${metadata.archive}`; + console.log(`Downloading AgentMail CLI ${release.version} for ${target}...`); + await download(url, archivePath); + + const actualChecksum = sha256(archivePath); + if (actualChecksum !== metadata.sha256) { + throw new Error( + `Checksum mismatch for ${metadata.archive}: expected ${metadata.sha256}, got ${actualChecksum}.`, + ); + } + + const extractRoot = join(temporaryRoot, target); + extract(archivePath, extractRoot, executableName); + const extractedExecutable = join(extractRoot, executableName); + if (!existsSync(extractedExecutable)) { + throw new Error(`${metadata.archive} did not contain ${executableName} at its root.`); + } + + mkdirSync(dirname(destination), { recursive: true }); + const stagedDestination = `${destination}.tmp-${process.pid}`; + copyFileSync(extractedExecutable, stagedDestination); + if (!target.startsWith("win32-")) { + chmodSync(stagedDestination, 0o755); + } + renameSync(stagedDestination, destination); + if (!preparedTargetIsValid(target, destinationRoot)) { + throw new Error(`Prepared AgentMail CLI executable failed validation for ${target}.`); + } +} + +async function main() { + mkdirSync(vendorParent, { recursive: true }); + const preparationLockToken = acquirePreparationLock(); + + let temporaryRoot; + let stagingParent; + const backupRoot = `${vendorRoot}.backup-${process.pid}`; + + try { + const targets = process.argv.includes("--all") ? allTargets : [currentTarget()]; + temporaryRoot = mkdtempSync(join(tmpdir(), "agentmail-cli-prepare-")); + stagingParent = mkdtempSync(join(vendorParent, ".agentmail-cli-stage-")); + const stagedVendorRoot = join(stagingParent, basename(vendorRoot)); + const existingVersion = existsSync(versionPath) + ? readFileSync(versionPath, "utf8").trim() + : undefined; + const requestedTreeIsComplete = + existingVersion === release.version && + existsSync(join(vendorRoot, "LICENSE")) && + targets.every((target) => preparedTargetIsValid(target, vendorRoot)); + if (requestedTreeIsComplete) { + for (const target of targets) { + console.log(`AgentMail CLI ${release.version} already prepared for ${target}.`); + } + } else { + // Build a complete replacement tree beside the live one. A failed download, checksum, + // extraction, or LICENSE fetch leaves the currently installed tree untouched. + if (existingVersion === release.version && existsSync(vendorRoot)) { + cpSync(vendorRoot, stagedVendorRoot, { recursive: true }); + } else { + mkdirSync(stagedVendorRoot, { recursive: true }); + } + + for (const target of targets) { + await prepareTarget(target, temporaryRoot, stagedVendorRoot); + } + + const licensePath = join(stagedVendorRoot, "LICENSE"); + if (!existsSync(licensePath)) { + const downloadedLicense = join(temporaryRoot, "LICENSE"); + await download( + `https://raw.githubusercontent.com/${release.repository}/v${release.version}/LICENSE`, + downloadedLicense, + ); + copyFileSync(downloadedLicense, licensePath); + } + // The marker lives only in the staged tree and therefore becomes visible with the complete + // requested tree, never before its binaries and license. + writeFileSync(join(stagedVendorRoot, "VERSION"), `${release.version}\n`); + + let originalMoved = false; + try { + if (existsSync(vendorRoot)) { + renameSync(vendorRoot, backupRoot); + originalMoved = true; + } + renameSync(stagedVendorRoot, vendorRoot); + } catch (error) { + if (originalMoved && !existsSync(vendorRoot) && existsSync(backupRoot)) { + renameSync(backupRoot, vendorRoot); + } + throw error; + } + if (originalMoved) { + rmSync(backupRoot, { recursive: true, force: true }); + } + } + + console.log( + `Prepared AgentMail CLI ${release.version} for ${targets.length} target${targets.length === 1 ? "" : "s"}.`, + ); + } finally { + if (temporaryRoot) { + rmSync(temporaryRoot, { recursive: true, force: true }); + } + if (stagingParent) { + rmSync(stagingParent, { recursive: true, force: true }); + } + releasePreparationLock(preparationLockToken); + } +} + +await main(); diff --git a/scripts/validate-plugin.mjs b/scripts/validate-plugin.mjs index 4ed272f..3fbe248 100644 --- a/scripts/validate-plugin.mjs +++ b/scripts/validate-plugin.mjs @@ -1,15 +1,20 @@ // Real OpenClaw host validation: link-install the built plugin and confirm the host actually loads -// it with both capabilities (the AgentMail channel and all declared tools). This catches invalid -// manifest/contract/schema shapes that a pure staleness check cannot — they would otherwise only -// surface at user startup. Kept separate from `plugin:check` (manifest staleness). +// its AgentMail channel and CLI command, then execute the packaged CLI. This catches invalid +// manifest/schema/command shapes and missing executables that a pure staleness check cannot. +// Kept separate from `plugin:check` (manifest staleness). import { execFileSync } from "node:child_process"; -import { mkdtempSync, rmSync } from "node:fs"; +import { createHash } from "node:crypto"; +import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; -import { join } from "node:path"; +import { delimiter, dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; const root = fileURLToPath(new URL("..", import.meta.url)); const openclaw = fileURLToPath(new URL("../node_modules/.bin/openclaw", import.meta.url)); +const cliRunner = await import(new URL("../dist/cli/runner.js", import.meta.url)); +const cliRelease = JSON.parse( + readFileSync(new URL("../dist/cli/agentmail-cli-release.json", import.meta.url), "utf8"), +); // Run against an isolated, disposable state dir so validation (invoked from prepack during // `npm pack`/`npm publish`) never touches the maintainer's real OpenClaw installation or its @@ -17,9 +22,11 @@ const openclaw = fileURLToPath(new URL("../node_modules/.bin/openclaw", import.m const stateDir = mkdtempSync(join(tmpdir(), "agentmail-plugin-validate-")); const childEnv = { ...process.env, + AGENTMAIL_API_KEY: process.env.AGENTMAIL_API_KEY || "am_plugin_validation", OPENCLAW_STATE_DIR: stateDir, OPENCLAW_CONFIG_DIR: stateDir, OPENCLAW_HOME: stateDir, + PATH: `${dirname(openclaw)}${delimiter}${process.env.PATH || ""}`, NO_COLOR: "1", FORCE_COLOR: "0", }; @@ -68,15 +75,117 @@ if (!/Status:\s*loaded/.test(inspect)) { if (!/channel:\s*agentmail/.test(inspect)) { fail("plugin manifest did not register the agentmail channel", inspect); } -const requiredTools = [ - "agentmail_list_inboxes", - "agentmail_create_inbox", - "agentmail_send_message", - "agentmail_reply_to_message", -]; -const missing = requiredTools.filter((tool) => !inspect.includes(tool)); -if (missing.length > 0) { - fail(`plugin did not register expected tools: ${missing.join(", ")}`, inspect); +if (!/CLI commands:\s*[\s\S]*\bagentmail\b/.test(inspect)) { + fail("plugin runtime did not register the agentmail CLI command", inspect); +} +if (/agentmail_list_inboxes/.test(inspect)) { + fail("plugin still registered the retired fixed AgentMail tool surface", inspect); +} + +const expectedCliVersion = readFileSync( + cliRunner.resolveAgentMailCliVendorPath("VERSION"), + "utf8", +).trim(); +const currentCliTarget = cliRunner.resolveAgentMailCliTarget().directory; +for (const [target, metadata] of Object.entries(cliRelease.assets)) { + const executable = cliRunner.resolveAgentMailCliVendorPath( + target, + metadata.executableName, + ); + if (!existsSync(executable)) { + if (target === currentCliTarget) { + fail(`bundled AgentMail CLI executable is missing for ${target}`); + } + continue; + } + const executableSha256 = createHash("sha256") + .update(readFileSync(executable)) + .digest("hex"); + if (executableSha256 !== metadata.executableSha256) { + fail( + `bundled AgentMail CLI executable checksum did not match for ${target}`, + `expected ${metadata.executableSha256}, got ${executableSha256}`, + ); + } + if (!target.startsWith("win32-") && (statSync(executable).mode & 0o111) === 0) { + fail(`bundled AgentMail CLI executable is not executable for ${target}`); + } +} +let cliHelp = ""; +try { + cliHelp = execFileSync(cliRunner.resolveAgentMailCliExecutable(), ["--help"], { + cwd: root, + encoding: "utf8", + stdio: ["ignore", "pipe", "pipe"], + env: cliRunner.sanitizeAgentMailCliEnvironment(childEnv), + }); +} catch (error) { + fail( + "could not inspect the bundled AgentMail CLI global options", + error.stdout || error.stderr || String(error), + ); +} +const restrictedOptions = cliHelp + .split("\n") + .filter((line) => /(--api-key\b|base URL|environment for API requests)/i.test(line)) + .flatMap((line) => + [...line.matchAll(/(?:^|[\s,])-+([a-z][a-z0-9-]*)\b/gi)].map((match) => + match[1].toLowerCase(), + ), + ) + .filter((value, index, values) => values.indexOf(value) === index) + .sort(); +const guardedRestrictedOptions = [...cliRunner.AGENTMAIL_CLI_RESTRICTED_OPTIONS].sort(); +if (JSON.stringify(restrictedOptions) !== JSON.stringify(guardedRestrictedOptions)) { + fail( + "bundled AgentMail CLI credential/endpoint options no longer match the passthrough guard", + `CLI: ${restrictedOptions.join(", ") || "(none)"}; guard: ${guardedRestrictedOptions.join(", ")}`, + ); +} +for (const option of restrictedOptions) { + for (const prefix of ["-", "--", "---"]) { + for (const args of [ + [`${prefix}${option}`, "untrusted", "inboxes", "list"], + [`${prefix}${option}=untrusted`, "inboxes", "list"], + ]) { + try { + cliRunner.withConfiguredBaseUrl(args, undefined); + fail(`endpoint guard accepted ${args[0]}`); + } catch (error) { + if (!String(error).includes("endpoint overrides are restricted")) { + throw error; + } + } + } + } +} +let cliVersion = ""; +try { + cliVersion = run(["agentmail", "--", "--version"]); +} catch (error) { + fail( + "openclaw could not execute the bundled AgentMail CLI", + error.stdout || error.stderr || String(error), + ); +} +if (!cliVersion.includes(expectedCliVersion)) { + fail( + `bundled AgentMail CLI version did not match ${expectedCliVersion}`, + cliVersion, + ); +} + +let skills = ""; +try { + skills = run(["skills", "list", "--eligible"]); +} catch (error) { + fail( + "openclaw could not list eligible skills", + error.stdout || error.stderr || String(error), + ); +} +if (!/\bagentmail\b/i.test(skills)) { + fail("plugin did not expose its AgentMail CLI skill", skills); } // Surface any load diagnostics the host reports for this plugin. @@ -92,4 +201,6 @@ try { // doctor is best-effort; the inspect assertions above are the authoritative gate. } -console.log("plugin:validate OK — host loaded the AgentMail channel and tools."); +console.log( + `plugin:validate OK — host loaded the AgentMail channel and AgentMail CLI ${expectedCliVersion}.`, +); diff --git a/skills/agentmail/SKILL.md b/skills/agentmail/SKILL.md new file mode 100644 index 0000000..5b770b9 --- /dev/null +++ b/skills/agentmail/SKILL.md @@ -0,0 +1,90 @@ +--- +name: agentmail +description: "Manage AgentMail inboxes, messages, threads, drafts, webhooks, domains, pods, and API keys with the bundled AgentMail CLI." +metadata: + { + "openclaw": + { + "requires": { "bins": ["openclaw"], "env": ["AGENTMAIL_API_KEY"] }, + "primaryEnv": "AGENTMAIL_API_KEY", + }, + } +--- + +# AgentMail CLI + +Use the AgentMail CLI bundled with this plugin through: + +```bash +openclaw agentmail -- +``` + +Always put `--` after `agentmail`; it prevents OpenClaw from interpreting AgentMail flags. +Never pass `--api-key`, `--base-url`, `--environment`, or print the API key. Authentication is +inherited from `AGENTMAIL_API_KEY`, and only operator-controlled plugin configuration may select +the API endpoint. +If a literal command value begins with `--base-url` or `--environment`, pass it with the containing +option's `--option=value` form so the endpoint guard can distinguish it from a global override. + +Run this command on the OpenClaw host, not inside an agent sandbox. The bundled executable and +skill-scoped credentials are available only to host execution. + +## Discover commands dynamically + +The CLI is the source of truth. Inspect help before guessing a resource, command, or flag: + +```bash +openclaw agentmail -- --help +openclaw agentmail -- inboxes --help +openclaw agentmail -- inboxes:messages --help +openclaw agentmail -- inboxes:messages send --help +``` + +Prefer machine-readable output: + +```bash +openclaw agentmail -- --format json --format-error json inboxes list +``` + +Use `--transform ''` when it can reduce a large response without losing +information needed for the task. + +## Common operations + +```bash +# Inboxes +openclaw agentmail -- --format json inboxes list +openclaw agentmail -- --format json inboxes create --display-name "My Agent" +openclaw agentmail -- --format json inboxes get --inbox-id + +# Messages +openclaw agentmail -- --format json inboxes:messages list --inbox-id +openclaw agentmail -- --format json inboxes:messages get \ + --inbox-id --message-id +openclaw agentmail -- --format json inboxes:messages send \ + --inbox-id \ + --to recipient@example.com \ + --subject "Subject" \ + --text "Plain-text body" +openclaw agentmail -- --format json inboxes:messages reply \ + --inbox-id --message-id --text "Reply body" +openclaw agentmail -- --format json inboxes:messages forward \ + --inbox-id --message-id --to recipient@example.com + +# Threads and drafts +openclaw agentmail -- --format json inboxes:threads list --inbox-id +openclaw agentmail -- --format json inboxes:drafts list --inbox-id + +# Other API resources +openclaw agentmail -- webhooks --help +openclaw agentmail -- domains --help +openclaw agentmail -- pods --help +openclaw agentmail -- api-keys --help +``` + +Before destructive operations, bulk changes, creating credentials, or sending email to a new +recipient, follow the user's approval and confirmation policy. Use idempotency flags when the +command help exposes them and an operation may be retried. + +For attachments, the CLI accepts `@path`, `@file://path`, and `@data://path` arguments. Read only +files the user has authorized and prefer `@data://` for binary content. diff --git a/src/channel/src/accounts.test.ts b/src/channel/src/accounts.test.ts index d66705f..71b8327 100644 --- a/src/channel/src/accounts.test.ts +++ b/src/channel/src/accounts.test.ts @@ -6,6 +6,7 @@ import { resolveAgentMailAccount, } from "./accounts.js"; import { AgentMailChannelConfigSchema } from "./config-schema.js"; +import { AGENTMAIL_MEDIA_MIN_MB } from "./config-schema.js"; import type { AgentMailChannelConfig } from "./types.js"; const paddedApi = " api-key "; @@ -171,9 +172,13 @@ describe("AgentMail account config", () => { expect(ids).not.toContain("Sales-US"); }); - it("rejects an impractically large mediaMaxMb", () => { + it("accepts positive fractional mediaMaxMb values and rejects zero or excessive values", () => { const runtime = AgentMailChannelConfigSchema.runtime; expect(runtime?.safeParse({ mediaMaxMb: 25 }).success).toBe(true); + expect(runtime?.safeParse({ mediaMaxMb: 0.5 }).success).toBe(true); + expect(runtime?.safeParse({ mediaMaxMb: AGENTMAIL_MEDIA_MIN_MB }).success).toBe(true); + expect(runtime?.safeParse({ mediaMaxMb: AGENTMAIL_MEDIA_MIN_MB / 2 }).success).toBe(false); + expect(runtime?.safeParse({ mediaMaxMb: 0 }).success).toBe(false); expect(runtime?.safeParse({ mediaMaxMb: 100_000 }).success).toBe(false); }); diff --git a/src/channel/src/accounts.ts b/src/channel/src/accounts.ts index a473c91..86d4eea 100644 --- a/src/channel/src/accounts.ts +++ b/src/channel/src/accounts.ts @@ -166,6 +166,32 @@ export function isAgentMailAccountConfigured(account: ResolvedAgentMailAccount): return Boolean(account.apiKey && account.inboxId); } +const inboxOwnersByConfig = new WeakMap>(); + +function resolveAgentMailInboxOwners(cfg: OpenClawConfig): Map { + const configIdentity = cfg as object; + const cached = inboxOwnersByConfig.get(configIdentity); + if (cached) { + return cached; + } + const owners = new Map(); + for (const accountId of listAgentMailAccountIds(cfg)) { + const candidate = resolveAgentMailAccount(cfg, accountId); + if (!candidate.enabled || !isAgentMailAccountConfigured(candidate)) { + continue; + } + const inboxId = normalizeAgentMailInboxId(candidate.inboxId); + const owner = owners.get(inboxId); + if (owner === undefined || candidate.accountId < owner) { + owners.set(inboxId, candidate.accountId); + } + } + // OpenClaw passes a fresh configuration snapshot on reload. Cache its ownership index so + // starting n accounts resolves the account set once instead of rescanning it n times. + inboxOwnersByConfig.set(configIdentity, owners); + return owners; +} + /** * Returns the id of an earlier-sorted account that also targets this account's inbox, or null when * the inbox is unique. Two accounts consuming one inbox would open separate durable journals (keyed @@ -179,24 +205,10 @@ export function findConflictingAgentMailInboxOwner( if (!account.inboxId) { return null; } - for (const otherId of listAgentMailAccountIds(cfg)) { - const other = resolveAgentMailAccount(cfg, otherId); - // Compare canonical (resolved) ids with a strict byte order — never localeCompare, whose ties - // between distinct ids could let both accounts believe they are the owner and double-reply. - if (other.accountId === account.accountId || other.accountId >= account.accountId) { - continue; - } - // Only an enabled AND fully-configured account can own the inbox; an enabled-but-unconfigured - // account must not block the real consumer and silently disable inbound mail. - if ( - other.enabled && - isAgentMailAccountConfigured(other) && - normalizeAgentMailInboxId(other.inboxId) === normalizeAgentMailInboxId(account.inboxId) - ) { - return other.accountId; - } - } - return null; + const owner = resolveAgentMailInboxOwners(cfg).get( + normalizeAgentMailInboxId(account.inboxId), + ); + return owner && owner < account.accountId ? owner : null; } export function inspectAgentMailAccount(cfg: OpenClawConfig, accountId?: string | null) { diff --git a/src/channel/src/catch-up.test.ts b/src/channel/src/catch-up.test.ts index 73ef9a3..f0c5066 100644 --- a/src/channel/src/catch-up.test.ts +++ b/src/channel/src/catch-up.test.ts @@ -5,13 +5,27 @@ import type { } from "openclaw/plugin-sdk/plugin-state-runtime"; import { describe, expect, it, vi } from "vitest"; import { + AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS, + AGENTMAIL_REST_CATCH_UP_LEGACY_MAX_ACCOUNTS, + AGENTMAIL_REST_CATCH_UP_MAX_ENTRIES_PER_ACCOUNT, + AGENTMAIL_REST_CATCH_UP_NAMESPACE, AGENTMAIL_REST_CATCH_UP_OVERLAP_MS, + type AgentMailCatchUpCursor, createAgentMailCatchUpSession, createAgentMailCatchUpSupervisor, } from "./catch-up.js"; -import { AgentMailIngressCapacityError } from "./durable-receive.js"; +import { sha256Hex } from "./digest.js"; +import { + AGENTMAIL_DURABLE_PENDING_TTL_MS, + AgentMailIngressCapacityError, +} from "./durable-receive.js"; import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; +const openKeyedStore = vi.hoisted(() => vi.fn()); +vi.mock("./runtime.js", () => ({ + getAgentMailRuntime: () => ({ state: { openKeyedStore } }), +})); + const account: ResolvedAgentMailAccount = { accountId: "default", enabled: true, @@ -24,17 +38,26 @@ const account: ResolvedAgentMailAccount = { mediaMaxBytes: 20 * 1024 * 1024, }; -function memoryStore(): PluginStateKeyedStore { +function memoryStore(maxEntries = Number.POSITIVE_INFINITY): PluginStateKeyedStore { const values = new Map>(); + const setValue = (key: string, value: T) => { + if (!values.has(key) && values.size >= maxEntries) { + const oldestKey = values.keys().next().value; + if (oldestKey !== undefined) { + values.delete(oldestKey); + } + } + values.set(key, { key, value, createdAt: Date.now() }); + }; return { async register(key, value) { - values.set(key, { key, value, createdAt: Date.now() }); + setValue(key, value); }, async registerIfAbsent(key, value) { if (values.has(key)) { return false; } - values.set(key, { key, value, createdAt: Date.now() }); + setValue(key, value); return true; }, async update(key, updateValue) { @@ -42,7 +65,7 @@ function memoryStore(): PluginStateKeyedStore { if (next === undefined) { return false; } - values.set(key, { key, value: next, createdAt: Date.now() }); + setValue(key, next); return true; }, async lookup(key) { @@ -86,6 +109,214 @@ function message(params: { } describe("AgentMail durable REST catch-up", () => { + it("isolates each account cursor in a one-entry state namespace", async () => { + openKeyedStore.mockReset(); + const legacyStore = memoryStore(); + openKeyedStore.mockImplementation(({ namespace }) => + namespace === AGENTMAIL_REST_CATCH_UP_NAMESPACE ? legacyStore : memoryStore(), + ); + const client = { inboxes: { messages: { list: vi.fn() } } } as never; + + await createAgentMailCatchUpSession({ account, client, now: () => 1_000 }); + await createAgentMailCatchUpSession({ + account: { ...account, accountId: "support", inboxId: "inbox_2" }, + client, + now: () => 1_000, + }); + + expect(openKeyedStore).toHaveBeenCalledTimes(4); + const options = openKeyedStore.mock.calls.map(([value]) => value); + const accountOptions = options.filter( + ({ overflowPolicy }) => overflowPolicy === "evict-oldest", + ); + expect(accountOptions).toHaveLength(2); + expect(accountOptions[0]).toMatchObject({ + maxEntries: AGENTMAIL_REST_CATCH_UP_MAX_ENTRIES_PER_ACCOUNT, + overflowPolicy: "evict-oldest", + defaultTtlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS, + }); + expect(accountOptions[0].namespace).toMatch( + new RegExp(`^${AGENTMAIL_REST_CATCH_UP_NAMESPACE}\\.`), + ); + expect(accountOptions[1].namespace).not.toBe(accountOptions[0].namespace); + }); + + it("reuses an account namespace when its inbox rotates", async () => { + openKeyedStore.mockReset(); + const stores = new Map>(); + openKeyedStore.mockImplementation(({ namespace }) => { + const existing = stores.get(namespace); + if (existing) { + return existing; + } + const created = memoryStore(); + stores.set(namespace, created); + return created; + }); + const client = { inboxes: { messages: { list: vi.fn() } } } as never; + + await createAgentMailCatchUpSession({ account, client, now: () => 1_000 }); + await createAgentMailCatchUpSession({ + account: { ...account, inboxId: "inbox_rotated" }, + client, + now: () => 2_000, + }); + + const accountNamespaces = openKeyedStore.mock.calls + .map(([options]) => options) + .filter(({ overflowPolicy }) => overflowPolicy === "evict-oldest") + .map(({ namespace }) => namespace); + expect(accountNamespaces).toHaveLength(2); + expect(new Set(accountNamespaces).size).toBe(1); + }); + + it("fences an overlapping session after a replacement claims the cursor", async () => { + const store = memoryStore(); + const list = vi.fn(async () => ({ count: 0, messages: [] })); + const client = { inboxes: { messages: { list } } } as never; + const oldSession = await createAgentMailCatchUpSession({ + account, + client, + store, + now: () => 1_000, + }); + const replacement = await createAgentMailCatchUpSession({ + account, + client, + store, + now: () => 2_000, + }); + + await oldSession.run({ receive: vi.fn(), abortSignal: new AbortController().signal }); + expect(list).not.toHaveBeenCalled(); + + await replacement.run({ receive: vi.fn(), abortSignal: new AbortController().signal }); + expect(list).toHaveBeenCalledOnce(); + }); + + it("prevents a late old-inbox write from evicting the rotated inbox cursor", async () => { + const store = memoryStore(1); + let finishOldPage!: (page: { + count: number; + messages: AgentMail.MessageItem[]; + }) => void; + const oldList = vi.fn( + async () => + await new Promise<{ count: number; messages: AgentMail.MessageItem[] }>((resolve) => { + finishOldPage = resolve; + }), + ); + const oldSession = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list: oldList } } } as never, + store, + now: () => 1_000, + }); + const oldRun = oldSession.run({ + receive: vi.fn(async () => undefined), + abortSignal: new AbortController().signal, + }); + await vi.waitFor(() => expect(oldList).toHaveBeenCalledOnce()); + + const rotatedAccount = { ...account, inboxId: "inbox_2" }; + const replacementList = vi.fn(async () => ({ count: 0, messages: [] })); + const replacement = await createAgentMailCatchUpSession({ + account: rotatedAccount, + client: { inboxes: { messages: { list: replacementList } } } as never, + store, + now: () => 2_000, + }); + finishOldPage({ + count: 1, + messages: [message({ id: "late_old", timestamp: 1_500 })], + }); + await oldRun; + + await expect( + replacement.run({ + receive: vi.fn(), + abortSignal: new AbortController().signal, + }), + ).resolves.toBeUndefined(); + expect(replacementList).toHaveBeenCalledOnce(); + expect((await store.entries()).map((entry) => entry.key)).toEqual([ + sha256Hex("default\ninbox_2"), + ]); + }); + + it("migrates a legacy shared cursor before establishing a new baseline", async () => { + openKeyedStore.mockReset(); + const legacyStore = memoryStore(); + const accountStore = memoryStore(); + const legacyCursor: AgentMailCatchUpCursor = { + version: 1, + baselineAtMs: 500, + highWaterAtMs: 900, + established: true, + }; + await legacyStore.register(sha256Hex("default\ninbox_1"), legacyCursor); + legacyStore.delete = vi.fn(async () => true); + openKeyedStore.mockImplementation(({ namespace }) => + namespace === AGENTMAIL_REST_CATCH_UP_NAMESPACE ? legacyStore : accountStore, + ); + const list = vi.fn(async () => ({ count: 0, messages: [] })); + + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + now: () => 10_000, + }); + await session.run({ + receive: vi.fn(), + abortSignal: new AbortController().signal, + }); + + expect(list).toHaveBeenCalledWith( + "inbox_1", + expect.objectContaining({ + after: new Date(Math.max(0, 900 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS)), + }), + expect.any(Object), + ); + expect(legacyStore.delete).toHaveBeenCalledOnce(); + expect(openKeyedStore).toHaveBeenCalledWith({ + namespace: AGENTMAIL_REST_CATCH_UP_NAMESPACE, + maxEntries: AGENTMAIL_REST_CATCH_UP_LEGACY_MAX_ACCOUNTS, + overflowPolicy: "reject-new", + }); + }); + + it("retries legacy cursor cleanup after a transient delete failure", async () => { + openKeyedStore.mockReset(); + const key = sha256Hex("default\ninbox_1"); + const legacyStore = memoryStore(); + const accountStore = memoryStore(); + await legacyStore.register(key, { + version: 1, + baselineAtMs: 500, + highWaterAtMs: 900, + established: true, + }); + const deleteLegacy = legacyStore.delete.bind(legacyStore); + legacyStore.delete = vi + .fn<(key: string) => Promise>() + .mockRejectedValueOnce(new Error("database busy")) + .mockImplementation(deleteLegacy); + openKeyedStore.mockImplementation(({ namespace }) => + namespace === AGENTMAIL_REST_CATCH_UP_NAMESPACE ? legacyStore : accountStore, + ); + const warn = vi.fn(); + const client = { inboxes: { messages: { list: vi.fn() } } } as never; + + await createAgentMailCatchUpSession({ account, client, now: () => 10_000, log: { warn } }); + expect(warn).toHaveBeenCalledWith(expect.stringContaining("could not remove")); + expect(await legacyStore.lookup(key)).toBeDefined(); + + await createAgentMailCatchUpSession({ account, client, now: () => 11_000, log: { warn } }); + expect(legacyStore.delete).toHaveBeenCalledTimes(2); + expect(await legacyStore.lookup(key)).toBeUndefined(); + }); + it("coalesces recovery requests into one bounded retry supervisor", async () => { const run = vi .fn<() => Promise>() @@ -264,6 +495,41 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("repairs a stored cursor after the host clock moves backwards", async () => { + const store = memoryStore(); + const key = sha256Hex("default\ninbox_1"); + await store.register(key, { + version: 1, + baselineAtMs: 2_000_000, + highWaterAtMs: 3_000_000, + established: true, + }); + const list = vi.fn(async () => ({ count: 0, messages: [] })); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store, + now: () => 1_000_000, + }); + + await session.run({ + receive: vi.fn(), + abortSignal: new AbortController().signal, + sinceBaseline: true, + }); + + expect(list).toHaveBeenCalledWith( + "inbox_1", + expect.objectContaining({ after: new Date(1_000_000) }), + expect.any(Object), + ); + expect(await store.lookup(key)).toMatchObject({ + baselineAtMs: 1_000_000, + highWaterAtMs: 1_000_000, + established: true, + }); + }); + it("lists from the monitoring baseline on a deep sweep", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ @@ -293,7 +559,7 @@ describe("AgentMail durable REST catch-up", () => { ); }); - it("floors a deep sweep at the dedupe horizon instead of the permanent baseline", async () => { + it("keeps deep recovery at the baseline beyond the former seven-day horizon", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ count: 0, messages: [] })); let nowMs = 1_000; @@ -307,16 +573,15 @@ describe("AgentMail durable REST catch-up", () => { await session.run({ receive, abortSignal: new AbortController().signal }); // baseline = 1_000 nowMs = 1_000 + 8 * 24 * 60 * 60 * 1000; // 8 days later await session.run({ receive, abortSignal: new AbortController().signal, sinceBaseline: true }); - // Older than the 7-day completed-tombstone TTL is not scanned, so the floor is now - 7d, not 1_000. - const expectedFloor = nowMs - 7 * 24 * 60 * 60 * 1000; + // Recovery remains at the monitoring baseline because pending ingress is retained for 30 days. expect(list).toHaveBeenLastCalledWith( "inbox_1", - expect.objectContaining({ after: new Date(expectedFloor) }), + expect.objectContaining({ after: new Date(1_000) }), expect.any(Object), ); }); - it("floors an idle high-water sweep at the dedupe horizon", async () => { + it("floors an idle high-water sweep at the durable recovery horizon", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ count: 0, messages: [] })); let nowMs = 1_000; @@ -328,11 +593,11 @@ describe("AgentMail durable REST catch-up", () => { }); const receive = vi.fn(async () => undefined); await session.run({ receive, abortSignal: new AbortController().signal }); // establish, highWater≈1_000 - nowMs = 1_000 + 8 * 24 * 60 * 60 * 1000; // 8 days later, inbox stayed idle + nowMs = 1_000 + AGENTMAIL_DURABLE_PENDING_TTL_MS + 24 * 60 * 60 * 1000; await session.run({ receive, abortSignal: new AbortController().signal }); // normal high-water sweep - // The high-water cursor never advanced, so without a floor the sweep would re-scan pre-horizon - // mail; it must instead query from now - 7d. - const expectedFloor = nowMs - 7 * 24 * 60 * 60 * 1000; + // The high-water cursor never advanced, so the sweep is bounded by the same 30-day horizon as + // durable pending ingress and completed tombstones. + const expectedFloor = nowMs - AGENTMAIL_DURABLE_PENDING_TTL_MS; expect(list).toHaveBeenLastCalledWith( "inbox_1", expect.objectContaining({ after: new Date(expectedFloor) }), @@ -340,6 +605,61 @@ describe("AgentMail durable REST catch-up", () => { ); }); + it("advances past malformed timestamps and clamps implausibly future timestamps", async () => { + const store = memoryStore(); + const key = sha256Hex("default\ninbox_1"); + await store.register(key, { + version: 1, + baselineAtMs: 100_000, + highWaterAtMs: 200_000, + established: true, + }); + const malformed = { + ...message({ id: "malformed", timestamp: 900_000 }), + timestamp: new Date(Number.NaN), + }; + const farFuture = message({ id: "future", timestamp: 100_000_000 }); + const list = vi + .fn() + .mockResolvedValueOnce({ count: 2, messages: [malformed, farFuture] }) + .mockResolvedValueOnce({ count: 0, messages: [] }); + const warn = vi.fn(); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store, + now: () => 1_000_000, + log: { warn }, + }); + const receive = vi.fn(async () => undefined); + + await session.run({ receive, abortSignal: new AbortController().signal }); + await session.run({ receive, abortSignal: new AbortController().signal }); + + expect(receive.mock.calls.map(([record]) => [record.messageId, record.receivedAt])).toEqual([ + ["malformed", 1_000_000], + ["future", 1_000_000], + ]); + expect(list).toHaveBeenNthCalledWith( + 1, + "inbox_1", + expect.objectContaining({ + before: new Date(1_000_001), + }), + expect.any(Object), + ); + expect(list).toHaveBeenNthCalledWith( + 2, + "inbox_1", + expect.objectContaining({ + after: new Date(1_000_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), + }), + expect.any(Object), + ); + expect(warn).toHaveBeenCalledWith(expect.stringContaining("invalid timestamp")); + expect(warn).toHaveBeenCalledWith(expect.stringContaining("clamped future timestamp")); + }); + it("pauses the pass when durable ingress reports capacity", async () => { const store = memoryStore(); const list = vi.fn(async () => ({ @@ -411,4 +731,107 @@ describe("AgentMail durable REST catch-up", () => { "message_2", ]); }); + + it("resumes a saved continuation after capacity pauses a later page", async () => { + const store = memoryStore(); + const key = sha256Hex("default\ninbox_1"); + await store.register(key, { + version: 1, + baselineAtMs: 100_000, + highWaterAtMs: 400_000, + established: true, + }); + const firstPage = { + count: 1, + messages: [message({ id: "newer", timestamp: 900_000 })], + nextPageToken: "page_2", + }; + const secondPage = { + count: 1, + // Defensive case: provider pagination claims ascending order but returns a later page with a + // timestamp below both the first page and the overlap window. + messages: [message({ id: "backdated", timestamp: 200_000 })], + }; + const list = vi.fn(async (_inboxId: string, options: { pageToken?: string }) => + options.pageToken ? secondPage : firstPage, + ); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store, + now: () => 1_000_000, + }); + let capacityPause = true; + const receive = vi.fn(async (record: AgentMailIngressRecord) => { + if (record.messageId === "backdated" && capacityPause) { + capacityPause = false; + throw new AgentMailIngressCapacityError(); + } + }); + + await session.run({ receive, abortSignal: new AbortController().signal }); + await session.run({ receive, abortSignal: new AbortController().signal }); + + expect(list).toHaveBeenNthCalledWith( + 3, + "inbox_1", + expect.objectContaining({ + after: new Date(400_000 - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS), + pageToken: "page_2", + }), + expect.any(Object), + ); + expect(receive.mock.calls.map(([value]) => value.messageId)).toEqual([ + "newer", + "backdated", + "backdated", + ]); + }); + + it("resumes the last completed page after a mid-page abort", async () => { + const store = memoryStore(); + const key = sha256Hex("default\ninbox_1"); + await store.register(key, { + version: 1, + baselineAtMs: 100_000, + highWaterAtMs: 400_000, + established: true, + }); + const list = vi.fn(async (_inboxId: string, options: { pageToken?: string }) => + options.pageToken + ? { count: 1, messages: [message({ id: "page_2", timestamp: 600_000 })] } + : { + count: 1, + messages: [message({ id: "page_1", timestamp: 500_000 })], + nextPageToken: "page_2", + }, + ); + const session = await createAgentMailCatchUpSession({ + account, + client: { inboxes: { messages: { list } } } as never, + store, + now: () => 1_000_000, + }); + const firstController = new AbortController(); + const receive = vi.fn(async (record: AgentMailIngressRecord) => { + if (record.messageId === "page_2" && !firstController.signal.aborted) { + firstController.abort(); + } + }); + + await session.run({ receive, abortSignal: firstController.signal }); + await session.run({ receive, abortSignal: new AbortController().signal }); + + expect(list).toHaveBeenNthCalledWith( + 3, + "inbox_1", + expect.objectContaining({ pageToken: "page_2" }), + expect.any(Object), + ); + expect(receive.mock.calls.map(([value]) => value.messageId)).toEqual([ + "page_1", + "page_2", + "page_2", + ]); + }); }); diff --git a/src/channel/src/catch-up.ts b/src/channel/src/catch-up.ts index 4a78fc1..235437d 100644 --- a/src/channel/src/catch-up.ts +++ b/src/channel/src/catch-up.ts @@ -1,10 +1,11 @@ -import type { AgentMail, AgentMailClient } from "agentmail"; +import type { AgentMailClient } from "agentmail"; +import { randomUUID } from "node:crypto"; import type { PluginStateKeyedStore } from "openclaw/plugin-sdk/plugin-state-runtime"; import { type AgentMailLog, errorText } from "./log.js"; import { createAgentMailClient } from "./client.js"; import { sha256Hex } from "./digest.js"; import { - AGENTMAIL_DURABLE_COMPLETED_TTL_MS, + AGENTMAIL_DURABLE_PENDING_TTL_MS, AgentMailIngressCapacityError, } from "./durable-receive.js"; import { @@ -20,7 +21,9 @@ const CURSOR_VERSION = 1; const PAGE_LIMIT = 100; export const AGENTMAIL_REST_CATCH_UP_NAMESPACE = "agentmail.rest-catch-up"; -export const AGENTMAIL_REST_CATCH_UP_MAX_ACCOUNTS = 1_000; +export const AGENTMAIL_REST_CATCH_UP_LEGACY_MAX_ACCOUNTS = 1_000; +export const AGENTMAIL_REST_CATCH_UP_MAX_ENTRIES_PER_ACCOUNT = 1; +export const AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS = 90 * 24 * 60 * 60 * 1000; export const AGENTMAIL_REST_CATCH_UP_OVERLAP_MS = 5 * 60_000; // Periodic overlap covers half-open sockets and provider webhook gaps; the less frequent deep sweep // lists from the baseline so back-dated mail below the overlap window is still recovered. Both run @@ -28,11 +31,21 @@ export const AGENTMAIL_REST_CATCH_UP_OVERLAP_MS = 5 * 60_000; export const AGENTMAIL_CATCH_UP_INTERVAL_MS = 60_000; export const AGENTMAIL_DEEP_SWEEP_INTERVAL_MS = 15 * 60_000; +export type AgentMailCatchUpCheckpoint = { + afterAtMs: number; + beforeAtMs: number; + pageToken: string; + highWaterAtMs: number; + sinceBaseline: boolean; +}; + export type AgentMailCatchUpCursor = { version: typeof CURSOR_VERSION; baselineAtMs: number; highWaterAtMs: number; established: boolean; + generation?: string; + checkpoint?: AgentMailCatchUpCheckpoint; }; export type AgentMailCatchUpSession = { @@ -154,10 +167,41 @@ function cursorKey(account: ResolvedAgentMailAccount): string { return sha256Hex(`${account.accountId}\n${account.inboxId}`); } +function cursorNamespace(account: ResolvedAgentMailAccount): string { + // Reuse one bounded namespace per account across inbox rotations. persistCursor fences every + // write by key and claimed generation, so an evicted old inbox key can never be reinserted. + return `${AGENTMAIL_REST_CATCH_UP_NAMESPACE}.${sha256Hex(account.accountId)}`; +} + function validTimestamp(value: unknown): value is number { return typeof value === "number" && Number.isFinite(value) && value >= 0; } +function normalizeCheckpoint(value: unknown): AgentMailCatchUpCheckpoint | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const checkpoint = value as Partial; + if ( + !validTimestamp(checkpoint.afterAtMs) || + !validTimestamp(checkpoint.beforeAtMs) || + checkpoint.beforeAtMs < checkpoint.afterAtMs || + typeof checkpoint.pageToken !== "string" || + checkpoint.pageToken.length === 0 || + !validTimestamp(checkpoint.highWaterAtMs) || + typeof checkpoint.sinceBaseline !== "boolean" + ) { + return undefined; + } + return { + afterAtMs: checkpoint.afterAtMs, + beforeAtMs: checkpoint.beforeAtMs, + pageToken: checkpoint.pageToken, + highWaterAtMs: checkpoint.highWaterAtMs, + sinceBaseline: checkpoint.sinceBaseline, + }; +} + function normalizeCursor(value: unknown): AgentMailCatchUpCursor | null { if (!value || typeof value !== "object") { return null; @@ -171,11 +215,45 @@ function normalizeCursor(value: unknown): AgentMailCatchUpCursor | null { ) { return null; } + const checkpoint = normalizeCheckpoint(cursor.checkpoint); return { version: CURSOR_VERSION, baselineAtMs: cursor.baselineAtMs, highWaterAtMs: Math.max(cursor.baselineAtMs, cursor.highWaterAtMs), established: cursor.established, + ...(typeof cursor.generation === "string" && cursor.generation + ? { generation: cursor.generation } + : {}), + ...(checkpoint ? { checkpoint } : {}), + }; +} + +function mergeCursor( + currentValue: unknown, + next: Pick, + upperBoundAtMs: number, + generation: string, + checkpoint: AgentMailCatchUpCheckpoint | null, +): AgentMailCatchUpCursor { + const current = normalizeCursor(currentValue); + const baselineAtMs = Math.min( + upperBoundAtMs, + current?.baselineAtMs ?? next.baselineAtMs, + next.baselineAtMs, + ); + return { + version: CURSOR_VERSION, + baselineAtMs, + // A wall-clock rollback must be allowed to lower a cursor that is now in the future. Within + // the current clock horizon, concurrent updates still merge monotonically. + highWaterAtMs: Math.max( + baselineAtMs, + Math.min(upperBoundAtMs, current?.highWaterAtMs ?? 0), + Math.min(upperBoundAtMs, next.highWaterAtMs), + ), + established: current?.established === true || next.established, + generation, + ...(checkpoint ? { checkpoint } : {}), }; } @@ -184,41 +262,54 @@ async function persistCursor(params: { key: string; baselineAtMs: number; highWaterAtMs: number; - upperBoundAtMs: number; established: boolean; + upperBoundAtMs: number; + generation: string; + checkpoint: AgentMailCatchUpCheckpoint | null; +}): Promise { + const update = params.store.update; + if (!update) { + throw new Error("AgentMail catch-up requires atomic keyed-store updates"); + } + return await update( + params.key, + (currentValue) => { + const current = normalizeCursor(currentValue); + if (!current || current.generation !== params.generation) { + return undefined; + } + return mergeCursor( + current, + params, + params.upperBoundAtMs, + params.generation, + params.checkpoint, + ); + }, + { ttlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS }, + ); +} + +async function claimCursorGeneration(params: { + store: PluginStateKeyedStore; + key: string; + generation: string; }): Promise { const update = params.store.update; - if (update) { - await update(params.key, (currentValue) => { + if (!update) { + throw new Error("AgentMail catch-up requires atomic keyed-store updates"); + } + const claimed = await update( + params.key, + (currentValue) => { const current = normalizeCursor(currentValue); - return { - version: CURSOR_VERSION, - baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max( - current?.baselineAtMs ?? params.baselineAtMs, - Math.min( - params.upperBoundAtMs, - Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), - ), - ), - established: current?.established === true || params.established, - }; - }); - return; + return current ? { ...current, generation: params.generation } : undefined; + }, + { ttlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS }, + ); + if (!claimed) { + throw new Error("AgentMail catch-up cursor disappeared while claiming its generation"); } - const current = normalizeCursor(await params.store.lookup(params.key)); - await params.store.register(params.key, { - version: CURSOR_VERSION, - baselineAtMs: current?.baselineAtMs ?? params.baselineAtMs, - highWaterAtMs: Math.max( - current?.baselineAtMs ?? params.baselineAtMs, - Math.min( - params.upperBoundAtMs, - Math.max(current?.highWaterAtMs ?? 0, params.highWaterAtMs), - ), - ), - established: current?.established === true || params.established, - }); } export async function createAgentMailCatchUpSession(params: { @@ -229,14 +320,44 @@ export async function createAgentMailCatchUpSession(params: { log?: AgentMailLog; }): Promise { const now = params.now ?? Date.now; - const store = - params.store ?? - getAgentMailRuntime().state.openKeyedStore({ + const key = cursorKey(params.account); + let store = params.store; + if (!store) { + const state = getAgentMailRuntime().state; + const accountStore = state.openKeyedStore({ + // One entry per account bounds inbox-rotation state. Generation-fenced updates cannot upsert + // an evicted key, so overlapping old sessions cannot displace the replacement cursor. + namespace: cursorNamespace(params.account), + maxEntries: AGENTMAIL_REST_CATCH_UP_MAX_ENTRIES_PER_ACCOUNT, + overflowPolicy: "evict-oldest", + defaultTtlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS, + }); + store = accountStore; + // Upgrade migration probes the older shared namespace. Probe it even when the replacement + // cursor exists so transient cleanup failures remain retryable on later startups. + const legacyStore = state.openKeyedStore({ namespace: AGENTMAIL_REST_CATCH_UP_NAMESPACE, - maxEntries: AGENTMAIL_REST_CATCH_UP_MAX_ACCOUNTS, + maxEntries: AGENTMAIL_REST_CATCH_UP_LEGACY_MAX_ACCOUNTS, overflowPolicy: "reject-new", }); - const key = cursorKey(params.account); + const legacyCursor = normalizeCursor(await legacyStore.lookup(key)); + if (!normalizeCursor(await accountStore.lookup(key)) && legacyCursor) { + // Copy before creating a fresh baseline so mail received between shutdown and upgraded + // startup remains inside the recovery window. + await accountStore.registerIfAbsent(key, legacyCursor, { + ttlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS, + }); + } + if (legacyCursor && normalizeCursor(await accountStore.lookup(key))) { + try { + await legacyStore.delete(key); + } catch (error) { + params.log?.warn?.( + `AgentMail could not remove a migrated catch-up cursor: ${errorText(error)}`, + ); + } + } + } const initialAtMs = now(); const initialCursor: AgentMailCatchUpCursor = { version: CURSOR_VERSION, @@ -244,7 +365,11 @@ export async function createAgentMailCatchUpSession(params: { highWaterAtMs: initialAtMs, established: false, }; - await store.registerIfAbsent(key, initialCursor); + await store.registerIfAbsent(key, initialCursor, { + ttlMs: AGENTMAIL_REST_CATCH_UP_CURSOR_TTL_MS, + }); + const generation = randomUUID(); + await claimCursorGeneration({ store, key, generation }); const client = params.client ?? createAgentMailClient(params.account); return { @@ -253,26 +378,53 @@ 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: - // an idle inbox's high-water stops advancing, so without it the same pre-horizon messages - // would become eligible again after their tombstones expire. + if (storedCursor.generation !== generation) { + // A replacement session now owns this inbox generation. The old worker must stop before it + // can overwrite the replacement's cursor or continuation checkpoint. + return; + } + // Never scan below the durable recovery horizon on any sweep. Pending work and completed + // tombstones share this retention period, so catch-up can recover mail throughout the entire + // time the durable queue promises to retain it without re-admitting expired tombstones. + // Using the shorter historical completed-only horizon could skip never-admitted mail while + // ingress remained backpressured. // - // 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 = runAtMs - AGENTMAIL_DURABLE_COMPLETED_TTL_MS; + // The deep sweep lists from max(baseline, recoveryFloor): it recovers back-dated mail within + // the durable retention window and since monitoring began, but deliberately does not scan + // before the baseline, which would re-inject pre-monitoring history. + const runAtMs = now(); + const recoveryFloorMs = runAtMs - AGENTMAIL_DURABLE_PENDING_TTL_MS; + const requestedSinceBaseline = sinceBaseline === true; + const resumableCheckpoint = + storedCursor.checkpoint && + storedCursor.checkpoint.sinceBaseline === requestedSinceBaseline && + storedCursor.checkpoint.beforeAtMs <= runAtMs + 1 && + storedCursor.checkpoint.afterAtMs >= recoveryFloorMs + ? storedCursor.checkpoint + : undefined; + const sweepAtMs = resumableCheckpoint + ? resumableCheckpoint.beforeAtMs - 1 + : runAtMs; + // If the host clock moved backwards, both stored bounds can be in the future. Clamp the + // effective cursor for this run and persist the repaired value after a successful sweep. + const effectiveBaselineAtMs = Math.min(storedCursor.baselineAtMs, sweepAtMs); + const effectiveHighWaterAtMs = Math.max( + effectiveBaselineAtMs, + Math.min(storedCursor.highWaterAtMs, sweepAtMs), + ); + const beforeMs = sweepAtMs + 1; const baseAfterMs = - sinceBaseline || !storedCursor.established - ? storedCursor.baselineAtMs + requestedSinceBaseline || !storedCursor.established + ? effectiveBaselineAtMs : effectiveHighWaterAtMs - AGENTMAIL_REST_CATCH_UP_OVERLAP_MS; - const afterMs = Math.max(0, baseAfterMs, dedupeFloorMs); - let highWaterAtMs = effectiveHighWaterAtMs; - let pageCursor: string | undefined; + const afterMs = resumableCheckpoint + ? resumableCheckpoint.afterAtMs + : Math.max(0, baseAfterMs, recoveryFloorMs); + let highWaterAtMs = Math.max( + effectiveHighWaterAtMs, + Math.min(sweepAtMs, resumableCheckpoint?.highWaterAtMs ?? 0), + ); + let pageCursor = resumableCheckpoint?.pageToken; let admitted = 0; do { const page = await client.inboxes.messages.list( @@ -282,10 +434,9 @@ 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), + // Exclude implausibly future provider rows from normal pagination. The per-row clamp + // below remains defensive in case a provider projection violates this filter. + before: new Date(beforeMs), ascending: true, includeSpam: false, includeBlocked: false, @@ -294,7 +445,6 @@ export async function createAgentMailCatchUpSession(params: { }, { abortSignal }, ); - let pageAdvanced = false; for (const message of page.messages) { if (abortSignal.aborted) { return; @@ -302,19 +452,36 @@ export async function createAgentMailCatchUpSession(params: { 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; + const providerReceivedAt = resolveReceivedAgentMailMessageTimestampMs( + message, + params.account.inboxId, + ); + const arrivedAt = now(); + const receivedAt = providerReceivedAt !== null + ? Math.min(providerReceivedAt, sweepAtMs) + : sweepAtMs; + if (providerReceivedAt === null) { + // Use local arrival time so a malformed newest row is admitted once and advances the + // cursor instead of being re-listed and warned about forever. + params.log?.warn?.( + `AgentMail catch-up used arrival time for message ${message.messageId} with an invalid timestamp`, + ); + } else if (providerReceivedAt > sweepAtMs) { + // Provider clock skew must not push the cursor into the far future and disable periodic + // overlap recovery. Preserve the row while clamping its ordering timestamp to this + // sweep's fixed wall-clock bound. + params.log?.warn?.( + `AgentMail catch-up clamped future timestamp for message ${message.messageId}`, + ); + } try { await receive({ accountId: params.account.accountId, inboxId: params.account.inboxId, messageId: message.messageId, transport: "rest", - receivedAt: messageTimestampMs, - arrivedAt: now(), + receivedAt, + arrivedAt, }); } catch (error) { if (error instanceof AgentMailIngressCapacityError) { @@ -329,38 +496,51 @@ export async function createAgentMailCatchUpSession(params: { throw error; } admitted += 1; - highWaterAtMs = Math.max( - highWaterAtMs, - Math.min(messageTimestampMs, scanUpperBoundAtMs), - ); - pageAdvanced = true; + highWaterAtMs = Math.max(highWaterAtMs, receivedAt); } - if (pageAdvanced) { - // Persist once per page. If admission fails mid-page, the cursor stays behind the page - // and durable message-id dedupe safely absorbs the repeated prefix on the next pass. - await persistCursor({ + pageCursor = page.nextPageToken; + if (pageCursor) { + // Save a continuation only after every message on this page was durably admitted. The + // committed high-water mark remains unchanged until the entire sweep completes, so an + // out-of-order later page can never be skipped after capacity or a process restart. + const persisted = await persistCursor({ store, key, - baselineAtMs: storedCursor.baselineAtMs, - highWaterAtMs, - upperBoundAtMs: scanUpperBoundAtMs, - established: false, + baselineAtMs: effectiveBaselineAtMs, + highWaterAtMs: effectiveHighWaterAtMs, + established: storedCursor.established, + upperBoundAtMs: sweepAtMs, + generation, + checkpoint: { + afterAtMs: afterMs, + beforeAtMs: beforeMs, + pageToken: pageCursor, + highWaterAtMs, + sinceBaseline: requestedSinceBaseline, + }, }); + if (!persisted) { + return; + } } - pageCursor = page.nextPageToken; } while (pageCursor && !abortSignal.aborted); if (abortSignal.aborted) { return; } - await persistCursor({ + const persisted = await persistCursor({ store, key, - baselineAtMs: storedCursor.baselineAtMs, + baselineAtMs: effectiveBaselineAtMs, highWaterAtMs, - upperBoundAtMs: scanUpperBoundAtMs, established: true, + upperBoundAtMs: sweepAtMs, + generation, + checkpoint: null, }); + if (!persisted) { + return; + } if (admitted > 0) { params.log?.info?.( `AgentMail WebSocket catch-up admitted ${admitted} message${admitted === 1 ? "" : "s"} for account ${params.account.accountId}`, diff --git a/src/channel/src/config-schema.ts b/src/channel/src/config-schema.ts index 60193df..6ed70a6 100644 --- a/src/channel/src/config-schema.ts +++ b/src/channel/src/config-schema.ts @@ -11,6 +11,9 @@ import { z } from "openclaw/plugin-sdk/zod"; // documented upper bound the schema enforces. export const AGENTMAIL_MEDIA_DEFAULT_MB = 20; export const AGENTMAIL_MEDIA_MAX_MB = 100; +// One byte expressed in MiB. Preserve useful fractional limits (for example 0.5 MiB) while +// rejecting positive values that floor to a misleading one-byte runtime limit. +export const AGENTMAIL_MEDIA_MIN_MB = 1 / (1024 * 1024); const SecretInputSchema = buildSecretInputSchema(); @@ -26,7 +29,11 @@ const AgentMailAccountConfigSchema = z allowFrom: AllowFromListSchema, // Bounded so a typo cannot request an impractically large per-message buffer. 100 MiB comfortably // exceeds typical provider attachment limits; accounts.ts still floors this to finite integer bytes. - mediaMaxMb: z.number().positive().max(AGENTMAIL_MEDIA_MAX_MB).optional(), + mediaMaxMb: z + .number() + .min(AGENTMAIL_MEDIA_MIN_MB) + .max(AGENTMAIL_MEDIA_MAX_MB) + .optional(), }) .strict(); diff --git a/src/channel/src/durable-receive.test.ts b/src/channel/src/durable-receive.test.ts index 2452ff7..ec8dbef 100644 --- a/src/channel/src/durable-receive.test.ts +++ b/src/channel/src/durable-receive.test.ts @@ -3,6 +3,7 @@ import { AGENTMAIL_DURABLE_COMPLETED_TTL_MS, AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES, AGENTMAIL_DURABLE_PENDING_TTL_MS, + AGENTMAIL_DURABLE_RETENTION, createAgentMailDurableInboundId, withAgentMailIngressCapacity, } from "./durable-receive.js"; @@ -258,6 +259,68 @@ describe("AgentMail durable ingress", () => { expect(dispatch).toHaveBeenCalledTimes(2); }); + it("bounds persistent release failures instead of pinning the active dispatch forever", async () => { + const release = vi.fn(async () => { + throw new Error("database read-only"); + }); + const dispatch = vi.fn(async () => { + throw new Error("temporary hydration failure"); + }); + const error = vi.fn(); + await processAgentMailIngress({ + journal: { + accept: async () => ({ kind: "accepted", duplicate: false, record: {} }), + complete: vi.fn(), + release, + } as never, + record, + dispatch, + retryDelayMs: () => 0, + log: { error }, + }); + + await vi.waitFor(() => expect(error).toHaveBeenCalledWith(expect.stringContaining("release"))); + expect(release).toHaveBeenCalledTimes(50); + expect(dispatch).toHaveBeenCalledOnce(); + }); + + it("retries storage failures while releasing an abandoned deferred turn", async () => { + const release = vi + .fn<() => Promise>() + .mockRejectedValueOnce(new Error("database busy")) + .mockResolvedValueOnce(true); + const complete = vi.fn(async () => undefined); + 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(complete).toHaveBeenCalledOnce()); + expect(release).toHaveBeenCalledTimes(2); + expect(dispatch).toHaveBeenCalledTimes(2); + }); + it("drops a poison message after the dispatch ceiling instead of blocking the queue", async () => { const complete = vi.fn(async () => undefined); const release = vi.fn(async () => true); @@ -789,10 +852,11 @@ describe("AgentMail durable ingress", () => { expect(complete).toHaveBeenCalledWith(createAgentMailDurableInboundId(record)); }); - it("keeps WhatsApp-aligned retention values", () => { + it("keeps completed dedupe for the full recovery horizon without an entry cap", () => { expect(AGENTMAIL_DURABLE_PENDING_TTL_MS).toBe(30 * 24 * 60 * 60 * 1000); - expect(AGENTMAIL_DURABLE_COMPLETED_TTL_MS).toBe(7 * 24 * 60 * 60 * 1000); + expect(AGENTMAIL_DURABLE_COMPLETED_TTL_MS).toBe(AGENTMAIL_DURABLE_PENDING_TTL_MS); expect(AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES).toBe(450); + expect(AGENTMAIL_DURABLE_RETENTION).not.toHaveProperty("completedMaxEntries"); }); it("replays pending records after restart and completes them", async () => { diff --git a/src/channel/src/durable-receive.ts b/src/channel/src/durable-receive.ts index b5539c9..6ed092a 100644 --- a/src/channel/src/durable-receive.ts +++ b/src/channel/src/durable-receive.ts @@ -4,8 +4,18 @@ import { getAgentMailRuntime } from "./runtime.js"; import type { AgentMailIngressRecord } from "./types.js"; export const AGENTMAIL_DURABLE_PENDING_MAX_ENTRIES = 450; +const AGENTMAIL_DURABLE_PRUNE_EVERY_ACCEPTS = 100; export const AGENTMAIL_DURABLE_PENDING_TTL_MS = 30 * 24 * 60 * 60 * 1000; -export const AGENTMAIL_DURABLE_COMPLETED_TTL_MS = 7 * 24 * 60 * 60 * 1000; +// Keep completed tombstones for the full pending recovery horizon. REST catch-up may remain behind +// while durable admission is full; expiring dedupe markers sooner either replays completed mail or +// forces catch-up to skip never-admitted messages. +export const AGENTMAIL_DURABLE_COMPLETED_TTL_MS = AGENTMAIL_DURABLE_PENDING_TTL_MS; +export const AGENTMAIL_DURABLE_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, +} as const; /** * Raised when durable ingress is already holding the maximum number of pending rows. Transports @@ -106,27 +116,24 @@ export function createAgentMailDurableInboundReceiveJournal(params: { stateDir: runtime.state.resolveStateDir(), }, ); - 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] } : {}), - }); + const prune = async () => { + await queue.prune(AGENTMAIL_DURABLE_RETENTION); }; + // Admission is serialized by withAgentMailIngressCapacity below. Prune once at startup and then + // in bounded batches instead of scanning the queue around every individual message. + let acceptsSincePrune = AGENTMAIL_DURABLE_PRUNE_EVERY_ACCEPTS; // 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(); + if (acceptsSincePrune >= AGENTMAIL_DURABLE_PRUNE_EVERY_ACCEPTS) { + await prune(); + acceptsSincePrune = 0; + } const result = await queue.enqueue(id.trim(), payload, options); - await prune(id); + acceptsSincePrune += 1; if (result.kind === "accepted") { return { kind: "accepted", duplicate: false, record: result.record }; } @@ -154,18 +161,13 @@ export function createAgentMailDurableInboundReceiveJournal(params: { }, 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; + return await queue.release(id, options); }, deletePending: async (id) => await queue.delete(id), fail: async (id, options) => { - const failed = await queue.fail(id, options); - await prune(id); - return failed; + return await queue.fail(id, options); }, }; return withAgentMailIngressCapacity( diff --git a/src/channel/src/gateway.ts b/src/channel/src/gateway.ts index 29ddc8d..917265a 100644 --- a/src/channel/src/gateway.ts +++ b/src/channel/src/gateway.ts @@ -5,6 +5,7 @@ import type { AgentMailLog } from "./log.js"; import { collectAgentMailAccountIdWarnings, findConflictingAgentMailInboxOwner, + isAgentMailAccountConfigured, } from "./accounts.js"; import { createAgentMailCatchUpSession, @@ -71,7 +72,7 @@ export function collectAgentMailSecurityWarnings(account: ResolvedAgentMailAccou export function collectAgentMailStartupWarnings(account: ResolvedAgentMailAccount): string[] { const warnings: string[] = []; - if (!account.apiKey || !account.inboxId) { + if (!isAgentMailAccountConfigured(account)) { warnings.push("- AgentMail: apiKey and inboxId are required."); } warnings.push(...collectAgentMailSecurityWarnings(account)); @@ -95,7 +96,7 @@ export async function startAgentMailGatewayAccount(params: { for (const warning of warnings) { params.log?.warn?.(warning); } - if (!params.account.apiKey || !params.account.inboxId) { + if (!isAgentMailAccountConfigured(params.account)) { return await waitUntilAbort(params.abortSignal); } const inboxOwner = findConflictingAgentMailInboxOwner(params.cfg, params.account); diff --git a/src/channel/src/inbound.test.ts b/src/channel/src/inbound.test.ts index 690a038..985ab39 100644 --- a/src/channel/src/inbound.test.ts +++ b/src/channel/src/inbound.test.ts @@ -398,7 +398,7 @@ describe("AgentMail REST-authoritative inbound", () => { expect(rm).toHaveBeenCalledWith("/tmp/ignored.bin", { force: true }); }); - it("cleans up attachments when the adoption hook itself fails", async () => { + it("preserves adopted attachments when journal completion fails", async () => { rm.mockClear(); loadAgentMailInboundAttachments.mockResolvedValueOnce({ paths: ["/tmp/b.bin"], @@ -432,8 +432,9 @@ describe("AgentMail REST-authoritative inbound", () => { onTurnAdopted, }), ).rejects.toThrow("journal.complete failed"); - // The flag is set only after the hook resolves, so a failed completion still cleans up media. - expect(rm).toHaveBeenCalledWith("/tmp/b.bin", { force: true }); + // Core invoked the hook only after adoption, so its recovery state and tools may still reference + // this media even though persisting the ingress completion marker failed. + expect(rm).not.toHaveBeenCalled(); }); it("denies an unauthorized hydrated sender without dispatch", async () => { @@ -452,6 +453,33 @@ describe("AgentMail REST-authoritative inbound", () => { expect(run).not.toHaveBeenCalled(); }); + it.each([ + ["non-string sender", { from: ["sender@example.com"] }], + ["missing thread id", { threadId: undefined }], + ["empty thread id", { threadId: " " }], + ])("settles a hydrated message with %s", async (_name, overrides) => { + const run = vi.fn(); + const warn = vi.fn(); + await expect( + dispatchAgentMailInboundEvent({ + cfg: {}, + account, + record, + channelRuntime: { inbound: { run } } as never, + client: { + inboxes: { + messages: { + get: vi.fn(async () => message(overrides as Partial)), + }, + }, + } as never, + log: { warn }, + }), + ).resolves.toBeUndefined(); + expect(run).not.toHaveBeenCalled(); + expect(warn).toHaveBeenCalled(); + }); + it("settles permanently unsafe hydrated messages without dispatch", async () => { const run = vi.fn(); const warn = vi.fn(); diff --git a/src/channel/src/inbound.ts b/src/channel/src/inbound.ts index 839432e..a503f86 100644 --- a/src/channel/src/inbound.ts +++ b/src/channel/src/inbound.ts @@ -188,13 +188,20 @@ export async function dispatchAgentMailInboundEvent(params: { } throw new AgentMailLabelPendingError(params.record.messageId); } - const sender = parseSingleFromMailbox(message.from); + const sender = + typeof message.from === "string" ? parseSingleFromMailbox(message.from) : null; if (!sender) { params.log?.warn?.( `AgentMail rejected message ${message.messageId} with an ambiguous From mailbox`, ); return; } + if (typeof message.threadId !== "string" || !message.threadId.trim()) { + params.log?.warn?.( + `AgentMail rejected message ${message.messageId} with an invalid thread id`, + ); + return; + } // Authoritative sender authorization: a default-deny allowlist (dmPolicy defaults to "allowlist", // empty allowFrom denies everyone). The rejected-label check above is anti-spoofing defense in // depth (SPF/DKIM/DMARC failures land as "unauthenticated"); it never widens authorization. @@ -287,20 +294,13 @@ export async function dispatchAgentMailInboundEvent(params: { const turnAdoptionLifecycle = { admission: "exclusive" as const, onAdopted: async () => { + // Core has already adopted the turn when this observer runs. Transfer media ownership before + // persisting the ingress marker so a marker failure cannot delete files referenced by the + // adopted turn's recovery state or tools. 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; - } + turnAdopted = true; + detachAbortCleanup(); + await params.onTurnAdopted?.(); }, onDeferred: () => { turnDeferred = true; diff --git a/src/channel/src/ingress.ts b/src/channel/src/ingress.ts index 178ddcc..6378c94 100644 --- a/src/channel/src/ingress.ts +++ b/src/channel/src/ingress.ts @@ -44,6 +44,7 @@ export type AgentMailIngressDispatch = ( // tests exercise. const AGENTMAIL_MAX_DISPATCH_ATTEMPTS = 50; const AGENTMAIL_MAX_COMPLETION_ATTEMPTS = 50; +const AGENTMAIL_MAX_RELEASE_ATTEMPTS = 50; type DeferredOutcome = "adopted" | "abandoned" | "completion-failed" | "aborted"; @@ -116,6 +117,37 @@ function nextDispatchDelayMs(params: { return Math.min(base, remaining); } +async function releaseAgentMailIngressWithRetry(params: { + journal: AgentMailJournal; + id: string; + record: AgentMailIngressRecord; + lastError: string; + abortSignal?: AbortSignal; + retryDelay: (attempt: number) => number; + log?: AgentMailLog; +}): Promise<"released" | "gone" | "aborted" | "exhausted"> { + let attempts = 0; + while (!params.abortSignal?.aborted) { + try { + return (await params.journal.release(params.id, { lastError: params.lastError })) + ? "released" + : "gone"; + } catch (error) { + attempts += 1; + if (attempts >= AGENTMAIL_MAX_RELEASE_ATTEMPTS) { + params.log?.error?.( + `AgentMail could not release message ${params.record.messageId} after ${attempts} attempts: ${errorText(error)}`, + ); + return "exhausted"; + } + if (!(await waitForRetry(params.abortSignal, params.retryDelay(attempts)))) { + return "aborted"; + } + } + } + return "aborted"; +} + export async function processAgentMailIngress(params: { journal: AgentMailJournal; record: AgentMailIngressRecord; @@ -316,10 +348,21 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro } return true; } - const released = await params.journal.release(params.id, { lastError }); - if (!released) { + const releaseOutcome = await releaseAgentMailIngressWithRetry({ + journal: params.journal, + id: params.id, + record: params.record, + lastError, + abortSignal: params.abortSignal, + retryDelay: params.retryDelay ?? retryDelayMs, + log: params.log, + }); + if (releaseOutcome === "gone") { return true; } + if (releaseOutcome !== "released") { + return false; + } if ( !(await waitForRetry( params.abortSignal, @@ -354,27 +397,21 @@ async function dispatchAgentMailIngressUntilSettled(params: DispatchParams): Pro return true; } const lastError = errorText(dispatchError); - while (!params.abortSignal?.aborted) { - try { - const released = await params.journal.release(params.id, { lastError }); - if (!released) { - // A concurrent completion or retention prune means this worker no longer owns a - // pending row. Redispatching without ownership could duplicate an adopted turn. - return true; - } - break; - } catch { - if ( - !(await waitForRetry( - params.abortSignal, - (params.retryDelay ?? retryDelayMs)(dispatchAttempts), - )) - ) { - return false; - } - } + const releaseOutcome = await releaseAgentMailIngressWithRetry({ + journal: params.journal, + id: params.id, + record: params.record, + lastError, + abortSignal: params.abortSignal, + retryDelay: params.retryDelay ?? retryDelayMs, + log: params.log, + }); + if (releaseOutcome === "gone") { + // A concurrent completion or retention prune means this worker no longer owns a pending + // row. Redispatching without ownership could duplicate an adopted turn. + return true; } - if (params.abortSignal?.aborted) { + if (releaseOutcome !== "released") { return false; } const shouldRetry = await waitForRetry( diff --git a/src/channel/src/media.test.ts b/src/channel/src/media.test.ts index a68595e..d41c638 100644 --- a/src/channel/src/media.test.ts +++ b/src/channel/src/media.test.ts @@ -1,6 +1,7 @@ import { MediaFetchError } from "openclaw/plugin-sdk/media-runtime"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { + AGENTMAIL_INBOUND_MAX_ATTACHMENTS, AgentMailMediaPolicyError, loadAgentMailInboundAttachments, loadAgentMailOutboundAttachments, @@ -99,6 +100,23 @@ describe("AgentMail inbound attachments", () => { expect(getAttachment).toHaveBeenCalledOnce(); }); + it("rejects excessive attachment counts before downloading", async () => { + const getAttachment = vi.fn(); + await expect( + loadAgentMailInboundAttachments({ + client: { inboxes: { messages: { getAttachment } } } as never, + inboxId: "inbox_1", + messageId: "message_1", + attachments: Array.from( + { length: AGENTMAIL_INBOUND_MAX_ATTACHMENTS + 1 }, + (_, index) => ({ attachmentId: `attachment-${index}`, size: 0 }), + ), + maxBytes: 100, + }), + ).rejects.toThrow("attachment limit"); + expect(getAttachment).not.toHaveBeenCalled(); + }); + it.each([undefined, Number.NaN, -1, 1.5])( "rejects malformed declared attachment size %s before downloading", async (size) => { diff --git a/src/channel/src/media.ts b/src/channel/src/media.ts index df86ae5..655c687 100644 --- a/src/channel/src/media.ts +++ b/src/channel/src/media.ts @@ -17,6 +17,8 @@ export type AgentMailInboundMedia = { export class AgentMailMediaPolicyError extends Error {} +export const AGENTMAIL_INBOUND_MAX_ATTACHMENTS = 25; + function isAcceptedAttachment(attachment: AgentMail.Attachment): boolean { const disposition = attachment.contentDisposition?.toLocaleLowerCase("en-US"); // An explicit attachment disposition wins even when the part also has a Content-ID. A bare @@ -53,6 +55,11 @@ export async function loadAgentMailInboundAttachments(params: { maxBytes: number; }): Promise { const accepted = params.attachments.filter(isAcceptedAttachment); + if (accepted.length > AGENTMAIL_INBOUND_MAX_ATTACHMENTS) { + throw new AgentMailMediaPolicyError( + `AgentMail message exceeds the ${AGENTMAIL_INBOUND_MAX_ATTACHMENTS}-attachment limit`, + ); + } let declaredBytes = 0; for (const attachment of accepted) { const declaredSize: unknown = attachment.size; diff --git a/src/channel/src/send.test.ts b/src/channel/src/send.test.ts index 5deb2b7..c1e00ef 100644 --- a/src/channel/src/send.test.ts +++ b/src/channel/src/send.test.ts @@ -156,6 +156,46 @@ describe("AgentMail reply-only outbound", () => { expect(reply).not.toHaveBeenCalled(); }); + it("rejects a non-string hydrated From during unknown-send reconciliation", async () => { + reply.mockClear(); + const malformedClient = { + inboxes: { + messages: { + get: vi.fn(async (_inboxId: string, messageId: string) => ({ + ...(await get("inbox_1", messageId)), + from: ["sender@example.com"], + })), + reply, + }, + }, + } as never; + await expect( + 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: 9_000, + retryCount: 1, + effectiveReplyToId: "msg_1", + payloads: [{ text: "Hello" }], + }, + { client: malformedClient, now: () => 10_000 }, + ), + ).rejects.toThrow("recipient is not an authorized triggering sender"); + expect(reply).not.toHaveBeenCalled(); + }); + it("rejects explicit and proactive message targets", async () => { const base = { cfg: { diff --git a/src/channel/src/send.ts b/src/channel/src/send.ts index ad9d426..b65f5db 100644 --- a/src/channel/src/send.ts +++ b/src/channel/src/send.ts @@ -167,7 +167,10 @@ async function sendBoundAgentMailReply( ) { throw new Error("AgentMail reply target did not hydrate to the configured inbox and message."); } - const sender = parseSingleFromMailbox(triggeringMessage.from); + const sender = + typeof triggeringMessage.from === "string" + ? parseSingleFromMailbox(triggeringMessage.from) + : null; if ( !sender || !isAgentMailSenderAllowed({ diff --git a/src/channel/src/websocket.test.ts b/src/channel/src/websocket.test.ts index f541179..ba439d6 100644 --- a/src/channel/src/websocket.test.ts +++ b/src/channel/src/websocket.test.ts @@ -95,6 +95,76 @@ describe("AgentMail WebSocket ingress", () => { expect(close).toHaveBeenCalledTimes(2); }); + it("does not reset reconnect backoff for open-close flaps", async () => { + handlers.clear(); + connect.mockClear(); + const reconnectDelay = vi.fn(() => 0); + let nowMs = 1_000; + const controller = new AbortController(); + const running = startAgentMailWebSocket({ + account, + abortSignal: controller.signal, + receive: vi.fn(async () => undefined), + catchUpSession: { run: catchUpRun }, + reconnectDelayMs: reconnectDelay, + now: () => nowMs, + }); + + await vi.waitFor(() => expect(handlers.has("open")).toBe(true)); + handlers.get("open")?.(); + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); + handlers.get("open")?.(); + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(3)); + expect(reconnectDelay.mock.calls.map(([attempt]) => attempt)).toEqual([1, 2]); + + handlers.get("open")?.(); + nowMs += 30_000; + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(4)); + expect(reconnectDelay.mock.calls.map(([attempt]) => attempt)).toEqual([1, 2, 1]); + + controller.abort(); + await running; + }); + + it("resets reconnect backoff when a short-lived connection delivers an event", async () => { + handlers.clear(); + connect.mockClear(); + const reconnectDelay = vi.fn(() => 0); + const controller = new AbortController(); + const running = startAgentMailWebSocket({ + account, + abortSignal: controller.signal, + receive: vi.fn(async () => undefined), + catchUpSession: { run: catchUpRun }, + reconnectDelayMs: reconnectDelay, + }); + + await vi.waitFor(() => expect(handlers.has("open")).toBe(true)); + handlers.get("open")?.(); + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(2)); + handlers.get("open")?.(); + handlers.get("message")?.({ + type: "event", + eventType: "message.received", + message: { + inboxId: "inbox_1", + messageId: "message_productive_socket", + labels: ["received"], + timestamp: new Date(1_234), + }, + }); + handlers.get("close")?.(); + await vi.waitFor(() => expect(connect).toHaveBeenCalledTimes(3)); + + expect(reconnectDelay.mock.calls.map(([attempt]) => attempt)).toEqual([1, 1]); + controller.abort(); + await running; + }); + it("stops cleanly when aborted before the initial socket opens", async () => { handlers.clear(); close.mockClear(); diff --git a/src/channel/src/websocket.ts b/src/channel/src/websocket.ts index 44ae86e..507152a 100644 --- a/src/channel/src/websocket.ts +++ b/src/channel/src/websocket.ts @@ -17,6 +17,7 @@ import { import type { AgentMailIngressRecord, ResolvedAgentMailAccount } from "./types.js"; const AGENTMAIL_WEBSOCKET_LIVE_QUEUE_MAX = 32; +const AGENTMAIL_WEBSOCKET_STABLE_CONNECTION_MS = 30_000; // A single record must not pin the one bounded live worker forever. After this many failed durable // admissions, hand the record to REST catch-up (which retains the provider-side source) so later // live events keep advancing. @@ -98,6 +99,7 @@ export async function startAgentMailWebSocket(params: { catchUpIntervalMs?: number; deepSweepIntervalMs?: number; client?: AgentMailClient; + now?: () => number; }): Promise { const client = params.client ?? createAgentMailClient(params.account); const catchUpSession = @@ -110,6 +112,7 @@ export async function startAgentMailWebSocket(params: { const retryDelay = params.retryDelayMs ?? websocketRetryDelayMs; const reconnectDelay = params.reconnectDelayMs ?? retryDelay; const liveQueueMax = params.liveQueueMax ?? AGENTMAIL_WEBSOCKET_LIVE_QUEUE_MAX; + const now = params.now ?? Date.now; const liveQueue: AgentMailIngressRecord[] = []; const queuedMessageIds = new Set(); let liveWorker: Promise | undefined; @@ -161,13 +164,13 @@ export async function startAgentMailWebSocket(params: { }); }; - const handleMessage = (event: unknown) => { + const handleMessage = (event: unknown): boolean => { if (!isReceivedEvent(event)) { - return; + return false; } if (!agentMailInboxIdsEqual(event.message.inboxId, params.account.inboxId)) { params.log?.warn?.("AgentMail WebSocket ignored an event for the wrong inbox"); - return; + return false; } // The event type itself is authoritative for live receipt. Provider label projection can lag // the WebSocket frame; durable hydration already retries that condition safely. @@ -175,17 +178,17 @@ export async function startAgentMailWebSocket(params: { if (messageTimestampMs === null) { params.log?.warn?.("AgentMail WebSocket received an event with an invalid timestamp"); catchUpSupervisor.request(); - return; + return false; } if (queuedMessageIds.has(event.message.messageId)) { - return; + return true; } if (queuedMessageIds.size >= liveQueueMax) { // Keep the process-local backlog bounded. REST catch-up remains the authoritative recovery // source for events dropped while durable admission is backpressured. params.log?.warn?.("AgentMail WebSocket live admission is full; scheduling REST catch-up"); catchUpSupervisor.request(); - return; + return true; } queuedMessageIds.add(event.message.messageId); liveQueue.push({ @@ -197,6 +200,7 @@ export async function startAgentMailWebSocket(params: { arrivedAt: Date.now(), }); runLiveWorker(); + return true; }; // Own reconnection. The pinned agentmail@0.5.16 can synthesize a normal close, disable its @@ -232,6 +236,7 @@ export async function startAgentMailWebSocket(params: { continue; } let subscribedForCurrentConnection = false; + let subscribedAtMs: number | undefined; const subscribe = () => { if (subscribedForCurrentConnection) { return; @@ -242,6 +247,7 @@ export async function startAgentMailWebSocket(params: { eventTypes: ["message.received"], }); subscribedForCurrentConnection = true; + subscribedAtMs = now(); params.log?.info?.( `AgentMail WebSocket subscribed for account ${params.account.accountId}`, ); @@ -260,7 +266,6 @@ export async function startAgentMailWebSocket(params: { resolve(); }; socket.on("open", () => { - reconnectAttempt = 0; subscribe(); }); socket.on("close", settleClosed); @@ -275,7 +280,13 @@ export async function startAgentMailWebSocket(params: { // for this connection so the outer loop recreates it. settleClosed(); }); - socket.on("message", handleMessage); + socket.on("message", (event) => { + if (handleMessage(event)) { + // A valid event proves the subscription is productive even when an intermediary + // recycles it before the 30-second stability threshold. + reconnectAttempt = 0; + } + }); }); // waitForOpen() does not settle on an aborted initial connection; close the already-open race // from readyState instead. @@ -287,6 +298,15 @@ export async function startAgentMailWebSocket(params: { if (params.abortSignal.aborted) { return; } + // An `open` event alone does not prove a healthy connection: resetting there turns repeated + // open/close flaps into a zero-attempt reconnect loop. Reset only after the socket remained + // subscribed for a meaningful interval. + if ( + subscribedAtMs !== undefined && + now() - subscribedAtMs >= AGENTMAIL_WEBSOCKET_STABLE_CONNECTION_MS + ) { + reconnectAttempt = 0; + } // Unexpected close: reconnect after a bounded backoff. REST catch-up covers the gap. params.log?.warn?.( `AgentMail WebSocket closed for account ${params.account.accountId}; reconnecting`, diff --git a/src/cli/agentmail-cli-release.json b/src/cli/agentmail-cli-release.json new file mode 100644 index 0000000..cd2b4ec --- /dev/null +++ b/src/cli/agentmail-cli-release.json @@ -0,0 +1,55 @@ +{ + "version": "0.7.14", + "repository": "agentmail-to/agentmail-cli", + "vendorDirectory": "vendor/agentmail", + "assets": { + "darwin-arm64": { + "archive": "agentmail_0.7.14_macos_arm64.zip", + "sha256": "3e4bdd699826dfbe043b3c1b3da330d7ab09aee81498fd194c2448cf8002da59", + "executableSha256": "5ed06a8071fa3269f7b45726a6c11d8bc09a6fe5c7fa6c4d94760494be8c9bb9", + "executableName": "agentmail" + }, + "darwin-x64": { + "archive": "agentmail_0.7.14_macos_amd64.zip", + "sha256": "2ad6a3d0e03441997c78424e03516b314d464933ebd5a882aeed31e5a6e0bf0c", + "executableSha256": "0a09b38f270123299bcbd30fde1faf9d28ecd4a9e9e820c6707c11d9b22c1574", + "executableName": "agentmail" + }, + "linux-arm64": { + "archive": "agentmail_0.7.14_linux_arm64.tar.gz", + "sha256": "3aaaa675793c6c182ce5877205e7211ed490ec92fcfef1da8ca555dd402d4a6e", + "executableSha256": "0793dc1a7913addc893e8aeb5717e2576dcdb2f12438299ee853756f78b1f280", + "executableName": "agentmail" + }, + "linux-ia32": { + "archive": "agentmail_0.7.14_linux_386.tar.gz", + "sha256": "d4ef844be2993363985388d59adfe6b3534ae071a879eafdd2978add795b2675", + "executableSha256": "22a5810bb85c6eff2cf69d00e569f50a64ee3c66fd1c9c522974881e70fe2259", + "executableName": "agentmail" + }, + "linux-x64": { + "archive": "agentmail_0.7.14_linux_amd64.tar.gz", + "sha256": "8b5304be5b29d2457234a7a3655eba87ff2b89e5c57d01a51788fdafe63b2b65", + "executableSha256": "b90acaaf52a3f5b77c8812e6092db1dfc66148b39300bacc84c74635cf991bd5", + "executableName": "agentmail" + }, + "win32-arm64": { + "archive": "agentmail_0.7.14_windows_arm64.zip", + "sha256": "5d06ccf7235d4c11cdf19f3916f9db79e6cd926757a6110ddac1603fe3f09260", + "executableSha256": "2c23a9576303a9e50a6b5ca8209348a594ef9b33d4f3ded0483abf967516d7c3", + "executableName": "agentmail.exe" + }, + "win32-ia32": { + "archive": "agentmail_0.7.14_windows_386.zip", + "sha256": "c313faede4952bf28bc6a1a8ba381845b376c8136ecf0a0e459ab092f101a088", + "executableSha256": "326e24592efe5feb8436a4aa72b75ace01d43432aec1d267700c0219efd20077", + "executableName": "agentmail.exe" + }, + "win32-x64": { + "archive": "agentmail_0.7.14_windows_amd64.zip", + "sha256": "1beabaa0c855b5f1091bdc3455dcce13141cf52a37d1d93644b000874af83d4d", + "executableSha256": "ad1b1c8d4b547dfbb10149687da02c07ba78b0660b200b413c08f55eeea5b15a", + "executableName": "agentmail.exe" + } + } +} diff --git a/src/cli/config.ts b/src/cli/config.ts new file mode 100644 index 0000000..bef998e --- /dev/null +++ b/src/cli/config.ts @@ -0,0 +1,46 @@ +import { buildJsonPluginConfigSchema } from "openclaw/plugin-sdk/plugin-entry"; + +export type AgentMailCliConfig = { + baseUrl?: string; +}; + +export const agentMailCliConfigJsonSchema = { + type: "object", + properties: { + baseUrl: { + type: "string", + description: "Optional AgentMail API base URL override for the bundled CLI.", + format: "uri", + }, + timeoutSeconds: { + type: "integer", + description: + "Deprecated tool setting retained for upgrade compatibility; the bundled CLI ignores it.", + minimum: 1, + maximum: 300, + }, + maxRetries: { + type: "integer", + description: + "Deprecated tool setting retained for upgrade compatibility; the bundled CLI ignores it.", + minimum: 0, + maximum: 10, + }, + }, + additionalProperties: false, +}; + +export const agentMailCliConfigSchema = buildJsonPluginConfigSchema( + agentMailCliConfigJsonSchema as Parameters[0], +); + +export function parseAgentMailCliConfig(value: unknown): AgentMailCliConfig { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return {}; + } + + const baseUrl = (value as Record).baseUrl; + return typeof baseUrl === "string" && baseUrl.trim() + ? { baseUrl: baseUrl.trim() } + : {}; +} diff --git a/src/cli/index.test.ts b/src/cli/index.test.ts new file mode 100644 index 0000000..420104c --- /dev/null +++ b/src/cli/index.test.ts @@ -0,0 +1,207 @@ +import { describe, expect, it, vi } from "vitest"; +import { createAgentMailCliPlugin } from "./index.js"; +import { + agentMailCliSignalExitCode, + resolveAgentMailCliTarget, + sanitizeAgentMailCliEnvironment, + withConfiguredBaseUrl, +} from "./runner.js"; + +describe("AgentMail CLI bridge", () => { + it("maps supported Node targets to packaged executable directories", () => { + expect(resolveAgentMailCliTarget("linux", "x64")).toEqual({ + directory: "linux-x64", + executableName: "agentmail", + }); + expect(resolveAgentMailCliTarget("darwin", "arm64")).toEqual({ + directory: "darwin-arm64", + executableName: "agentmail", + }); + expect(resolveAgentMailCliTarget("win32", "x64")).toEqual({ + directory: "win32-x64", + executableName: "agentmail.exe", + }); + expect(() => resolveAgentMailCliTarget("darwin", "ia32")).toThrow( + "does not support darwin/ia32", + ); + }); + + it("allows only the operator-configured API base URL", () => { + expect(withConfiguredBaseUrl(["inboxes", "list"], "https://example.test/v0")).toEqual([ + "--base-url", + "https://example.test/v0", + "inboxes", + "list", + ]); + expect(withConfiguredBaseUrl(["inboxes", "list"], undefined)).toEqual([ + "inboxes", + "list", + ]); + expect(() => + withConfiguredBaseUrl( + ["--base-url=https://override.test/v0", "inboxes", "list"], + "https://example.test/v0", + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["inboxes", "list", "--base-url", "https://override.test/v0"], + undefined, + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["--environment", "development", "inboxes", "list"], + undefined, + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["--environment=development", "inboxes", "list"], + "https://example.test/v0", + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["-base-url", "https://override.test/v0", "inboxes", "list"], + undefined, + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["-environment=development", "inboxes", "list"], + undefined, + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["---base-url=https://override.test/v0", "inboxes", "list"], + undefined, + ), + ).toThrow("endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["--api-key", "attacker-controlled", "inboxes", "list"], + undefined, + ), + ).toThrow("credential and endpoint overrides are restricted"); + expect(() => + withConfiguredBaseUrl( + ["-api-key=attacker-controlled", "inboxes", "list"], + undefined, + ), + ).toThrow("credential and endpoint overrides are restricted"); + expect( + withConfiguredBaseUrl( + ["--transform", "--base-url=literal-output", "inboxes", "list"], + undefined, + ), + ).toEqual(["--transform", "--base-url=literal-output", "inboxes", "list"]); + expect( + withConfiguredBaseUrl( + ["inboxes:messages", "send", "--subject", "--base-url=is restricted"], + undefined, + ), + ).toEqual([ + "inboxes:messages", + "send", + "--subject", + "--base-url=is restricted", + ]); + }); + + it("removes inherited endpoint selectors while preserving credentials", () => { + expect( + sanitizeAgentMailCliEnvironment({ + AGENTMAIL_API_KEY: "am_test", + AGENTMAIL_BASE_URL: "https://attacker.example", + agentmail_environment: "development", + HTTPS_PROXY: "https://attacker.example", + no_proxy: "api.agentmail.to", + OTHER_VALUE: "kept", + }), + ).toEqual({ + AGENTMAIL_API_KEY: "am_test", + OTHER_VALUE: "kept", + }); + }); + + it("maps terminating signals to conventional shell exit codes", () => { + expect(agentMailCliSignalExitCode("SIGINT")).toBe(130); + expect(agentMailCliSignalExitCode("SIGTERM")).toBe(143); + }); + + it("registers one passthrough command and invokes the bundled CLI runner", async () => { + const runCli = vi.fn().mockResolvedValue(0); + const plugin = createAgentMailCliPlugin(runCli); + let registrar: + | ((context: { program: FakeProgram }) => void | Promise) + | undefined; + let registrationOptions: unknown; + let action: ((args: string[]) => Promise) | undefined; + + const command = { + description: vi.fn().mockReturnThis(), + helpOption: vi.fn().mockReturnThis(), + allowUnknownOption: vi.fn().mockReturnThis(), + allowExcessArguments: vi.fn().mockReturnThis(), + action: vi.fn((handler: (args: string[]) => Promise) => { + action = handler; + return command; + }), + }; + type FakeProgram = { + command: (definition: string) => typeof command; + }; + const program: FakeProgram = { + command: vi.fn().mockReturnValue(command), + }; + + (plugin.register as unknown as (api: { + pluginConfig: Record; + registerCli: ( + callback: (context: { program: FakeProgram }) => void | Promise, + options: unknown, + ) => void; + }) => void)({ + pluginConfig: { baseUrl: "https://example.test/v0" }, + registerCli(callback, options) { + registrar = callback; + registrationOptions = options; + }, + }); + + expect(registrationOptions).toEqual({ + commands: ["agentmail"], + descriptors: [ + { + name: "agentmail", + description: "Run the bundled AgentMail CLI", + hasSubcommands: false, + }, + ], + }); + + await registrar?.({ program }); + expect(program.command).toHaveBeenCalledWith("agentmail [arguments...]"); + + await action?.(["inboxes", "list", "--format", "json"]); + expect(runCli).toHaveBeenCalledWith([ + "--base-url", + "https://example.test/v0", + "inboxes", + "list", + "--format", + "json", + ]); + + const priorExitCode = process.exitCode; + const error = vi.spyOn(console, "error").mockImplementation(() => undefined); + runCli.mockRejectedValueOnce(new Error("bundled CLI could not start")); + await action?.(["inboxes", "list"]); + expect(error).toHaveBeenCalledWith("bundled CLI could not start"); + expect(process.exitCode).toBe(1); + process.exitCode = priorExitCode; + error.mockRestore(); + }); +}); diff --git a/src/cli/index.ts b/src/cli/index.ts new file mode 100644 index 0000000..116f6f1 --- /dev/null +++ b/src/cli/index.ts @@ -0,0 +1,64 @@ +import { + definePluginEntry, + type OpenClawPluginDefinition, +} from "openclaw/plugin-sdk/plugin-entry"; +import { + agentMailCliConfigSchema, + parseAgentMailCliConfig, +} from "./config.js"; +import { + runAgentMailCli, + withConfiguredBaseUrl, +} from "./runner.js"; + +type RunAgentMailCli = typeof runAgentMailCli; + +export function createAgentMailCliPlugin( + runCli: RunAgentMailCli = runAgentMailCli, +): OpenClawPluginDefinition { + return definePluginEntry({ + id: "agentmail", + name: "AgentMail", + description: "Run the bundled AgentMail CLI through OpenClaw.", + configSchema: agentMailCliConfigSchema, + register(api) { + const config = parseAgentMailCliConfig(api.pluginConfig); + + api.registerCli( + ({ program }) => { + program + .command("agentmail [arguments...]") + .description("Run the bundled AgentMail CLI (use -- before AgentMail arguments)") + .helpOption(false) + .allowUnknownOption(true) + .allowExcessArguments(true) + .action(async (args: string[]) => { + try { + const exitCode = await runCli(withConfiguredBaseUrl(args, config.baseUrl)); + if (exitCode !== 0) { + process.exitCode = exitCode; + } + } catch (error) { + console.error(error instanceof Error ? error.message : String(error)); + process.exitCode = 1; + } + }); + }, + { + commands: ["agentmail"], + descriptors: [ + { + name: "agentmail", + description: "Run the bundled AgentMail CLI", + hasSubcommands: false, + }, + ], + }, + ); + }, + }); +} + +const agentMailCliPlugin: OpenClawPluginDefinition = createAgentMailCliPlugin(); + +export default agentMailCliPlugin; diff --git a/src/cli/runner.ts b/src/cli/runner.ts new file mode 100644 index 0000000..9b2832a --- /dev/null +++ b/src/cli/runner.ts @@ -0,0 +1,164 @@ +import { spawn } from "node:child_process"; +import { accessSync, constants } from "node:fs"; +import { constants as osConstants } from "node:os"; +import { fileURLToPath } from "node:url"; +import release from "./agentmail-cli-release.json" with { type: "json" }; + +export type AgentMailCliTarget = { + directory: string; + executableName: "agentmail" | "agentmail.exe"; +}; + +export const AGENTMAIL_CLI_ENDPOINT_OPTIONS = ["base-url", "environment"] as const; +export const AGENTMAIL_CLI_RESTRICTED_OPTIONS = [ + "api-key", + ...AGENTMAIL_CLI_ENDPOINT_OPTIONS, +] as const; + +const restrictedOptions = new Set(AGENTMAIL_CLI_RESTRICTED_OPTIONS); +const optionsWithSeparateValues = new Set([ + "format", + "format-error", + "transform", + "transform-error", + "subject", +]); +const restrictedEnvironmentKeys = new Set([ + "AGENTMAIL_BASE_URL", + "AGENTMAIL_ENVIRONMENT", + "ALL_PROXY", + "HTTP_PROXY", + "HTTPS_PROXY", + "NO_PROXY", +]); + +function parseOptionToken(token: string): { + name: string; + hasInlineValue: boolean; +} | null { + const option = /^-+(.+)$/.exec(token)?.[1]; + if (!option) { + return null; + } + const equalsAt = option.indexOf("="); + return { + name: (equalsAt < 0 ? option : option.slice(0, equalsAt)).toLowerCase(), + hasInlineValue: equalsAt >= 0, + }; +} + +export function resolveAgentMailCliVendorPath(...segments: string[]): string { + return fileURLToPath( + new URL(`../../${release.vendorDirectory}/${segments.join("/")}`, import.meta.url), + ); +} + +export function resolveAgentMailCliTarget( + platform: NodeJS.Platform = process.platform, + arch: NodeJS.Architecture = process.arch, +): AgentMailCliTarget { + const directory = `${platform}-${arch}`; + const asset = release.assets[directory as keyof typeof release.assets]; + if (!asset) { + throw new Error(`The bundled AgentMail CLI does not support ${platform}/${arch}.`); + } + if (asset.executableName !== "agentmail" && asset.executableName !== "agentmail.exe") { + throw new Error(`The bundled AgentMail CLI metadata is invalid for ${platform}/${arch}.`); + } + + return { + directory, + executableName: asset.executableName, + }; +} + +export function resolveAgentMailCliExecutable( + platform: NodeJS.Platform = process.platform, + arch: NodeJS.Architecture = process.arch, +): string { + const target = resolveAgentMailCliTarget(platform, arch); + return resolveAgentMailCliVendorPath(target.directory, target.executableName); +} + +export function withConfiguredBaseUrl( + args: readonly string[], + baseUrl: string | undefined, +): string[] { + let consumesNextValue = false; + for (const arg of args) { + if (consumesNextValue) { + consumesNextValue = false; + continue; + } + const option = parseOptionToken(arg); + if (!option) { + continue; + } + if (restrictedOptions.has(option.name)) { + throw new Error( + "AgentMail API credential and endpoint overrides are restricted to operator-controlled configuration.", + ); + } + consumesNextValue = !option.hasInlineValue && optionsWithSeparateValues.has(option.name); + } + return baseUrl ? ["--base-url", baseUrl, ...args] : [...args]; +} + +export function sanitizeAgentMailCliEnvironment( + source: NodeJS.ProcessEnv, +): NodeJS.ProcessEnv { + const sanitized = { ...source }; + for (const key of Object.keys(sanitized)) { + if (restrictedEnvironmentKeys.has(key.toUpperCase())) { + delete sanitized[key]; + } + } + return sanitized; +} + +export function agentMailCliSignalExitCode(signal: NodeJS.Signals): number { + const signalNumber = osConstants.signals[signal]; + return typeof signalNumber === "number" ? 128 + signalNumber : 1; +} + +export async function runAgentMailCli( + args: readonly string[], + options: { + executable?: string; + env?: NodeJS.ProcessEnv; + } = {}, +): Promise { + const executable = options.executable ?? resolveAgentMailCliExecutable(); + + try { + accessSync(executable, process.platform === "win32" ? constants.F_OK : constants.X_OK); + } catch { + throw new Error( + `The bundled AgentMail CLI executable is missing or not executable at ${executable}. ` + + "Reinstall the AgentMail plugin from its published package.", + ); + } + + return await new Promise((resolve, reject) => { + const child = spawn(executable, [...args], { + env: sanitizeAgentMailCliEnvironment(options.env ?? process.env), + stdio: "inherit", + windowsHide: false, + }); + + child.once("error", (error) => { + reject( + new Error(`Failed to start the bundled AgentMail CLI: ${error.message}`, { + cause: error, + }), + ); + }); + child.once("exit", (code, signal) => { + if (signal) { + resolve(agentMailCliSignalExitCode(signal)); + return; + } + resolve(code ?? 1); + }); + }); +} diff --git a/src/index.test.ts b/src/index.test.ts new file mode 100644 index 0000000..1d45d42 --- /dev/null +++ b/src/index.test.ts @@ -0,0 +1,33 @@ +import { describe, expect, it, vi } from "vitest"; +import entry from "./index.js"; + +function registrationApi(mode: "full" | "discovery" | "tool-discovery" | "cli-metadata") { + return { + registrationMode: mode, + pluginConfig: {}, + runtime: {}, + registerChannel: vi.fn(), + registerCli: vi.fn(), + }; +} + +describe("combined AgentMail plugin entry", () => { + it("registers the channel and CLI from one full runtime", () => { + const api = registrationApi("full"); + entry.register(api as never); + expect(api.registerChannel).toHaveBeenCalledOnce(); + expect(api.registerCli).toHaveBeenCalledOnce(); + }); + + it("keeps channel work out of CLI metadata and tool discovery", () => { + const cliApi = registrationApi("cli-metadata"); + entry.register(cliApi as never); + expect(cliApi.registerChannel).not.toHaveBeenCalled(); + expect(cliApi.registerCli).toHaveBeenCalledOnce(); + + const toolApi = registrationApi("tool-discovery"); + entry.register(toolApi as never); + expect(toolApi.registerChannel).not.toHaveBeenCalled(); + expect(toolApi.registerCli).not.toHaveBeenCalled(); + }); +}); diff --git a/src/index.ts b/src/index.ts index be77551..d43b339 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,31 +2,33 @@ 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"; +import { agentMailPlugin } from "./channel/channel-plugin-api.js"; +import cliEntry from "./cli/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. +// One runtime entry must own both surfaces. Multiple `openclaw.extensions` entries form a plugin +// pack and receive distinct runtime identities; the host can then select the CLI-only runtime at +// gateway startup and never start the channel worker. 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, + "AgentMail for OpenClaw: a CLI-backed skill plus a durable, allowlisted, reply-only email channel.", + configSchema: cliEntry.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); + if (api.registrationMode !== "cli-metadata") { + api.registerChannel({ plugin: agentMailPlugin as never }); + setAgentMailRuntime(api.runtime); + } + if ( + api.registrationMode === "full" || + api.registrationMode === "discovery" || + api.registrationMode === "cli-metadata" + ) { + cliEntry.register?.(api); } }, }); diff --git a/src/tools/client.ts b/src/tools/client.ts deleted file mode 100644 index d3851bc..0000000 --- a/src/tools/client.ts +++ /dev/null @@ -1,74 +0,0 @@ -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( - { - baseUrl: Type.Optional( - Type.String({ - description: "Optional AgentMail API base URL override.", - format: "uri", - }), - ), - timeoutSeconds: Type.Optional( - Type.Integer({ - description: "Maximum time to wait for an AgentMail API request.", - minimum: 1, - maximum: 300, - }), - ), - maxRetries: Type.Optional( - Type.Integer({ - description: "Number of times the AgentMail SDK retries a request.", - minimum: 0, - maximum: 10, - }), - ), - }, - { additionalProperties: false }, -); - -export type AgentMailConfig = Static; - -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. Configure channels.agentmail.apiKey or set AGENTMAIL_API_KEY and restart OpenClaw.", - ); - } - - return new AgentMailClient({ - apiKey, - ...(config.baseUrl ? { baseUrl: config.baseUrl } : {}), - ...(config.timeoutSeconds ? { timeoutInSeconds: config.timeoutSeconds } : {}), - ...(config.maxRetries !== undefined ? { maxRetries: config.maxRetries } : {}), - }); -} - -export function requestOptions(signal?: AbortSignal): { abortSignal?: AbortSignal } { - return signal ? { abortSignal: signal } : {}; -} diff --git a/src/tools/index.test.ts b/src/tools/index.test.ts deleted file mode 100644 index a77295c..0000000 --- a/src/tools/index.test.ts +++ /dev/null @@ -1,243 +0,0 @@ -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { getToolPluginMetadata } from "openclaw/plugin-sdk/tool-plugin"; - -const sdk = vi.hoisted(() => { - const messages = { - list: vi.fn(), - search: vi.fn(), - get: vi.fn(), - send: vi.fn(), - reply: vi.fn(), - forward: vi.fn(), - update: vi.fn(), - }; - const inboxes = { - list: vi.fn(), - create: vi.fn(), - messages, - }; - - return { - constructor: vi.fn(), - inboxes, - messages, - }; -}); - -vi.mock("agentmail", () => ({ - AgentMailClient: class MockAgentMailClient { - readonly inboxes = sdk.inboxes; - - constructor(options?: unknown) { - sdk.constructor(options); - } - }, -})); - -import entry from "./index.js"; - -type RegisteredTool = { - name: string; - execute: ( - toolCallId: string, - params: Record, - signal?: AbortSignal, - ) => Promise; -}; - -function registerTools( - pluginConfig: Record = {}, - hostConfig: Record = {}, -): RegisteredTool[] { - const registered: RegisteredTool[] = []; - const register = entry.register as unknown as (api: { - pluginConfig: Record; - registerTool: (tool: RegisteredTool) => void; - }) => void; - - register({ - pluginConfig, - config: hostConfig, - registerTool(tool) { - registered.push(tool); - }, - } as never); - - return registered; -} - -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}`); - } - return found; -} - -describe("agentmail", () => { - beforeEach(() => { - vi.stubEnv("AGENTMAIL_API_KEY", "am_test"); - vi.clearAllMocks(); - }); - - afterEach(() => { - vi.unstubAllEnvs(); - }); - - it("declares all AgentMail tool metadata", () => { - expect(getToolPluginMetadata(entry)?.tools.map((tool) => tool.name)).toEqual([ - "agentmail_list_inboxes", - "agentmail_create_inbox", - "agentmail_list_messages", - "agentmail_search_messages", - "agentmail_get_message", - "agentmail_send_message", - "agentmail_reply_to_message", - "agentmail_forward_message", - "agentmail_update_message_labels", - ]); - }); - - it("uses the API key and optional plugin client settings", async () => { - sdk.inboxes.list.mockResolvedValue({ count: 0, inboxes: [] }); - const signal = new AbortController().signal; - const tool = findTool("agentmail_list_inboxes", { - baseUrl: "https://example.test/v0", - timeoutSeconds: 15, - maxRetries: 0, - }); - - await tool.execute("call-1", { limit: 5 }, signal); - - expect(sdk.constructor).toHaveBeenCalledWith({ - apiKey: "am_test", - baseUrl: "https://example.test/v0", - timeoutInSeconds: 15, - maxRetries: 0, - }); - expect(sdk.inboxes.list).toHaveBeenCalledWith( - { limit: 5 }, - { abortSignal: signal }, - ); - }); - - 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; - const tool = findTool("agentmail_send_message"); - - await tool.execute( - "call-2", - { - inboxId: "agent@agentmail.to", - to: ["person@example.com"], - subject: "Hello", - text: "Plain text", - idempotencyKey: "send-123", - }, - signal, - ); - - expect(sdk.messages.send).toHaveBeenCalledWith( - "agent@agentmail.to", - { - to: ["person@example.com"], - subject: "Hello", - text: "Plain text", - }, - { abortSignal: signal, idempotencyKey: "send-123" }, - ); - }); - - it("converts message filter timestamps to Date values", async () => { - sdk.messages.list.mockResolvedValue({ count: 0, messages: [] }); - const tool = findTool("agentmail_list_messages"); - - await tool.execute("call-3", { - inboxId: "agent@agentmail.to", - before: "2026-07-21T12:00:00.000Z", - after: "2026-07-20T12:00:00.000Z", - }); - - expect(sdk.messages.list).toHaveBeenCalledWith( - "agent@agentmail.to", - { - before: new Date("2026-07-21T12:00:00.000Z"), - after: new Date("2026-07-20T12:00:00.000Z"), - }, - {}, - ); - }); - - it("requires a configured API key before making a request", async () => { - vi.stubEnv("AGENTMAIL_API_KEY", " "); - const tool = findTool("agentmail_list_inboxes"); - - await expect(tool.execute("call-4", {})).rejects.toThrow( - "Configure channels.agentmail.apiKey or set AGENTMAIL_API_KEY", - ); - expect(sdk.constructor).not.toHaveBeenCalled(); - }); - - it("requires at least one message label change", async () => { - const tool = findTool("agentmail_update_message_labels"); - - await expect( - tool.execute("call-5", { - inboxId: "agent@agentmail.to", - messageId: "msg_1", - }), - ).rejects.toThrow("Provide addLabels or removeLabels"); - expect(sdk.messages.update).not.toHaveBeenCalled(); - }); -}); diff --git a/src/tools/index.ts b/src/tools/index.ts deleted file mode 100644 index 958ed1c..0000000 --- a/src/tools/index.ts +++ /dev/null @@ -1,377 +0,0 @@ -import { Type } from "typebox"; -import { defineToolPlugin } from "openclaw/plugin-sdk/tool-plugin"; -import { - agentMailConfigSchema, - createAgentMailClient, - requestOptions, -} from "./client.js"; - -const nonEmptyString = (description: string) => - Type.String({ description, minLength: 1 }); - -const optionalRecipients = (description: string) => - Type.Optional( - Type.Array(nonEmptyString(description), { - description, - maxItems: 50, - }), - ); - -const optionalLabels = Type.Optional( - Type.Array(nonEmptyString("Label name."), { - description: "Message labels.", - maxItems: 100, - }), -); - -const optionalDateTime = (description: string) => - Type.Optional(Type.String({ description, format: "date-time" })); - -const attachmentsSchema = Type.Optional( - Type.Array( - Type.Object( - { - filename: Type.Optional(nonEmptyString("Attachment filename.")), - contentType: Type.Optional(nonEmptyString("MIME content type.")), - contentDisposition: Type.Optional( - Type.Union([Type.Literal("attachment"), Type.Literal("inline")], { - description: "Whether the attachment is downloaded or displayed inline.", - }), - ), - contentId: Type.Optional(nonEmptyString("Content ID for an inline attachment.")), - content: Type.Optional(nonEmptyString("Base64-encoded attachment content.")), - url: Type.Optional(nonEmptyString("Public URL from which AgentMail can fetch the attachment.")), - }, - { additionalProperties: false }, - ), - { description: "Attachments to include in the message." }, - ), -); - -const composeFields = { - to: optionalRecipients("Recipient email address."), - cc: optionalRecipients("CC recipient email address."), - bcc: optionalRecipients("BCC recipient email address."), - replyTo: optionalRecipients("Reply-to email address."), - labels: optionalLabels, - text: Type.Optional(Type.String({ description: "Plain-text message body." })), - html: Type.Optional(Type.String({ description: "HTML message body." })), - attachments: attachmentsSchema, -}; - -const messageLocationFields = { - inboxId: nonEmptyString("Inbox ID or email address that owns the message."), - messageId: nonEmptyString("AgentMail message ID."), -}; - -function parseDate(value: string | undefined): Date | undefined { - if (!value) { - return undefined; - } - const date = new Date(value); - if (Number.isNaN(date.getTime())) { - // Fail fast with a clear message instead of passing an Invalid Date into the SDK, which would - // otherwise surface later as an opaque RangeError. - throw new Error(`Invalid ISO 8601 timestamp: ${JSON.stringify(value)}`); - } - return date; -} - -export default defineToolPlugin({ - id: "agentmail", - name: "AgentMail", - description: "Create AgentMail inboxes and send, receive, search, and manage email.", - configSchema: agentMailConfigSchema, - tools: (tool) => [ - tool({ - name: "agentmail_list_inboxes", - label: "List AgentMail inboxes", - description: "List the email inboxes available to the configured AgentMail account.", - parameters: Type.Object({ - limit: Type.Optional( - Type.Integer({ description: "Maximum number of inboxes to return.", minimum: 1 }), - ), - pageToken: Type.Optional(nonEmptyString("Pagination token from a previous response.")), - ascending: Type.Optional( - Type.Boolean({ description: "Return inboxes in ascending creation order." }), - ), - }, { additionalProperties: false }), - execute: async (params, config, context) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.list(params, requestOptions(context.signal)); - }, - }), - tool({ - name: "agentmail_create_inbox", - label: "Create AgentMail inbox", - description: - "Create an AgentMail inbox. Use clientId when the operation may be retried to avoid duplicate inboxes.", - parameters: Type.Object( - { - username: Type.Optional(nonEmptyString("Requested email username.")), - domain: Type.Optional(nonEmptyString("Verified domain; defaults to agentmail.to.")), - displayName: Type.Optional(nonEmptyString("Human-readable sender display name.")), - clientId: Type.Optional(nonEmptyString("Idempotency key for inbox creation.")), - metadata: Type.Optional( - Type.Record( - Type.String(), - Type.Union([Type.String(), Type.Number(), Type.Boolean()]), - { description: "Custom inbox metadata." }, - ), - ), - }, - { additionalProperties: false }, - ), - execute: async (params, config, context) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.create(params, requestOptions(context.signal)); - }, - }), - tool({ - name: "agentmail_list_messages", - label: "List AgentMail messages", - description: - "List messages in an AgentMail inbox, newest first by default. Supports labels and exact-field substring filters.", - parameters: Type.Object( - { - inboxId: nonEmptyString("Inbox ID or email address to read."), - limit: Type.Optional( - Type.Integer({ description: "Maximum number of messages to return.", minimum: 1 }), - ), - pageToken: Type.Optional(nonEmptyString("Pagination token from a previous response.")), - labels: optionalLabels, - before: optionalDateTime("Only include messages before this ISO 8601 timestamp."), - after: optionalDateTime("Only include messages after this ISO 8601 timestamp."), - ascending: Type.Optional( - Type.Boolean({ description: "Return messages oldest first." }), - ), - includeSpam: Type.Optional(Type.Boolean({ description: "Include spam messages." })), - includeBlocked: Type.Optional(Type.Boolean({ description: "Include blocked messages." })), - includeUnauthenticated: Type.Optional( - Type.Boolean({ description: "Include unauthenticated messages." }), - ), - includeTrash: Type.Optional(Type.Boolean({ description: "Include trashed messages." })), - from: Type.Optional( - Type.Array(nonEmptyString("Sender substring filter."), { - description: "Sender substring filters; all values must match.", - }), - ), - to: Type.Optional( - Type.Array(nonEmptyString("Recipient substring filter."), { - description: "Recipient substring filters; all values must match.", - }), - ), - subject: Type.Optional( - Type.Array(nonEmptyString("Subject substring filter."), { - description: "Subject substring filters; all values must match.", - }), - ), - }, - { additionalProperties: false }, - ), - execute: async ({ inboxId, before, after, ...params }, config, context) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.list( - inboxId, - { - ...params, - ...(before ? { before: parseDate(before) } : {}), - ...(after ? { after: parseDate(after) } : {}), - }, - requestOptions(context.signal), - ); - }, - }), - tool({ - name: "agentmail_search_messages", - label: "Search AgentMail messages", - description: - "Full-text search an AgentMail inbox across sender, recipients, subject, and body, ranked by relevance.", - parameters: Type.Object( - { - inboxId: nonEmptyString("Inbox ID or email address to search."), - query: nonEmptyString("Full-text search query."), - limit: Type.Optional( - Type.Integer({ description: "Maximum number of matches to return.", minimum: 1, maximum: 100 }), - ), - pageToken: Type.Optional(nonEmptyString("Pagination token from a previous response.")), - before: optionalDateTime("Only include messages before this ISO 8601 timestamp."), - after: optionalDateTime("Only include messages after this ISO 8601 timestamp."), - }, - { additionalProperties: false }, - ), - execute: async ({ inboxId, query, before, after, ...params }, config, context) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.search( - inboxId, - { - q: query, - ...params, - ...(before ? { before: parseDate(before) } : {}), - ...(after ? { after: parseDate(after) } : {}), - }, - requestOptions(context.signal), - ); - }, - }), - tool({ - name: "agentmail_get_message", - label: "Get AgentMail message", - description: - "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, context.api.config); - return client.inboxes.messages.get( - inboxId, - messageId, - requestOptions(context.signal), - ); - }, - }), - tool({ - name: "agentmail_send_message", - label: "Send AgentMail message", - description: - "Send a new email from an AgentMail inbox. Provide both text and HTML when practical for accessibility and deliverability.", - parameters: Type.Object( - { - inboxId: nonEmptyString("Inbox ID or email address to send from."), - to: Type.Array(nonEmptyString("Recipient email address."), { - description: "Primary recipients.", - minItems: 1, - maxItems: 50, - }), - subject: nonEmptyString("Email subject."), - text: Type.Optional(Type.String({ description: "Plain-text message body." })), - html: Type.Optional(Type.String({ description: "HTML message body." })), - cc: composeFields.cc, - bcc: composeFields.bcc, - replyTo: composeFields.replyTo, - labels: composeFields.labels, - attachments: composeFields.attachments, - idempotencyKey: Type.Optional( - nonEmptyString("Key that prevents duplicate sends when retrying the same operation."), - ), - }, - { additionalProperties: false }, - ), - execute: async ({ inboxId, idempotencyKey, ...message }, config, context) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.send(inboxId, message, { - ...requestOptions(context.signal), - ...(idempotencyKey ? { idempotencyKey } : {}), - }); - }, - }), - tool({ - name: "agentmail_reply_to_message", - label: "Reply to AgentMail message", - description: "Reply to an existing message while keeping the AgentMail thread intact.", - parameters: Type.Object( - { - ...messageLocationFields, - text: composeFields.text, - html: composeFields.html, - to: composeFields.to, - cc: composeFields.cc, - bcc: composeFields.bcc, - replyTo: composeFields.replyTo, - labels: composeFields.labels, - attachments: composeFields.attachments, - replyAll: Type.Optional( - Type.Boolean({ description: "Reply to all original recipients." }), - ), - idempotencyKey: Type.Optional( - nonEmptyString("Key that prevents duplicate replies when retrying the same operation."), - ), - }, - { additionalProperties: false }, - ), - execute: async ( - { inboxId, messageId, idempotencyKey, ...message }, - config, - context, - ) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.reply(inboxId, messageId, message, { - ...requestOptions(context.signal), - ...(idempotencyKey ? { idempotencyKey } : {}), - }); - }, - }), - tool({ - name: "agentmail_forward_message", - label: "Forward AgentMail message", - description: "Forward an existing AgentMail message to one or more recipients.", - parameters: Type.Object( - { - ...messageLocationFields, - to: Type.Array(nonEmptyString("Recipient email address."), { - description: "Forward recipients.", - minItems: 1, - maxItems: 50, - }), - subject: Type.Optional(nonEmptyString("Optional subject override.")), - text: composeFields.text, - html: composeFields.html, - cc: composeFields.cc, - bcc: composeFields.bcc, - replyTo: composeFields.replyTo, - labels: composeFields.labels, - attachments: composeFields.attachments, - idempotencyKey: Type.Optional( - nonEmptyString("Key that prevents duplicate forwards when retrying the same operation."), - ), - }, - { additionalProperties: false }, - ), - execute: async ( - { inboxId, messageId, idempotencyKey, ...message }, - config, - context, - ) => { - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.forward(inboxId, messageId, message, { - ...requestOptions(context.signal), - ...(idempotencyKey ? { idempotencyKey } : {}), - }); - }, - }), - tool({ - name: "agentmail_update_message_labels", - label: "Update AgentMail message labels", - description: - "Add or remove labels on a message, for example adding read and removing unread after processing it.", - parameters: Type.Object( - { - ...messageLocationFields, - addLabels: Type.Optional( - Type.Array(nonEmptyString("Label to add."), { minItems: 1 }), - ), - removeLabels: Type.Optional( - Type.Array(nonEmptyString("Label to remove."), { minItems: 1 }), - ), - }, - { additionalProperties: false }, - ), - execute: async ( - { inboxId, messageId, ...labels }, - config, - context, - ) => { - if (!labels.addLabels && !labels.removeLabels) { - throw new Error("Provide addLabels or removeLabels."); - } - - const client = createAgentMailClient(config, context.api.config); - return client.inboxes.messages.update( - inboxId, - messageId, - labels, - requestOptions(context.signal), - ); - }, - }), - ], -}); diff --git a/tsconfig.json b/tsconfig.json index a529e29..8ca9412 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -3,6 +3,7 @@ "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext", + "resolveJsonModule": true, "strict": true, "declaration": true, "outDir": "dist",