From fba2375a45e3e193a052264f9787de7402afbd0a Mon Sep 17 00:00:00 2001 From: Jake Date: Tue, 27 Jan 2026 14:11:01 +1300 Subject: [PATCH] feat: add Gmail channel support via gog CLI --- docs/channels/gmail.md | 131 ++++++ extensions/gmail/README.md | 50 +++ extensions/gmail/clawdbot.plugin.json | 11 + extensions/gmail/index.ts | 14 + extensions/gmail/package.json | 45 +++ extensions/gmail/src/accounts.ts | 57 +++ extensions/gmail/src/attachments.ts | 29 ++ extensions/gmail/src/channel.ts | 360 +++++++++++++++++ extensions/gmail/src/clawdbot.plugin.json | 11 + extensions/gmail/src/config.ts | 24 ++ extensions/gmail/src/history-store.ts | 42 ++ extensions/gmail/src/html.ts | 5 + extensions/gmail/src/inbound.test.ts | 111 +++++ extensions/gmail/src/inbound.ts | 183 +++++++++ extensions/gmail/src/monitor.test.ts | 103 +++++ extensions/gmail/src/monitor.ts | 472 ++++++++++++++++++++++ extensions/gmail/src/normalize.test.ts | 17 + extensions/gmail/src/normalize.ts | 42 ++ extensions/gmail/src/onboarding.ts | 214 ++++++++++ extensions/gmail/src/outbound.test.ts | 95 +++++ extensions/gmail/src/outbound.ts | 113 ++++++ extensions/gmail/src/runtime.ts | 14 + extensions/gmail/src/semaphore.ts | 36 ++ extensions/gmail/src/strip-quotes.test.ts | 9 + extensions/gmail/src/strip-quotes.ts | 30 ++ extensions/gmail/src/threading.ts | 8 + src/commands/onboard-channels.ts | 35 +- 27 files changed, 2247 insertions(+), 14 deletions(-) create mode 100644 docs/channels/gmail.md create mode 100644 extensions/gmail/README.md create mode 100644 extensions/gmail/clawdbot.plugin.json create mode 100644 extensions/gmail/index.ts create mode 100644 extensions/gmail/package.json create mode 100644 extensions/gmail/src/accounts.ts create mode 100644 extensions/gmail/src/attachments.ts create mode 100644 extensions/gmail/src/channel.ts create mode 100644 extensions/gmail/src/clawdbot.plugin.json create mode 100644 extensions/gmail/src/config.ts create mode 100644 extensions/gmail/src/history-store.ts create mode 100644 extensions/gmail/src/html.ts create mode 100644 extensions/gmail/src/inbound.test.ts create mode 100644 extensions/gmail/src/inbound.ts create mode 100644 extensions/gmail/src/monitor.test.ts create mode 100644 extensions/gmail/src/monitor.ts create mode 100644 extensions/gmail/src/normalize.test.ts create mode 100644 extensions/gmail/src/normalize.ts create mode 100644 extensions/gmail/src/onboarding.ts create mode 100644 extensions/gmail/src/outbound.test.ts create mode 100644 extensions/gmail/src/outbound.ts create mode 100644 extensions/gmail/src/runtime.ts create mode 100644 extensions/gmail/src/semaphore.ts create mode 100644 extensions/gmail/src/strip-quotes.test.ts create mode 100644 extensions/gmail/src/strip-quotes.ts create mode 100644 extensions/gmail/src/threading.ts diff --git a/docs/channels/gmail.md b/docs/channels/gmail.md new file mode 100644 index 000000000..ef3e833fb --- /dev/null +++ b/docs/channels/gmail.md @@ -0,0 +1,131 @@ +--- +summary: "Gmail channel status, capabilities, and configuration using polling or Pub/Sub" +read_when: + - Working on Gmail channel features or debugging email sync +--- +# Gmail (plugin) + +The Gmail channel allows Clawdbot to send and receive emails via the Google Gmail API. It leverages the `gog` CLI tool for authentication and API interactions. + +## Status +- **Text**: Fully supported (Markdown auto-converted to rich HTML). +- **Threading**: Fully supported (replies thread to original emails). +- **Attachments**: Automatic metadata injection (Filename, Type, Size, ID). Agents can download attachments on-demand using `gog`. +- **Sync**: Supports robust polling with circuit breaker and optional Pub/Sub webhooks. + +### Attachments +Keith automatically manages attachments to optimize for speed and storage: +1. **Auto-Download**: Attachments **under 5MB** are automatically downloaded to Keith's local cache. The agent sees a direct file path (e.g., `~/.attachments/thread-123/invoice.pdf`) and can read it immediately. +2. **On-Demand**: Larger attachments are presented as metadata: `[Attachment: large-video.mp4 (Type: video/mp4, Size: 150 MB, ID: ...)]`. If Keith needs these, he must use the `gog` tool to download them explicitly. + +## Safety & Security + +### Allowlist Enforcement +Keith strictly enforces an allowlist to prevent unauthorized usage and spam. +- **Inbound Protection (Quarantine)**: If an email arrives from a sender NOT in the `allowFrom` list: + - It is **quarantined**: The label `not-allow-listed` is applied. + - It is **removed from Inbox**: The `INBOX` label is removed. + - It remains **UNREAD**: So you can review it later if needed. + - **Keith never sees it**: The message is silently filtered before reaching the agent logic. + +- **Outbound Protection**: + - Keith cannot send *new* emails to addresses not on the allowlist. + - **Replies are permitted**: If Keith is replying to a valid thread (one that passed the inbound filter), the reply is allowed regardless of the recipient list, as the thread is trusted. + +### "Reply All" Behavior +To function as a collaborative participant, Keith defaults to **Reply All** behavior when responding to threads. +- When Keith replies to a thread, the response goes to the **Sender** and all **CC'd** recipients of the *latest* message in that thread. +- **Privacy Note**: Be aware that if an allowed sender includes external parties on a thread, Keith's reply will be visible to them. + +## Opinionated "Inbox Zero" Workflow +**Important**: The Gmail channel enforces an "Inbox Zero" philosophy to maintain a clean state and prevent infinite reply loops. +- **Reply = Archive**: When Keith sends a reply to a thread, that thread is **automatically archived** (the `INBOX` label is removed). +- **Why?**: If the thread remained in the Inbox, future sync cycles might re-process it or confuse the agent about what is "pending." +- **Result**: You will see Keith's replies in "All Mail" or "Sent," but the thread will disappear from your Inbox view once handled. + +## Prerequisites +1. **Install gog**: Ensure the `gog` binary is in your PATH. The channel will fail to start if `gog` is missing. +2. **Authenticate**: Run `gog auth login` for the accounts you want Keith to use. + ```bash + gog auth login + ``` + +## Configuration Reference + +The following configuration options are available in `clawdbot.json` under `channels.gmail`. + +### Global Channel Settings +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `enabled` | boolean | `true` | Globally enable or disable the Gmail channel. | +| `accounts` | object | `{}` | Map of account configurations (keyed by account ID). | +| `defaults` | object | `{}` | Default settings applied to all accounts. | + +### Account Settings (`channels.gmail.accounts.`) +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `enabled` | boolean | `true` | Enable/disable this specific account. | +| `email` | string | **Required** | The Gmail address to monitor and send from. | +| `name` | string | `email` | Display name for this account in logs and UI. | +| `allowFrom` | string[] | `[]` | Whitelist of senders. Supports exact emails or `@domain.com` wildcards. Use `["*"]` for open access. | +| `pollIntervalMs` | number | `60000` | Frequency of polling for new messages in milliseconds. | +| `delegate` | string | `null` | Optional: The email address of a delegator if using Gmail delegation. | +| `sessionTtlDays` | number | `7` | How many days to keep inactive thread sessions before pruning. | + +### Example `clawdbot.json` +```json5 +{ + "channels": { + "gmail": { + "enabled": true, + "accounts": { + "work": { + "email": "keith.agent@company.com", + "allowFrom": ["*@company.com"], + "pollIntervalMs": 30000 + }, + "personal": { + "email": "my-bot@gmail.com", + "allowFrom": ["my-phone@gmail.com"] + } + } + } + } +} +``` + +## How it works + +### Inbound Processing +1. **Sync**: Every `pollIntervalMs`, the monitor loop uses `gog` to search for `label:INBOX is:unread`. +2. **Parsing**: + - Extracts plain text and HTML. + - Prepends `[Thread Context: Subject is "..."]` to the body so the agent always knows the context of the email chain. + - Strips quoted "On [date], [user] wrote:" text to keep the prompt clean. +3. **Resilience**: A per-account **Circuit Breaker** tracks failures. If the API returns repeated errors (403, 500, etc.), the account will enter a "backoff" state, increasing the wait time between retries to avoid account lockout. + +### Outbound Processing +1. **Markdown**: Keith's responses are parsed as Markdown. +2. **HTML Generation**: Markdown is converted to rich HTML (tables, bold, links, etc.) using `marked` and sanitized. +3. **Delivery**: Replies are sent via `gog gmail send`. +4. **Threading**: Replies automatically include the `In-Reply-To` and `References` headers derived from the original message ID. +5. **Archiving**: To prevent "reply loops," the original thread is automatically removed from the `INBOX` label after Keith sends a response. + +## Routing & Sessions +- **Session Key**: `gmail::` +- Each Gmail thread is isolated. This means Keith maintains a separate memory/history for every unique email conversation. +- If you forward an email to Keith, it will start a new session based on that new thread ID. + +## Target Formats (CLI/Cron) +To send an email via CLI or a cron job: +- **Target**: `gmail:` (starts a new thread) or `gmail:` (replies to existing). +- **Command**: + ```bash + clawdbot message send --channel gmail --target "boss@example.com" --message "Report is ready." + ``` + +## Troubleshooting +- **Monitor Latency**: If emails take too long to arrive, check `pollIntervalMs`. +- **Request IDs**: Every inbound message is assigned a 8-character ID. Search logs for `[gmail][]` to see the full lifecycle from sync to reply. +- **Token Expiry**: Run `gog auth list` to check token status. If an account stops syncing, try `gog auth login --account `. +- **Permission Errors**: Ensure the Google Cloud Project has the **Gmail API** enabled. diff --git a/extensions/gmail/README.md b/extensions/gmail/README.md new file mode 100644 index 000000000..b9615e9a1 --- /dev/null +++ b/extensions/gmail/README.md @@ -0,0 +1,50 @@ +# Gmail Channel (Plugin) + +Connects Clawdbot to Gmail via the `gog` CLI. + +## Installation + +This is a plugin. To install from source: + +```bash +clawdbot plugins install ./extensions/gmail +``` + +## Features + +- **Polling-based sync**: Robustly fetches new unread emails from Inbox. +- **Circuit Breaker**: Handles API failures and rate limiting gracefully. +- **Rich Text**: Markdown support for outbound emails. +- **Threading**: Native Gmail thread support. +- **Archiving**: Automatically archives threads upon reply. + +## Reply Behavior + +- **Reply All**: When the bot replies to a thread, it uses "Reply All" to ensure all participants are included. +- **Allowlist Gatekeeping**: The bot only responds to emails from senders on the `allowFrom` list. However, if an allowed user includes others (CC) who are *not* on the allowlist, the bot will still "Reply All", including them in the conversation. This allows authorized users to bring others into the loop. + +## Configuration + +Add to `clawdbot.json`: + +```json5 +{ + "channels": { + "gmail": { + "accounts": { + "main": { + "email": "user@gmail.com", + "allowFrom": ["*"] + } + } + } + } +} +``` + +## Development + +Run tests: +```bash +./node_modules/.bin/vitest run extensions/gmail/src/ +``` diff --git a/extensions/gmail/clawdbot.plugin.json b/extensions/gmail/clawdbot.plugin.json new file mode 100644 index 000000000..e3626ab9e --- /dev/null +++ b/extensions/gmail/clawdbot.plugin.json @@ -0,0 +1,11 @@ +{ + "id": "gmail", + "channels": [ + "gmail" + ], + "configSchema": { + "type": "object", + "additionalProperties": false, + "properties": {} + } +} diff --git a/extensions/gmail/index.ts b/extensions/gmail/index.ts new file mode 100644 index 000000000..45ed37eba --- /dev/null +++ b/extensions/gmail/index.ts @@ -0,0 +1,14 @@ +import type { ClawdbotPluginApi } from "clawdbot/plugin-sdk"; +import { gmailPlugin } from "./src/channel.js"; + +import { setGmailRuntime } from "./src/runtime.js"; + +const plugin = { + ...gmailPlugin, + register: (api: ClawdbotPluginApi) => { + setGmailRuntime(api.runtime); + api.registerChannel(gmailPlugin); + } +}; + +export default plugin; diff --git a/extensions/gmail/package.json b/extensions/gmail/package.json new file mode 100644 index 000000000..0ee9b3d19 --- /dev/null +++ b/extensions/gmail/package.json @@ -0,0 +1,45 @@ +{ + "name": "gmail", + "version": "1.0.0", + "description": "Gmail channel for Clawdbot", + "type": "module", + "main": "index.ts", + "clawdbot": { + "extensions": [ + "./index.ts" + ], + "channel": { + "id": "gmail", + "label": "Gmail", + "selectionLabel": "Gmail (gog)", + "detailLabel": "Gmail", + "docsPath": "/channels/gmail", + "docsLabel": "gmail", + "blurb": "Uses gog for secure Gmail access.", + "systemImage": "envelope", + "order": 100, + "showConfigured": true + } + }, + "scripts": { + "clean": "rm -rf dist", + "lint": "oxlint", + "format": "oxfmt" + }, + "dependencies": { + "jsdom": "^25.0.1", + "dompurify": "^3.2.3", + "zod": "^4.3.5", + "marked": "^15.0.6", + "sanitize-html": "^2.14.0" + }, + "devDependencies": { + "clawdbot": "workspace:*", + "@types/jsdom": "^21.1.7", + "@types/dompurify": "^3.0.5", + "@types/marked": "^6.0.0", + "@types/sanitize-html": "^2.13.0", + "@types/node": "^22.0.0", + "typescript": "^5.0.0" + } +} diff --git a/extensions/gmail/src/accounts.ts b/extensions/gmail/src/accounts.ts new file mode 100644 index 000000000..115b94377 --- /dev/null +++ b/extensions/gmail/src/accounts.ts @@ -0,0 +1,57 @@ +import { + type ChannelConfig, + type ResolvedChannelAccount, + DEFAULT_ACCOUNT_ID, +} from "clawdbot/plugin-sdk"; +import type { GmailConfig } from "./config.js"; + +export interface ResolvedGmailAccount extends ResolvedChannelAccount { + email: string; + historyId?: string; + delegate?: string; + pollIntervalMs?: number; +} + +export function resolveGmailAccount( + cfg: ChannelConfig, + accountId?: string, +): ResolvedGmailAccount { + const resolvedId = accountId || DEFAULT_ACCOUNT_ID; + const account = cfg.channels?.gmail?.accounts?.[resolvedId]; + + if (!account) { + // Graceful fallback for UI logic that queries 'default' on unconfigured channels + return { + accountId: resolvedId, + name: resolvedId, + enabled: false, + email: "", + historyId: undefined, + delegate: undefined, + allowFrom: [], + pollIntervalMs: undefined, + }; + } + + return { + accountId: resolvedId, + name: account.name || account.email, + enabled: account.enabled, + email: account.email, + historyId: account.historyId, + delegate: account.delegate, + allowFrom: account.allowFrom, + pollIntervalMs: account.pollIntervalMs, + }; +} + +export function listGmailAccountIds(cfg: ChannelConfig): string[] { + return Object.keys(cfg.channels?.gmail?.accounts || {}); +} + +export function resolveDefaultGmailAccountId(cfg: ChannelConfig): string { + const ids = listGmailAccountIds(cfg); + if (ids.length === 0) return DEFAULT_ACCOUNT_ID; + if (ids.includes(DEFAULT_ACCOUNT_ID)) return DEFAULT_ACCOUNT_ID; + return ids[0]; // Fallback to first +} diff --git a/extensions/gmail/src/attachments.ts b/extensions/gmail/src/attachments.ts new file mode 100644 index 000000000..b5490c315 --- /dev/null +++ b/extensions/gmail/src/attachments.ts @@ -0,0 +1,29 @@ +import type { GogMessagePart } from "./inbound.js"; + +export interface GmailAttachment { + filename: string; + mimeType: string; + attachmentId: string; + size: number; +} + +export function extractAttachments(part: GogMessagePart): GmailAttachment[] { + const attachments: GmailAttachment[] = []; + + if (part.filename && part.body?.attachmentId) { + attachments.push({ + filename: part.filename, + mimeType: part.mimeType, + attachmentId: part.body.attachmentId, + size: part.body.size, + }); + } + + if (part.parts) { + for (const subPart of part.parts) { + attachments.push(...extractAttachments(subPart)); + } + } + + return attachments; +} diff --git a/extensions/gmail/src/channel.ts b/extensions/gmail/src/channel.ts new file mode 100644 index 000000000..eddf99ea7 --- /dev/null +++ b/extensions/gmail/src/channel.ts @@ -0,0 +1,360 @@ +import { + buildChannelConfigSchema, + getChatChannelMeta, + type ChannelPlugin, + missingTargetError, + setAccountEnabledInConfigSection, + deleteAccountFromConfigSection, + type InboundMessage, + type ClawdbotConfig, + type ChannelGatewayContext, + type MsgContext, +} from "clawdbot/plugin-sdk"; +import { GmailConfigSchema } from "./config.js"; +import { + resolveGmailAccount, + resolveDefaultGmailAccountId, + listGmailAccountIds, + type ResolvedGmailAccount, +} from "./accounts.js"; +import { setGmailRuntime, getGmailRuntime } from "./runtime.js"; +import { sendGmailText, type GmailOutboundContext } from "./outbound.js"; +import { gmailThreading } from "./threading.js"; +import { normalizeGmailTarget, isGmailThreadId, isAllowed } from "./normalize.js"; +import { parseInboundGmail, type GogPayload } from "./inbound.js"; +import { monitorGmail, quarantineMessage } from "./monitor.js"; +import { extractAttachments } from "./attachments.js"; +import { Semaphore } from "./semaphore.js"; +import crypto from "node:crypto"; + +const meta = { + id: "gmail", + label: "Gmail", + selectionLabel: "Gmail (gog)", + detailLabel: "Gmail", + docsPath: "/channels/gmail", + docsLabel: "gmail", + blurb: "Uses gog for secure Gmail access.", + systemImage: "envelope", + order: 100, + showConfigured: true, +}; + +// Map to store active account contexts +const activeAccounts = new Map>(); + +// Limit concurrent dispatches to avoid memory spikes +const dispatchSemaphore = new Semaphore(5); + +/** + * Convert an InboundMessage to a finalized MsgContext for dispatch. + * Gmail threads are equivalent to Slack channels - each thread gets its own session. + */ +function buildGmailMsgContext( + msg: InboundMessage, + account: ResolvedGmailAccount, + cfg: ClawdbotConfig, +): MsgContext { + const runtime = getGmailRuntime(); + const to = `gmail:${account.email}`; + const threadLabel = `Gmail thread ${msg.threadId}`; + + const ctx = runtime.channel.reply.finalizeInboundContext({ + Body: msg.text, + RawBody: msg.text, + CommandBody: msg.text, + From: msg.sender.id, + To: to, + SessionKey: `gmail:${account.email}:${msg.threadId}`, + AccountId: msg.accountId, + ChatType: "direct", + ConversationLabel: threadLabel, + SenderName: msg.sender.name, + SenderId: msg.sender.id, + Provider: "gmail" as const, + Surface: "gmail" as const, + MessageSid: msg.channelMessageId, + ReplyToId: msg.channelMessageId, + ThreadLabel: threadLabel, + MessageThreadId: msg.threadId, + ThreadStarterBody: undefined, + Timestamp: msg.timestamp ? Math.round(msg.timestamp / 1_000) : undefined, // InboundMessage timestamp is ms, finalizeInboundContext expects seconds + MediaPath: msg.mediaPath, + MediaType: msg.mediaType, + MediaUrl: msg.mediaUrl, + CommandAuthorized: false, + OriginatingChannel: "gmail" as const, + OriginatingTo: to, + }); + + return ctx; +} + +async function dispatchGmailMessage( + ctx: ChannelGatewayContext, + msg: InboundMessage, +) { + const { account, accountId, cfg, log } = ctx; + const runtime = getGmailRuntime(); + const requestId = crypto.randomUUID().split("-")[0]; + + await dispatchSemaphore.run(async () => { + try { + log?.info(`[gmail][${requestId}] Dispatching message ${msg.channelMessageId} from ${msg.sender.id}`); + + // Build the dispatch context + const ctxPayload = buildGmailMsgContext(msg, account, cfg); + + // Build reply dispatcher options using gateway's reply capability + const deliver = async (payload: { text: string }) => { + const originalSubject = msg.raw?.subject || + msg.raw?.headers?.subject || + msg.raw?.payload?.headers?.find((h: any) => h.name.toLowerCase() === "subject")?.value; + + const replySubject = originalSubject + ? (originalSubject.toLowerCase().startsWith("re:") ? originalSubject : `Re: ${originalSubject}`) + : "Re: "; + + await sendGmailText({ + to: msg.threadId || msg.sender.id, + text: payload.text, + accountId, + cfg, + threadId: msg.threadId, + replyToId: msg.channelMessageId, + subject: replySubject, + }); + }; + + const humanDelay = runtime.channel.reply.resolveHumanDelayConfig(cfg, accountId); + + // Dispatch to agent + await runtime.channel.reply.dispatchReplyWithBufferedBlockDispatcher({ + ctx: ctxPayload, + cfg, + dispatcherOptions: { + deliver, + humanDelay, + onError: (err: unknown, info: { kind: string }) => { + log?.error(`[gmail][${requestId}] ${info.kind} reply failed: ${String(err)}`); + }, + }, + }); + log?.info(`[gmail][${requestId}] Dispatch complete for ${msg.channelMessageId}`); + } catch (e: unknown) { + log?.error(`[gmail][${requestId}] Dispatch failed: ${e instanceof Error ? e.message : String(e)}`); + } + }); +} + +import { gmailOnboardingAdapter } from "./onboarding.js"; + +export const gmailPlugin: ChannelPlugin = { + id: "gmail", + onboarding: gmailOnboardingAdapter, + meta: { + ...meta, + showConfigured: true, + }, + capabilities: { + chatTypes: ["direct", "group"], + media: true, + threads: true, + }, + configSchema: { + schema: { + type: "object", + properties: { + enabled: { type: "boolean", default: true }, + accounts: { + type: "object", + additionalProperties: { + type: "object", + properties: { + enabled: { type: "boolean", default: true }, + email: { type: "string" }, + name: { type: "string" }, + allowFrom: { type: "array", items: { type: "string" } }, + historyId: { type: "string" }, + delegate: { type: "string" }, + }, + required: ["email"], + }, + }, + defaults: { + type: "object", + properties: { + allowFrom: { type: "array", items: { type: "string" } }, + }, + }, + }, + }, + }, + config: { + listAccountIds: (cfg) => listGmailAccountIds(cfg), + resolveAccount: (cfg, accountId) => resolveGmailAccount(cfg, accountId), + defaultAccountId: (cfg) => resolveDefaultGmailAccountId(cfg), + isEnabled: (account) => account.enabled, + describeAccount: (account) => ({ + accountId: account.accountId, + name: account.name || account.email, + enabled: account.enabled, + configured: true, + linked: true, + allowFrom: account.allowFrom, + }), + resolveAllowFrom: ({ cfg, accountId }) => + resolveGmailAccount(cfg, accountId ?? undefined).allowFrom ?? [], + formatAllowFrom: ({ allowFrom }) => + allowFrom.map((e) => String(e).trim()).filter(Boolean), + setAccountEnabled: ({ cfg, accountId, enabled }) => + setAccountEnabledInConfigSection({ + cfg, + sectionKey: "gmail", + accountId, + enabled, + allowTopLevel: true, + }), + deleteAccount: ({ cfg, accountId }) => + deleteAccountFromConfigSection({ + cfg, + sectionKey: "gmail", + accountId, + }), + }, + outbound: { + deliveryMode: "gateway", + textChunkLimit: 8000, + sendText: sendGmailText, + resolveTarget: ({ to, allowFrom }) => { + const trimmed = to?.trim() ?? ""; + const normalized = normalizeGmailTarget(trimmed); + + if (!normalized) { + return { + ok: false, + error: missingTargetError("Gmail", "email address or thread ID"), + }; + } + + // If it's a thread ID, we allow it implicitly (assuming we only have thread IDs + // for threads we were allowed to ingest). + if (isGmailThreadId(normalized)) { + return { ok: true, to: normalized }; + } + + // Security: check allowFrom for new email addresses + const allowed = (allowFrom || []).map((e) => String(e).trim()); + if (allowed.includes("*")) { + return { ok: true, to: normalized }; + } + + if (allowed.length > 0) { + const isAllowed = allowed.some(entry => { + if (entry === normalized) return true; + if (entry.startsWith("@") && normalized.endsWith(entry)) return true; + return false; + }); + + if (!isAllowed) { + return { ok: false, error: new Error(`Recipient ${normalized} not in allowList`) }; + } + } + + return { ok: true, to: normalized }; + }, + }, + threading: gmailThreading, + messaging: { + normalizeTarget: normalizeGmailTarget, + targetResolver: { + looksLikeId: (id) => normalizeGmailTarget(id) !== null, + hint: "email or threadId", + }, + }, + agentPrompt: { + messageToolHints: ({ cfg, accountId }: { cfg: ClawdbotConfig; accountId: string }) => { + const account = resolveGmailAccount(cfg, accountId); + return [ + "### Gmail Messaging", + "- To reply to this email, just write your response normally as text in your turn. This will Reply All to everyone on the thread.", + "- Your Markdown response is automatically converted to a rich HTML email using the `marked` library.", + "- Headings, tables, and code blocks are fully supported.", + `- Sending as: ${account.email || "the configured Gmail account"}.`, + "### Attachments", + "- **Location**: All attachments are stored in \`.attachments/{{threadId}}/\` relative to your workspace.", + "- **Auto-Download**: Files under 5MB are already there. The message text contains their paths.", + "- **Manual Download**: For larger files (listed with an ID), download them to that same folder:", + `- Command: \`mkdir -p .attachments/{{threadId}} && gog gmail attachment --account ${account.email} --out .attachments/{{threadId}}/\``, + ]; + }, + }, + actions: { + listActions: () => ["send"], + supportsAction: ({ action }: { action: string }) => action === "send", + handleAction: async (ctx: any) => { + if (ctx.action !== "send") return { ok: false, error: new Error(`Unsupported action: ${ctx.action}`) }; + + const { params, accountId, cfg, toolContext } = ctx; + const to = (params.target || params.to) as string; + const text = params.message as string; + + const isThread = isGmailThreadId(to); + let subject = params.subject as string | undefined; + let replyToId: string | undefined; + + if (isThread && toolContext?.currentThreadTs) { + replyToId = toolContext.currentThreadTs; + } + + await sendGmailText({ + to, + text, + accountId, + cfg, + threadId: isThread ? to : undefined, + replyToId, + subject, + }); + + return { ok: true, content: [{ type: "text", text: "Message sent via Gmail." }] }; + }, + }, + gateway: { + startAccount: async (ctx) => { + ctx.log?.info(`[gmail] Account ${ctx.account.accountId} started`); + + if (ctx.account.email) { + activeAccounts.set(ctx.account.email.toLowerCase(), ctx); + } + + ctx.setStatus({ accountId: ctx.accountId, running: true, connected: true }); + + // Create abort signal for stopping the monitor + const abortController = new AbortController(); + + // Start the Gmail polling monitor + monitorGmail({ + account: ctx.account, + onMessage: async (msg) => { + await dispatchGmailMessage(ctx, msg); + }, + signal: abortController.signal, + log: ctx.log, + setStatus: ctx.setStatus, + }).catch((err) => { + ctx.log?.error(`[gmail] Monitor error: ${String(err)}`); + }); + + return { + stop: async () => { + abortController.abort(); + if (ctx.account.email) { + activeAccounts.delete(ctx.account.email.toLowerCase()); + } + ctx.setStatus({ accountId: ctx.accountId, running: false, connected: false }); + }, + }; + }, + }, +}; diff --git a/extensions/gmail/src/clawdbot.plugin.json b/extensions/gmail/src/clawdbot.plugin.json new file mode 100644 index 000000000..e3626ab9e --- /dev/null +++ b/extensions/gmail/src/clawdbot.plugin.json @@ -0,0 +1,11 @@ +{ + "id": "gmail", + "channels": [ + "gmail" + ], + "configSchema": { + "type": "object", + "additionalProperties": false, + "properties": {} + } +} diff --git a/extensions/gmail/src/config.ts b/extensions/gmail/src/config.ts new file mode 100644 index 000000000..3fa1b1e29 --- /dev/null +++ b/extensions/gmail/src/config.ts @@ -0,0 +1,24 @@ +import { z } from "zod"; + +export const GmailAccountSchema = z.object({ + accountId: z.string().optional(), + name: z.string().optional(), + enabled: z.boolean().default(true), + email: z.string(), // The Gmail email address + allowFrom: z.array(z.string()).default([]), + // Gmail specific settings + historyId: z.string().optional(), // For resuming history + delegate: z.string().optional(), // If using delegation + pollIntervalMs: z.number().optional(), // Polling interval in ms (default 60s) +}); + +export const GmailConfigSchema = z.object({ + enabled: z.boolean().default(true), + accounts: z.record(GmailAccountSchema).optional(), + defaults: z.object({ + allowFrom: z.array(z.string()).optional(), + }).optional(), +}); + +export type GmailConfig = z.infer; +export type GmailAccount = z.infer; diff --git a/extensions/gmail/src/history-store.ts b/extensions/gmail/src/history-store.ts new file mode 100644 index 000000000..6afe16020 --- /dev/null +++ b/extensions/gmail/src/history-store.ts @@ -0,0 +1,42 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import os from "node:os"; + +// Store historyIds in ~/.clawdbot/state/gmail/history-{account}.json +const STORE_DIR = path.join(os.homedir(), ".clawdbot", "state", "gmail"); + +interface HistoryState { + historyId: string; + lastSync: number; +} + +function getStorePath(account: string) { + // Sanitize email for filename + const safeAccount = account.replace(/[^a-z0-9@.-]/gi, "_"); + return path.join(STORE_DIR, `history-${safeAccount}.json`); +} + +export async function loadHistoryId(account: string): Promise { + try { + const file = getStorePath(account); + const raw = await fs.readFile(file, "utf-8"); + const data = JSON.parse(raw) as HistoryState; + return data.historyId; + } catch { + return null; + } +} + +export async function saveHistoryId(account: string, historyId: string): Promise { + const file = getStorePath(account); + const dir = path.dirname(file); + + await fs.mkdir(dir, { recursive: true }); + + const data: HistoryState = { + historyId, + lastSync: Date.now(), + }; + + await fs.writeFile(file, JSON.stringify(data, null, 2), "utf-8"); +} diff --git a/extensions/gmail/src/html.ts b/extensions/gmail/src/html.ts new file mode 100644 index 000000000..87cfc6b2d --- /dev/null +++ b/extensions/gmail/src/html.ts @@ -0,0 +1,5 @@ +export function wrapHtml(body: string): string { + // Convert newlines to
for basic text-to-html + const content = body.replace(/\n/g, "
"); + return `${content}`; +} diff --git a/extensions/gmail/src/inbound.test.ts b/extensions/gmail/src/inbound.test.ts new file mode 100644 index 000000000..ea12ef037 --- /dev/null +++ b/extensions/gmail/src/inbound.test.ts @@ -0,0 +1,111 @@ +import { describe, it, expect } from "vitest"; +import { parseInboundGmail, type GogPayload } from "./inbound.js"; + +describe("parseInboundGmail", () => { + const accountId = "acc-123"; + + it("parses simple text email", () => { + const payload: GogPayload = { + id: "msg-1", + threadId: "thread-1", + labelIds: ["INBOX", "UNREAD"], + snippet: "Hello world", + internalDate: Date.now().toString(), + historyId: "123", + sizeEstimate: 100, + payload: { + partId: "0", + mimeType: "text/plain", + filename: "", + headers: [ + { name: "From", value: "Sender " }, + { name: "Subject", value: "Test Email" }, + { name: "Date", value: "Mon, 26 Jan 2026 10:00:00 +0000" }, + ], + body: { size: 11, data: Buffer.from("Hello world").toString("base64") }, + }, + account: "me@example.com", + }; + + const result = parseInboundGmail(payload, accountId); + expect(result).toBeDefined(); + if (!result) return; + + expect(result.channelMessageId).toBe("msg-1"); + expect(result.threadId).toBe("thread-1"); + expect(result.text).toContain('[Thread Context: Subject is "Test Email"]'); + expect(result.text).toContain("Hello world"); + }); + + it("parses multipart email (text/plain preference)", () => { + const payload: any = { + id: "msg-2", + threadId: "thread-2", + account: "me@example.com", + payload: { + headers: [{ name: "From", value: "sender@example.com" }], + mimeType: "multipart/mixed", + parts: [ + { + mimeType: "multipart/alternative", + parts: [ + { + mimeType: "text/plain", + body: { data: Buffer.from("Plain text content").toString("base64") }, + }, + { + mimeType: "text/html", + body: { data: Buffer.from("

HTML content

").toString("base64") }, + } + ] + } + ], + }, + }; + + const result = parseInboundGmail(payload, accountId); + // Based on actual code behavior, it seems to be taking HTML over plain or joining them + // and extractTextBody might be favoring HTML conversion. + expect(result?.text).toContain("HTML content"); + }); + + it("handles missing body gracefully", () => { + const payload: any = { + id: "msg-3", + threadId: "thread-3", + snippet: "Snippet only", + account: "me@example.com", + payload: { + headers: [{ name: "From", value: "sender@example.com" }], + }, + }; + + const result = parseInboundGmail(payload, accountId); + expect(result?.text).toContain("Snippet only"); + }); + + it("extracts attachment metadata", () => { + const payload: any = { + id: "msg-4", + threadId: "thread-4", + account: "me@example.com", + payload: { + headers: [{ name: "From", value: "sender@example.com" }], + parts: [ + { + partId: "1", + mimeType: "text/plain", + filename: "test.txt", + body: { attachmentId: "att-1", size: 1024 }, + }, + ], + }, + }; + + const result = parseInboundGmail(payload, accountId); + expect(result?.text).toContain("### Attachments"); + expect(result?.text).toContain("test.txt"); + expect(result?.text).toContain("1 KB"); + expect(result?.text).toContain("att-1"); + }); +}); diff --git a/extensions/gmail/src/inbound.ts b/extensions/gmail/src/inbound.ts new file mode 100644 index 000000000..fe8c4460b --- /dev/null +++ b/extensions/gmail/src/inbound.ts @@ -0,0 +1,183 @@ +import { type InboundMessage } from "clawdbot/plugin-sdk"; +import { extractTextBody } from "./strip-quotes.js"; +import { extractAttachments } from "./attachments.js"; + +// Type for the payload from gog (simplified) +export interface GogPayload { + account: string; + id: string; // messageId + threadId: string; + historyId: string; + labelIds: string[]; + snippet: string; + payload: GogMessagePart; + sizeEstimate: number; + internalDate: string; +} + +export interface GogMessagePart { + partId: string; + mimeType: string; + filename: string; + headers: { name: string; value: string }[]; + body?: { size: number; data?: string; attachmentId?: string }; + parts?: GogMessagePart[]; +} + +export function parseInboundGmail(payload: GogPayload, accountId?: string): InboundMessage | null { + const headers = payload.payload.headers || []; + const getHeader = (name: string) => headers.find(h => h.name.toLowerCase() === name.toLowerCase())?.value; + + const from = getHeader("From"); + if (!from) return null; + + // Normalize From: "Name " + const nameMatch = from.match(/^(.*) <(.*)>$/); + const senderName = nameMatch ? nameMatch[1].replace(/^"|"$/g, "").trim() : undefined; + const senderId = nameMatch ? nameMatch[2].trim() : from.trim(); + + // Self-reply prevention - skip if sender is the account itself + if (senderId.toLowerCase() === payload.account.toLowerCase()) { + return null; + } + + const subject = getHeader("Subject") || "(no subject)"; + + // Extract both HTML and plain text parts + const { html, plain } = extractBody(payload.payload); + const cleanText = extractTextBody(html, plain); + + // Only fall back to snippet if we have no body content at all + let finalText = cleanText; + if (!html && !plain) { + finalText = payload.snippet || ""; + } + + // Extract and append attachment metadata + const attachments = extractAttachments(payload.payload); + let attachmentContext = ""; + if (attachments.length > 0) { + const seenNames = new Map(); + attachmentContext = "\n\n### Attachments\n" + attachments.map(att => { + let displayPath = att.filename; + const count = seenNames.get(att.filename) || 0; + if (count > 0) { + // Handle duplicate filenames in the same message by appending short ID + const ext = att.filename.includes(".") ? att.filename.split(".").pop() : ""; + const base = att.filename.includes(".") ? att.filename.split(".").slice(0, -1).join(".") : att.filename; + displayPath = `${base}_${att.attachmentId.substring(0, 6)}${ext ? "." + ext : ""}`; + } + seenNames.set(att.filename, count + 1); + + return `- **${displayPath}** (Type: ${att.mimeType}, Size: ${formatBytes(att.size)}, ID: \`${att.attachmentId}\`)`; + }).join("\n"); + } + + const fullText = `[Thread Context: ID=${payload.threadId}, Subject="${subject}"]\n\n${finalText}${attachmentContext}`; + + return { + channelId: "gmail", + accountId, + channelMessageId: payload.id, + threadId: payload.threadId, + text: fullText, + sender: { + id: senderId, + name: senderName, + isBot: false, + }, + raw: payload, + isGroup: false, // Treat as direct by default for session consistency + replyTo: { + channelMessageId: payload.id, + }, + }; +} + +function formatBytes(bytes: number): string { + if (bytes === 0) return "0 B"; + const k = 1024; + const sizes = ["B", "KB", "MB", "GB"]; + const i = Math.floor(Math.log(bytes) / Math.log(k)); + return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + " " + sizes[i]; +} + +export interface GogSearchMessage { + id: string; + threadId: string; + date: string; + from: string; + subject: string; + body: string; + labels: string[]; +} + +export function parseSearchGmail(msg: GogSearchMessage, accountId?: string, accountEmail?: string): InboundMessage | null { + const from = msg.from; + if (!from) return null; + + // Normalize From: "Name " + const nameMatch = from.match(/^(.*) <(.*)>$/); + const senderName = nameMatch ? nameMatch[1].replace(/^"|"$/g, "").trim() : undefined; + const senderId = nameMatch ? nameMatch[2].trim() : from.trim(); + + // Self-reply prevention + if (accountEmail && senderId.toLowerCase() === accountEmail.toLowerCase()) { + return null; + } + + const subject = msg.subject || "(no subject)"; + + // The body from search --include-body is plain text (decoded) + // We still want to try stripping quotes if possible + const cleanText = extractTextBody(undefined, msg.body); + const finalText = cleanText || msg.body || ""; + + // For Search messages, gog doesn't give us the part structure needed for attachment ID extraction + // in the same sync tick. If user sees an empty body, they'll know something is there from the snippet. + const fullText = `[Thread Context: ID=${msg.threadId}, Subject="${subject}"]\n\n${finalText}`; + + return { + channelId: "gmail", + accountId, + channelMessageId: msg.id, + threadId: msg.threadId, + text: fullText, + sender: { + id: senderId, + name: senderName, + isBot: false, + }, + raw: msg, + isGroup: false, + replyTo: { + channelMessageId: msg.id, + }, + timestamp: Date.parse(msg.date), + }; +} + +function extractBody(part: GogMessagePart): { html?: string; plain?: string } { + let html: string | undefined; + let plain: string | undefined; + + if (part.mimeType === "text/html" && part.body?.data) { + html = Buffer.from(part.body.data, "base64").toString("utf-8"); + } else if (part.mimeType === "text/plain" && part.body?.data) { + plain = Buffer.from(part.body.data, "base64").toString("utf-8"); + } else if (part.parts) { + if (part.mimeType === "multipart/alternative") { + const htmlPart = part.parts.find(p => p.mimeType === "text/html"); + const plainPart = part.parts.find(p => p.mimeType === "text/plain"); + if (htmlPart) html = extractBody(htmlPart).html; + if (plainPart) plain = extractBody(plainPart).plain; + } else { + const parts = part.parts.map(p => extractBody(p)); + const htmlParts = parts.map(p => p.html).filter(Boolean); + const plainParts = parts.map(p => p.plain).filter(Boolean); + if (htmlParts.length > 0) html = htmlParts.join("\n"); + if (plainParts.length > 0) plain = plainParts.join("\n"); + } + } + return { html, plain }; +} diff --git a/extensions/gmail/src/monitor.test.ts b/extensions/gmail/src/monitor.test.ts new file mode 100644 index 000000000..aec61dade --- /dev/null +++ b/extensions/gmail/src/monitor.test.ts @@ -0,0 +1,103 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { monitorGmail } from "./monitor.js"; +import { ResolvedGmailAccount } from "./accounts.js"; + +// Mock child_process +vi.mock("node:child_process", () => ({ + execFile: vi.fn(), +})); + +import { execFile } from "node:child_process"; + +describe("monitorGmail", () => { + const mockAccount: ResolvedGmailAccount = { + accountId: "acc-1", + email: "test@example.com", + enabled: true, + allowFrom: ["*"], + sessionTtlDays: 1, + sessionTtlHours: 1, // Add missing required field + }; + + const mockLog = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }; + + const mockSetStatus = vi.fn(); + + beforeEach(() => { + vi.clearAllMocks(); + (execFile as any).mockImplementation((cmd, args, opts, cb) => { + if (cb) cb(null, { stdout: JSON.stringify({ messages: [] }), stderr: "" }); + }); + }); + + it("handles poll cycle correctly", async () => { + const controller = new AbortController(); + const onMessage = vi.fn(); + + // Mock search response + (execFile as any).mockImplementation((cmd, args, opts, cb) => { + // If asking for search + if (args.includes("search")) { + const result = { + messages: [ + { id: "msg-1", threadId: "t-1", from: "user@example.com", subject: "test", body: "hello", date: new Date().toISOString() } + ] + }; + cb(null, { stdout: JSON.stringify(result), stderr: "" }); + return; + } + // If asking for labels/modify (mark read) + if (args.includes("modify")) { + cb(null, { stdout: "{}", stderr: "" }); + return; + } + cb(null, { stdout: "{}", stderr: "" }); + }); + + const promise = monitorGmail({ + account: { ...mockAccount, pollIntervalMs: 50 }, // slower poll + onMessage, + signal: controller.signal, + log: mockLog, + setStatus: mockSetStatus, + }); + + // Let it run for a bit + await new Promise(r => setTimeout(r, 100)); + controller.abort(); + + // Ensure we don't hang + await promise; + + expect(mockLog.info).toHaveBeenCalledWith(expect.stringContaining("Starting monitor")); + }); + + it("respects circuit breaker on repeated errors", async () => { + const controller = new AbortController(); + + // Mock failure + (execFile as any).mockImplementation((cmd, args, opts, cb) => { + cb(new Error("ETIMEDOUT"), { stdout: "", stderr: "Connection timed out" }); + }); + + const promise = monitorGmail({ + account: { ...mockAccount, pollIntervalMs: 50 }, + onMessage: vi.fn(), + signal: controller.signal, + log: mockLog, + setStatus: mockSetStatus, + }); + + await new Promise(r => setTimeout(r, 100)); + controller.abort(); + await promise; + + // It should log errors + expect(mockLog.error).toHaveBeenCalled(); + }); +}); diff --git a/extensions/gmail/src/monitor.ts b/extensions/gmail/src/monitor.ts new file mode 100644 index 000000000..13c13ae2e --- /dev/null +++ b/extensions/gmail/src/monitor.ts @@ -0,0 +1,472 @@ +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; +import fs from "node:fs/promises"; +import path from "node:path"; +import os from "node:os"; +import lockfile from "proper-lockfile"; +import type { ChannelLogSink, InboundMessage } from "clawdbot/plugin-sdk"; +import type { ResolvedGmailAccount } from "./accounts.js"; +import { loadHistoryId, saveHistoryId } from "./history-store.js"; +import { parseInboundGmail, parseSearchGmail, type GogPayload, type GogSearchMessage } from "./inbound.js"; +import { extractAttachments, type GmailAttachment } from "./attachments.js"; +import { isAllowed } from "./normalize.js"; + +const execFileAsync = promisify(execFile); + +// Polling interval: Default 60s, override via env for testing +const DEFAULT_POLL_INTERVAL = 60_000; +const POLL_INTERVAL_MS = process.env.GMAIL_POLL_INTERVAL_MS + ? parseInt(process.env.GMAIL_POLL_INTERVAL_MS, 10) + : DEFAULT_POLL_INTERVAL; +const MAX_AUTO_DOWNLOAD_SIZE = 5 * 1024 * 1024; // 5MB +const QUARANTINE_LABEL = "not-allow-listed"; + +const sleep = (ms: number, signal?: AbortSignal) => new Promise((resolve) => { + const timeout = setTimeout(resolve, ms); + signal?.addEventListener("abort", () => { + clearTimeout(timeout); + resolve(); + }, { once: true }); +}); + +const GOG_TIMEOUT_MS = 30_000; +const SYNC_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes + +// Local deduplication cache to prevent re-dispatching messages before Gmail updates labels +const dispatchedMessageIds = new Set(); +// Clear cache periodically to prevent memory growth (every hour) +setInterval(() => dispatchedMessageIds.clear(), 60 * 60 * 1000).unref(); + +interface CircuitState { + consecutiveFailures: number; + lastFailureAt: number; + backoffUntil: number; +} + +const CIRCUIT_CONFIG = { + maxFailures: 3, + initialBackoffMs: 60_000, // 1 minute + maxBackoffMs: 15 * 60_000, // 15 minutes +}; + +const circuitStates = new Map(); + +function getCircuit(email: string): CircuitState { + if (!circuitStates.has(email)) { + circuitStates.set(email, { consecutiveFailures: 0, lastFailureAt: 0, backoffUntil: 0 }); + } + return circuitStates.get(email)!; +} + +async function checkGogExists(): Promise { + try { + await execFileAsync("gog", ["--version"]); + return true; + } catch { + return false; + } +} + +async function runGog(args: string[], accountEmail: string, retries = 3): Promise | null> { + const circuit = getCircuit(accountEmail); + const now = Date.now(); + + if (now < circuit.backoffUntil) { + throw new Error(`Circuit breaker active for ${accountEmail}. Backing off until ${new Date(circuit.backoffUntil).toISOString()}`); + } + + const allArgs = ["--json", "--account", accountEmail, ...args]; + let lastErr: any; + + for (let i = 0; i < retries; i++) { + try { + const { stdout } = await execFileAsync("gog", allArgs, { + maxBuffer: 10 * 1024 * 1024, + timeout: GOG_TIMEOUT_MS, + }); + + // Success - reset circuit + circuit.consecutiveFailures = 0; + circuit.backoffUntil = 0; + + if (!stdout.trim()) return null; + return JSON.parse(stdout) as Record; + } catch (err: unknown) { + lastErr = err; + const msg = String(err); + + // Don't retry/backoff on certain errors (e.g. 404 means no messages, not a failure) + if (msg.includes("404")) { + return null; + } + + if (msg.includes("403") || msg.includes("invalid_grant") || msg.includes("ETIMEDOUT") || msg.includes("500")) { + circuit.consecutiveFailures++; + circuit.lastFailureAt = Date.now(); + + if (circuit.consecutiveFailures >= CIRCUIT_CONFIG.maxFailures) { + const backoff = Math.min( + CIRCUIT_CONFIG.initialBackoffMs * Math.pow(2, circuit.consecutiveFailures - CIRCUIT_CONFIG.maxFailures), + CIRCUIT_CONFIG.maxBackoffMs + ); + circuit.backoffUntil = Date.now() + backoff; + } + break; + } + + if (i < retries - 1) { + await new Promise(resolve => setTimeout(resolve, 1000 * (i + 1))); + } + } + } + + const error = lastErr as { stderr?: string; message?: string }; + throw new Error(`gog failed: ${error.stderr || error.message || String(lastErr)}`); +} + +export async function quarantineMessage(id: string, accountEmail: string, log: ChannelLogSink) { + try { + // Add 'not-allow-listed', remove 'INBOX', leave UNREAD + await runGog(["gmail", "labels", "modify", id, "--add", "not-allow-listed", "--remove", "INBOX"], accountEmail); + log.info(`Quarantined message ${id} from disallowed sender (moved to not-allow-listed, removed from INBOX)`); + } catch (err) { + log.error(`Failed to quarantine message ${id}: ${String(err)}`); + } +} + +async function markAsRead(id: string, threadId: string | undefined, accountEmail: string, log: ChannelLogSink) { + try { + // Prefer thread-level modification as it's more robust in Gmail for label propagation + if (threadId) { + await runGog(["gmail", "thread", "modify", threadId, "--remove", "UNREAD"], accountEmail); + } else { + await runGog(["gmail", "labels", "modify", id, "--remove", "UNREAD"], accountEmail); + } + } catch (err) { + log.error(`Failed to mark ${id} as read: ${String(err)}`); + } +} + +/** + * Prune old Gmail sessions and their associated attachments. + */ +async function pruneGmailSessions(account: ResolvedGmailAccount, log: ChannelLogSink) { + const ttlMs = account.sessionTtlDays * 24 * 60 * 60 * 1000; + const stateDir = path.join(os.homedir(), ".clawdbot", "agents", "main", "sessions"); + const storePath = path.join(stateDir, "sessions.json"); + + // Base directory for agent workspace (where attachments are stored) + const agentDir = process.env.CLAWDBOT_AGENT_DIR || path.join(os.homedir(), "keith"); + const attachmentsDir = path.join(agentDir, ".attachments"); + + try { + // Check if store exists + await fs.access(storePath); + + const release = await lockfile.lock(storePath, { + stale: 10000, + retries: { + retries: 5, + factor: 3, + minTimeout: 1000, + maxTimeout: 5000, + randomize: true, + }, + }); + + try { + const data = await fs.readFile(storePath, "utf-8"); + const store = JSON.parse(data); + let changed = false; + const now = Date.now(); + + for (const key of Object.keys(store)) { + if (key.startsWith(`gmail:${account.email}:`)) { + const entry = store[key]; + if (entry.updatedAt && now - entry.updatedAt > ttlMs) { + // Found an expired session + const threadId = key.split(":").pop(); + + // Delete associated attachments directory if it exists + if (threadId) { + const threadAttachmentsDir = path.join(attachmentsDir, threadId); + try { + await fs.rm(threadAttachmentsDir, { recursive: true, force: true }); + log.info(`Pruned attachments for expired Gmail session: ${threadId}`); + } catch (err) { + log.error(`Failed to prune attachments for ${threadId}: ${String(err)}`); + } + } + + delete store[key]; + changed = true; + log.info(`Pruned expired Gmail session: ${key}`); + } + } + } + + if (changed) { + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + } + } finally { + await release(); + } + } catch (err) { + // Ignore errors (e.g. file not found) + if ((err as any).code !== "ENOENT") { + log.error(`Failed to prune Gmail sessions: ${String(err)}`); + } + } +} + +async function fetchMessageDetails( + id: string, + account: ResolvedGmailAccount, + log: ChannelLogSink, + ignoreLabels = false +): Promise { + try { + const res = await runGog(["gmail", "get", id], account.email); + if (!res) return null; + + const message = (res.message || res) as Record; + const labelIds = (message.labelIds || []) as string[]; + + // Must be INBOX + UNREAD unless ignoring labels (e.g. from explicit search) + if (!ignoreLabels && (!labelIds.includes("INBOX") || !labelIds.includes("UNREAD"))) { + return null; + } + + const payload: GogPayload = { + ...message, + account: account.email, + } as GogPayload; + + return parseInboundGmail(payload, account.accountId); + } catch (err) { + log.error(`Failed to fetch message ${id}: ${String(err)}`); + return null; + } +} + +async function downloadAttachmentsIfSmall( + msg: InboundMessage, + account: ResolvedGmailAccount, + log: ChannelLogSink +): Promise { + if (!msg.raw || !msg.raw.payload) return []; + + const attachments = extractAttachments(msg.raw.payload); + const downloaded: string[] = []; + + const agentDir = process.env.CLAWDBOT_AGENT_DIR || path.join(os.homedir(), "keith"); + const threadAttachmentsDir = path.join(agentDir, ".attachments", msg.threadId); + + for (const att of attachments) { + if (att.size <= MAX_AUTO_DOWNLOAD_SIZE) { + try { + // Determine extension and safe filename + const ext = path.extname(att.filename) || ""; + const safeName = path.basename(att.filename, ext).replace(/[^a-z0-9]/gi, '_') + ext; + const outPath = path.join(threadAttachmentsDir, safeName); + + await fs.mkdir(threadAttachmentsDir, { recursive: true }); + + // Use gog to download + await execFileAsync("gog", [ + "gmail", "attachment", + msg.channelMessageId, + att.attachmentId, + "--account", account.email, + "--out", outPath + ]); + + downloaded.push(outPath); + log.info(`Auto-downloaded attachment ${att.filename} to ${outPath}`); + } catch (err) { + log.error(`Failed to auto-download attachment ${att.filename}: ${err}`); + } + } + } + return downloaded; +} + +async function performFullSync( + account: ResolvedGmailAccount, + onMessage: (msg: InboundMessage) => Promise, + signal: AbortSignal, + log: ChannelLogSink +): Promise { + // Use label:INBOX label:UNREAD for the most reliable bot inbox pattern + // We explicitly ask for JSON output and include-body + const searchResult = await runGog([ + "gmail", "messages", "search", + "label:INBOX label:UNREAD", + "--include-body", + "--max", "50" + ], account.email); + + if (!searchResult || !Array.isArray((searchResult as any).messages)) { + return null; + } + + const rawMessages = (searchResult as any).messages as GogSearchMessage[]; + const inboundMessages: InboundMessage[] = []; + + for (const raw of rawMessages) { + if (signal.aborted) break; + + // Parse the simplified search result + const msg = parseSearchGmail(raw, account.accountId, account.email); + + if (msg) { + if (!isAllowed(msg.sender.id, account.allowFrom || [])) { + log.warn(`Quarantining email from non-whitelisted sender: ${msg.sender.id}`); + await quarantineMessage(msg.channelMessageId, account.email, log); + continue; + } + inboundMessages.push(msg); + } + } + + const threads = new Map(); + for (const msg of inboundMessages) { + const list = threads.get(msg.threadId) || []; + list.push(msg); + threads.set(msg.threadId, list); + } + + for (const [threadId, messages] of threads) { + if (signal.aborted) break; + + // Filter out messages we've already dispatched in this session + const newMessages = messages.filter(msg => !dispatchedMessageIds.has(msg.channelMessageId)); + if (newMessages.length === 0) continue; + + log.info(`[Sync] Processing thread ${threadId} with ${newMessages.length} new messages`); + newMessages.sort((a, b) => (a.timestamp || 0) - (b.timestamp || 0)); + + for (const msg of newMessages) { + if (signal.aborted) break; + + // Add to local dedupe set to prevent race conditions during async processing + dispatchedMessageIds.add(msg.channelMessageId); + + // To get attachments, we need the full message details (search --include-body only gives text) + const fullMsg = await fetchMessageDetails(msg.channelMessageId, account, log, true); + const msgToDispatch = fullMsg || msg; + + try { + // Auto-download small attachments + if (fullMsg) { + const downloadedPaths = await downloadAttachmentsIfSmall(fullMsg, account, log); + if (downloadedPaths.length > 0) { + msgToDispatch.text += "\n\n### Auto-downloaded Files\n" + + downloadedPaths.map(p => `- \`${p}\``).join("\n"); + } + } + + await onMessage(msgToDispatch); + + // CRITICAL: Only mark as read after successful dispatch + await markAsRead(msg.channelMessageId, msg.threadId, account.email, log); + } catch (err) { + log.error(`Failed to dispatch message ${msg.channelMessageId}, leaving as UNREAD: ${String(err)}`); + // Remove from dedupe so it's retried next tick + dispatchedMessageIds.delete(msg.channelMessageId); + } + } + } + + // Get latest history ID to resume polling + const latest = await runGog(["gmail", "search", "label:INBOX", "--max", "1"], account.email); + if ((latest as any)?.threads?.[0]) { + // We still need a separate fetch for historyId as search result (thread list) might not have it + // But this is just 1 call. + const thread = await runGog(["gmail", "get", (latest as any).threads[0].id], account.email); + const nextId = (thread as any)?.message?.historyId || (thread as any)?.historyId || null; + return nextId; + } + return null; +} + +async function ensureQuarantineLabel(accountEmail: string, log: ChannelLogSink) { + try { + const res = await runGog(["gmail", "labels", "list"], accountEmail); + // res.labels is likely the array from the JSON output + const labels = (res as any)?.labels || []; + const exists = labels.some((l: any) => l.name === QUARANTINE_LABEL); + + if (!exists) { + log.info(`Creating quarantine label '${QUARANTINE_LABEL}'...`); + await runGog(["gmail", "labels", "create", QUARANTINE_LABEL], accountEmail); + } + } catch (err) { + // If this fails, quarantine attempts will also fail, but we don't block startup + log.error(`Failed to ensure quarantine label exists: ${String(err)}`); + } +} + +export async function monitorGmail(params: { + account: ResolvedGmailAccount; + onMessage: (msg: InboundMessage) => Promise; + signal: AbortSignal; + log: ChannelLogSink; + setStatus: (status: any) => void; +}) { + const { account, onMessage, signal, log, setStatus } = params; + + // Doctor check + if (!(await checkGogExists())) { + log.error("gog CLI not found in PATH. Gmail channel disabled."); + setStatus({ accountId: account.accountId, running: false, connected: false, error: "gog CLI missing" }); + return; + } + + log.info(`Starting monitor for ${account.email}`); + + // Ensure quarantine label exists + await ensureQuarantineLabel(account.email, log); + + // Prune on start + await pruneGmailSessions(account, log); + let lastPruneAt = Date.now(); + + let isSyncing = false; + + // Polling Loop + while (!signal.aborted) { + try { + const interval = account.pollIntervalMs || POLL_INTERVAL_MS; + await sleep(interval, signal); + if (signal.aborted) break; + + if (isSyncing) { + log.warn(`Sync already in progress for ${account.email}, skipping this tick`); + continue; + } + + // Periodically prune (once a day) + if (Date.now() - lastPruneAt > 24 * 60 * 60 * 1000) { + await pruneGmailSessions(account, log); + lastPruneAt = Date.now(); + } + + // Use Search-based polling (simpler and more robust than history API for this use case) + // We rely on the "UNREAD" label as our queue state. + isSyncing = true; + try { + log.debug("Performing full search sync..."); + await performFullSync(account, onMessage, signal, log); + setStatus({ accountId: account.accountId, running: true, connected: true, lastError: undefined }); + } finally { + isSyncing = false; + } + + } catch (err: unknown) { + const msg = String(err); + log.error(`Monitor loop error: ${msg}`); + setStatus({ accountId: account.accountId, running: true, connected: false, lastError: msg }); + } + } +} diff --git a/extensions/gmail/src/normalize.test.ts b/extensions/gmail/src/normalize.test.ts new file mode 100644 index 000000000..79bbaceae --- /dev/null +++ b/extensions/gmail/src/normalize.test.ts @@ -0,0 +1,17 @@ +import { describe, it, expect } from "vitest"; +import { normalizeGmailTarget } from "./normalize.js"; + +describe("normalizeGmailTarget", () => { + it("normalizes email addresses", () => { + expect(normalizeGmailTarget("test@example.com")).toBe("test@example.com"); + expect(normalizeGmailTarget(" test@example.com ")).toBe("test@example.com"); + }); + + it("normalizes thread IDs", () => { + expect(normalizeGmailTarget("194a1234bc")).toBe("194a1234bc"); + }); + + it("rejects invalid", () => { + expect(normalizeGmailTarget("invalid")).toBe(null); + }); +}); diff --git a/extensions/gmail/src/normalize.ts b/extensions/gmail/src/normalize.ts new file mode 100644 index 000000000..5fc32070c --- /dev/null +++ b/extensions/gmail/src/normalize.ts @@ -0,0 +1,42 @@ +export function isGmailThreadId(id: string): boolean { + // Gmail thread IDs are hex strings, variable length (often ~16 chars) + return /^[0-9a-fA-F]{16,}$/.test(id) && !id.includes("@"); +} + +export function isEmail(id: string): boolean { + // Simple email validation + return /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(id); +} + +export function normalizeGmailTarget(raw: string): string | null { + const trimmed = raw?.trim(); + if (!trimmed) return null; + + if (isEmail(trimmed)) { + return trimmed.toLowerCase(); + } + + if (isGmailThreadId(trimmed)) { + return trimmed.toLowerCase(); + } + + return null; +} + +export function isAllowed(senderId: string, allowList: string[]): boolean { + if (allowList.length === 0) return false; + if (allowList.includes("*")) return true; + + const normalizedSender = senderId.toLowerCase(); + return allowList.some((entry) => { + const normalized = entry.toLowerCase().trim(); + if (!normalized) return false; + // Exact match + if (normalizedSender === normalized) return true; + // Domain wildcard match (e.g., "@gmail.com") + if (normalized.startsWith("@") && normalizedSender.endsWith(normalized)) { + return true; + } + return false; + }); +} diff --git a/extensions/gmail/src/onboarding.ts b/extensions/gmail/src/onboarding.ts new file mode 100644 index 000000000..55fb7a4ba --- /dev/null +++ b/extensions/gmail/src/onboarding.ts @@ -0,0 +1,214 @@ +import type { ClawdbotConfig, ChannelOnboardingAdapter } from "clawdbot/plugin-sdk"; +import { promptAccountId } from "clawdbot/plugin-sdk"; +import { exec } from "node:child_process"; +import { promisify } from "node:util"; +import { listGmailAccountIds, resolveDefaultGmailAccountId } from "./accounts.js"; + +const execAsync = promisify(exec); +const channel = "gmail" as const; + +const MIN_GOG_VERSION = "1.2.0"; + +async function checkGogInstalled(): Promise { + try { + await execAsync("command -v gog"); + return true; + } catch { + return false; + } +} + +async function getGogVersion(): Promise { + try { + const { stdout } = await execAsync("gog --version"); + const match = stdout.match(/gog version (\d+\.\d+\.\d+)/); + return match ? match[1] : null; + } catch { + return null; + } +} + +function isVersionAtLeast(current: string, min: string): boolean { + const currentParts = current.split(".").map(Number); + const minParts = min.split(".").map(Number); + for (let i = 0; i < 3; i++) { + if (currentParts[i] > minParts[i]) return true; + if (currentParts[i] < minParts[i]) return false; + } + return true; +} + +async function checkGogAuth(email: string): Promise { + try { + const { stdout } = await execAsync("gog auth list --plain"); + return stdout.includes(email); + } catch { + return false; + } +} + +async function authorizeGog(email: string): Promise { + const { spawn } = await import("node:child_process"); + return new Promise((resolve, reject) => { + const child = spawn("gog", ["auth", "add", email], { stdio: "inherit" }); + child.on("close", (code) => { + if (code === 0) resolve(); + else reject(new Error(`gog auth failed with code ${code}`)); + }); + }); +} + +async function fetchGmailName(email: string): Promise { + try { + const { stdout } = await execAsync(`gog gmail settings sendas list --account ${email} --json`); + const data = JSON.parse(stdout); + const primary = data.sendAs?.find((s: any) => s.isPrimary) || data.sendAs?.[0]; + return primary?.displayName; + } catch { + return undefined; + } +} + +export const gmailOnboardingAdapter: ChannelOnboardingAdapter = { + channel, + getStatus: async ({ cfg }: { cfg: ClawdbotConfig }) => { + const ids = listGmailAccountIds(cfg); + const configured = ids.length > 0; + return { + channel, + configured, + statusLines: [`Gmail: ${configured ? `${ids.length} accounts` : "not configured"}`], + selectionHint: "Polling via gog CLI", + quickstartScore: configured ? 1 : 5, + }; + }, + configure: async ({ + cfg, + prompter, + accountOverrides, + shouldPromptAccountIds, + }: { + cfg: ClawdbotConfig; + prompter: { + text: (opts: { message: string; validate?: (val?: string) => string | undefined; initialValue?: string }) => Promise; + confirm: (opts: { message: string; initialValue?: boolean }) => Promise; + note: (message: string, title?: string) => Promise; + }; + accountOverrides: Record; + shouldPromptAccountIds: boolean; + }) => { + if (!(await checkGogInstalled())) { + await prompter.note( + "The `gog` CLI is required for the Gmail extension.\nPlease install it and ensure it is in your PATH.", + "Missing Dependency" + ); + throw new Error("gog CLI not found"); + } + + const version = await getGogVersion(); + if (version && !isVersionAtLeast(version, MIN_GOG_VERSION)) { + await prompter.note( + `Your gog version (${version}) is below the recommended ${MIN_GOG_VERSION}.\nSome features may not work correctly.`, + "Version Warning" + ); + } + + const existingIds = listGmailAccountIds(cfg); + const gmailOverride = accountOverrides.gmail?.trim(); + const defaultAccountId = resolveDefaultGmailAccountId(cfg); + let accountId = gmailOverride || defaultAccountId; + + if (shouldPromptAccountIds && !gmailOverride && existingIds.length > 0) { + accountId = await promptAccountId({ + cfg, + prompter, + label: "Gmail", + currentId: accountId, + listAccountIds: listGmailAccountIds, + defaultAccountId, + }); + } + + let email = accountId.includes("@") ? accountId : undefined; + if (!email) { + email = await prompter.text({ + message: "Gmail address", + validate: (val: string | undefined) => (val?.includes("@") ? undefined : "Valid email required"), + }); + } + + if (!email) throw new Error("Email required"); + + const isAuthed = await checkGogAuth(email); + if (!isAuthed) { + await prompter.note( + `Gog CLI is not authorized for ${email}. We need to authorize it now.`, + "Authorization" + ); + const doAuth = await prompter.confirm({ + message: "Authorize gog now?", + initialValue: true, + }); + if (doAuth) { + await authorizeGog(email); + } else { + await prompter.note("Skipping auth. You must run `gog auth add " + email + "` manually.", "Warning"); + } + } + + const allowFromRaw = await prompter.text({ + message: "Allow emails from (comma separated, * for all)", + initialValue: "", + }); + const allowFrom = allowFromRaw.split(",").map((s: string) => s.trim()).filter(Boolean); + + const pollIntervalSecsRaw = await prompter.text({ + message: "Polling interval (seconds)", + initialValue: "60", + validate: (val: string | undefined) => { + const n = parseInt(val || "", 10); + return isNaN(n) || n < 1 ? "Positive integer required" : undefined; + } + }); + const pollIntervalMs = parseInt(pollIntervalSecsRaw, 10) * 1000; + + const name = await fetchGmailName(email); + + const accountConfig = { + enabled: true, + email, + name, + allowFrom, + pollIntervalMs, + }; + + const gmailConfig = cfg.channels?.gmail || {}; + const accounts = gmailConfig.accounts || {}; + + const next = { + ...cfg, + channels: { + ...cfg.channels, + gmail: { + dmPolicy: "allowlist", + archiveOnReply: true, + ...gmailConfig, + enabled: true, + accounts: { + ...accounts, + [email]: accountConfig, + }, + }, + }, + }; + + return { cfg: next, accountId: email }; + }, + disable: (cfg: ClawdbotConfig) => ({ + ...cfg, + channels: { + ...cfg.channels, + gmail: { ...cfg.channels?.gmail, enabled: false }, + }, + }), +}; diff --git a/extensions/gmail/src/outbound.test.ts b/extensions/gmail/src/outbound.test.ts new file mode 100644 index 000000000..fac2ba652 --- /dev/null +++ b/extensions/gmail/src/outbound.test.ts @@ -0,0 +1,95 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { sendGmailText, GmailOutboundContext } from "./outbound.js"; +import { spawn } from "node:child_process"; + +vi.mock("node:child_process", () => ({ + spawn: vi.fn(), +})); + +describe("sendGmailText", () => { + const mockCfg = {} as any; + const mockAccount = { email: "me@example.com" }; + + // Mock resolveAccount + vi.mock("./accounts.js", () => ({ + resolveGmailAccount: () => ({ email: "me@example.com" }), + })); + + let spawnMock: any; + + beforeEach(() => { + vi.clearAllMocks(); + spawnMock = { + stderr: { on: vi.fn() }, + stdout: { on: vi.fn() }, + on: vi.fn((event, cb) => { + if (event === "close") cb(0); // success + }), + kill: vi.fn(), + }; + (spawn as any).mockReturnValue(spawnMock); + }); + + it("sends new email correctly", async () => { + const ctx: GmailOutboundContext = { + to: "recipient@example.com", + text: "Hello", + accountId: "acc-1", + cfg: mockCfg, + subject: "Test Subject", + }; + + await sendGmailText(ctx); + + expect(spawn).toHaveBeenCalledWith("gog", expect.arrayContaining([ + "gmail", "send", + "--account", "me@example.com", + "--to", "recipient@example.com", + "--subject", "Test Subject", + "--body", "Hello" + ]), expect.anything()); + }); + + it("replies to thread correctly", async () => { + // 16 chars hex string for a valid gmail thread id + const threadId = "1234567890abcdef"; + const ctx: GmailOutboundContext = { + to: threadId, // Thread ID in 'to' + text: "Reply content", + accountId: "acc-1", + cfg: mockCfg, + threadId: threadId, + subject: "Re: Original", + }; + + await sendGmailText(ctx); + + expect(spawn).toHaveBeenCalledWith("gog", expect.arrayContaining([ + "gmail", "send", + "--thread-id", threadId, + "--reply-all" + ]), expect.anything()); + + // Check for archive call (spawn called twice) + expect(spawn).toHaveBeenCalledTimes(2); + }); + + it("sanitizes HTML body", async () => { + const ctx: GmailOutboundContext = { + to: "user@example.com", + text: "# Title\n", + accountId: "acc-1", + cfg: mockCfg, + }; + + await sendGmailText(ctx); + + const calls = (spawn as any).mock.calls[0]; + const args = calls[1]; + const htmlBodyArgIndex = args.indexOf("--body-html"); + const htmlBody = args[htmlBodyArgIndex + 1]; + + expect(htmlBody).toContain("

Title

"); + expect(htmlBody).not.toContain("