integration-webhooks

# Integration Webhooks ## Rule Integration webhooks (Slack, Telegram, WhatsApp, email, Google Docs, etc.) must **enqueue work to SQL and return 200 immediately**, then process the work in a **separate fresh function execution** kicked off by a self-fired HTTP POST. SQL is always the source of truth. Long-lived hosts use an in-process retry loop as a best-effort safety net; serverless deployments must use an external durable sweep when recovery cannot depend on a process staying alive. Do not run agent loops inside the webhook handler itself. Do not rely on fire-and-forget `Promise`s after `return`ing from a serverless handler — they get killed when the function freezes. ## Why Messaging platforms expect a 200 response within a tight window — Slack will retry after 3 seconds, and a retried event triggers duplicate agent runs. At the same time, an agent loop replying to the message can take 30–60+ seconds because it may make multiple LLM calls and tool calls. Past attempts that don't work cross-host: - **Fire-and-forget `Promise.then(...)` after returning** — Lambda/Vercel/CF freeze the execution context the moment the response goes out. The promise is silently killed, the user gets no reply, and there's no error in the logs. - **A platform background function as the queue** — host-specific execution is useful as a delivery mechanism, but it cannot replace the SQL row or atomic claim shared by every host. - **Cloudflare `event.waitUntil()`** — CF Workers only, not portable. - **Vercel Fluid / `after()`** — Vercel-only, gated behind specific runtimes. - **A long-lived in-process queue** — fine on a single Node box, but on serverless every cold start gets a fresh queue and any pending work is lost. The only universal answer: **persist the work, then trigger a brand new function execution to do it.** SQL is the queue, a self-webhook is the trigger, and a recurring job is the safety net. ## The Flow ``` ┌──────────┐ 1. POST /integrations/:platform/webhook │ Platform │────────────────────────────────────────────►┌──────────────────┐ └──────────┘ │ Webhook handler │ │ (function exec 1)│ └──────────────────┘ │ 2. INSERT INTO integration_pending_tasks (status='pending', payload=...) │ 3. dispatch POST /integrations/process-task — portable self-fire by default — acknowledged background handoff when enabled │ 4. return 200 to platform ◄───────────┘ ┌──────────────────┐ 5. POST arrives at processor │ Processor │ (separate fresh function) │ (function exec 2)│ └──────────────────┘ │ 6. claimPendingTask(id) → status='processing' 7. runAgentLoop(...) — full timeout budget here 8. adapter.sendResponse(...) back to platform 9. markTaskCompleted(id) ┌──────────────────────────────────────────────┐ │ Recovery sweep (every 60s) — safety net │ │ Re-fires processor for tasks stuck in │ │ 'pending' or 'processing' beyond timeout. │ │ Caps retries at 3 then marks 'failed'. │ └──────────────────────────────────────────────┘ ``` The webhook handler does as little as possible. The fresh function execution that handles `_process-task` gets its own full timeout budget for the agent loop. ## Key Files | File | Purpose | | ----------------------------------------------------------------------- | ---------------------------------------------------------------------- | | `packages/core/src/integrations/plugin.ts` | Mounts `/_agent-native/integrations/*` routes | | `packages/core/src/integrations/webhook-handler.ts` | Verifies signature, parses, enqueues task, fires processor | | `packages/core/src/integrations/pending-tasks-store.ts` | SQL queue: `insertPendingTask`, `claimPendingTask`, `markTaskCompleted`, `markTaskFailed` | | `packages/core/src/integrations/pending-tasks-retry-job.ts` | Recurring retry sweep (`startPendingTasksRetryJob`, `retryStuckPendingTasks`) | | `packages/core/src/integrations/integration-durable-dispatch.ts` | Default-off acknowledged dispatch, scoped rollout, and outcome recording | | `packages/core/src/integrations/types.ts` | `PlatformAdapter`, `IncomingMessage`, `OutgoingMessage` | | `packages/core/src/integrations/adapters/{slack,telegram,whatsapp,email,google-docs}.ts` | One adapter per platform | ## Routes All under `/_agent-native/integrations/`: | Method | Path | Purpose | | ------ | -------------------------- | ------------------------------------------------------------- | | POST | `/:platform/webhook` | Platform pings this. Verifies, enqueues, returns 200 quickly. | | POST | `/process-task` | Processor target. Claims a task and runs the agent loop. | | POST | `/retry-stuck-tasks` | Signed, bounded recovery sweep for durable schedulers. | | GET | `/status` | All integrations status (settings UI). | | GET | `/:platform/status` | One platform's status. | | POST | `/:platform/enable` | Enable an integration. | | POST | `/:platform/disable` | Disable an integration. | | POST | `/:platform/setup` | Platform-specific setup (e.g. Telegram webhook registration). | ## Settings › Channels Each app's Channels page (`packages/core/src/client/integrations/ChannelsPage.tsx`) lists the channels the deployment mounts. It reads and changes them only through two core actions, which the agent calls too: - `list-messaging-channels` (GET): each mounted adapter's state, webhook URL, and its own `getRequiredEnvKeys()`. For owners and admins each key also says where its value comes from (`saved`, `environment`, `elsewhere`, or null). - `manage-messaging-channel`: `save-credentials`, `enable`, `disable`, `remove-credentials`, `register-webhook`. Owners and admins only, checked in `integrations/channel-settings.ts`. Rules that keep it working in every app: - Credentials are saved as workspace-scoped secrets for the org (`solo:<email>` without one), not through `/env-vars`, so no template has to declare channel keys. `resolveSecret` finds them in Settings requests and in webhooks, which run as the admin who turned the channel on. - An adapter key the runtime reads only from `process.env` sets `deploymentOnly: true` in `getRequiredEnvKeys()` (Google Docs' service account key). The page then points at the deployment instead of offering an input, and `save-credentials` refuses it. - The plugin publishes its adapters and the enable/disable and webhook registration side effects through `setMountedChannels` (`integrations/mounted-channels.ts`); the `/:platform/enable|disable|setup` routes and the actions share them. Add a new side effect there, not in a route. - A new channel adapter needs a catalog entry in `integrations/catalog.ts`; `catalog.spec.ts` fails until the catalog and `BUILT_IN_INTEGRATION_ADAPTER_IDS` match. Add its about line under `settingsShell.channels.about.*` in every core locale. - `webhookSetup` decides the page's rows: `manual` shows the webhook URL to copy, `automatic` adds a Register button (`register-webhook`), and no value (Google Docs) shows no webhook URL. - An app adds its own settings to one channel's page with `registerChannelSettingsExtensions([{ id, platform, component }])` from `@agent-native/core/client/settings`, at module scope next to its settings route. Clips registers Slack link previews this way. ## SQL Schema The pending-task queue lives in `integration_pending_tasks`: ```sql CREATE TABLE IF NOT EXISTS integration_pending_tasks ( id TEXT PRIMARY KEY, platform TEXT NOT NULL, external_thread_id TEXT NOT NULL, payload TEXT NOT NULL, -- JSON-serialized IncomingMessage owner_email TEXT NOT NULL, org_id TEXT, status TEXT NOT NULL, -- pending | processing | completed | failed attempts INTEGER NOT NULL DEFAULT 0, dispatch_attempts INTEGER NOT NULL DEFAULT 0, last_dispatch_at INTEGER, last_dispatch_outcome TEXT, dispatch_scope TEXT, -- persisted channel scope for recovery error_message TEXT, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL, completed_at INTEGER ); CREATE INDEX IF NOT EXISTS idx_pending_tasks_status_created ON integration_pending_tasks(status, created_at); CREATE INDEX IF NOT EXISTS idx_pending_tasks_dispatch_scope ON integration_pending_tasks(platform, dispatch_scope); ``` The store layer creates this lazily on first use via `ensureTable()` and uses PostgreSQL `BIGINT` columns for millisecond timestamps and counters. `claimPendingTask` is the critical concurrency primitive: it atomically flips `pending` → `processing` and increments `attempts`, returning `null` if another worker beat us to it. Both the initial dispatch and every recovery sweep funnel through the same processor endpoint, and `claimPendingTask` is what prevents the same task from being processed twice. ### Netlify durable recovery The portable self-dispatch remains the default. To emit and use Netlify's acknowledged background worker plus a one-minute scheduled recovery function, set `AGENT_INTEGRATION_DURABLE_DISPATCH=true` at build and runtime. A production rollout can be narrowed with the comma-separated `AGENT_INTEGRATION_DURABLE_DISPATCH_SCOPES` allowlist. Supported values are `<platform>:*`, `<platform>:<external-thread-id>`, and, for adapters that expose one, `<platform>:<channel-id>` (for example `slack:C123`). Channel-scoped handoffs persist that scope separately from the provider thread identity so a later bounded recovery sweep applies the same allowlist decision. Workspace deploys emit this pair only for the Dispatch control-plane app, which owns workspace messaging integrations. Standalone templates emit their own pair when the same flag is enabled. The background handoff waits only for Netlify's enqueue acknowledgement, not for the agent loop. A failed or non-2xx acknowledgement falls back to the portable processor. The scheduled function calls the HMAC-authenticated, bounded `/retry-stuck-tasks` route; its atomic compare-and-set update prevents overlapping sweeps from dispatching the same stale row twice. Rows acknowledged by the 15-minute background worker use a 16-minute stale cutoff, while portable synchronous dispatches retain the shorter serverless cutoff. Apply that lease predicate before `LIMIT` so healthy long runs cannot hide recoverable work. The background processor writes its lease marker in the same atomic update that claims the row; best-effort dispatch telemetry is not trusted for replay safety. ## Adding a New Platform Adapter 1. **Implement `PlatformAdapter`** in `packages/core/src/integrations/adapters/<platform>.ts`: ```ts export function myPlatformAdapter(): PlatformAdapter { return { platform: "myplatform", label: "MyPlatform", getRequiredEnvKeys: () => [ { name: "MYPLATFORM_TOKEN", label: "MyPlatform Bot Token", scope: "global" }, { name: "MYPLATFORM_SIGNING_SECRET", label: "MyPlatform Signing Secret", scope: "global" }, ], async handleVerification(event) { // Platform-specific challenge response, if any return { handled: false }; }, async verifyWebhook(event) { // Verify HMAC/signature with a server-side secret and constant-time comparison. // Never leave this as a permissive stub in production. return verifyMyPlatformSignature(event); }, async parseIncomingMessage(event) { // Map raw payload → IncomingMessage, or null to ignore return null; }, async sendResponse(message, context) { // POST back to the platform's API }, formatAgentResponse(text) { return { text, platformContext: {} }; }, async getStatus(baseUrl) { return { platform: "myplatform", label: "MyPlatform", enabled: false, configured: false }; }, }; } ``` 2. **Register it** in `getDefaultAdapters()` inside `plugin.ts`. The webhook, queue, processor, and retry job are shared infrastructure — you do not write any of that per-adapter. 3. **Declare required env keys** so the secrets/onboarding UI surfaces them. See `secrets` and `onboarding` skills. 4. **Update the platform's webhook URL** to point at `${baseUrl}/_agent-native/integrations/<platform>/webhook`. For platforms with a registration API (Telegram), implement `POST /:platform/setup`. Never hardcode bot tokens, signing secrets, verification tokens, webhook URLs, channel/customer identifiers, or copied platform payloads with real private data inside the adapter, tests, docs, prompts, or fixtures. `getRequiredEnvKeys()` declares credential names only. Values come from deployment configuration, registered secrets, OAuth, or scoped credential stores, and tests should use obvious fake placeholders. The adapter is **only** responsible for: - platform-specific verification (signatures, challenges) - payload → `IncomingMessage` mapping - agent text → platform format - delivering the response back to the platform It does **not** know about the queue, the processor, retries, or the agent loop. Those are handled by the shared webhook handler. ### Slack non-message events Slack's Events API also sends events that are not agent messages, such as `link_shared` for app unfurls. Do not map those into `IncomingMessage` unless they should actually run the agent. Handle them as short, provider-specific webhook work: verify the Slack signature, return `200 OK` quickly, and call the provider API needed for the event (`chat.unfurl` for link previews). If a single Slack app must handle both agent chat and app unfurls, put a dispatcher in front of the one Slack Events Request URL and route message events to the integration webhook flow while routing `link_shared` to the app-specific unfurl handler. ## Long-Running Agent Work The processor endpoint runs in a fresh function execution with its own full timeout (typically 30–60s on Netlify/Vercel, longer on background-friendly hosts). That budget is dedicated entirely to the agent loop — there is no platform-side timer racing it. If a single agent run might exceed the function timeout (large multi-step plans, deep delegation chains), the agent should: 1. Send an interim acknowledgement back to the platform so the user knows the request landed (`adapter.sendResponse({ text: "Working on it..." })`). 2. Persist intermediate state in chat-thread data, application state, or a recurring job so the next invocation can pick up where this one left off. The retry job will only re-fire tasks stuck in `processing` for over 5 minutes, so a normal long-running reply is safe. ## Cross-Platform Considerations - **Portable correctness, optional host acceleration.** Every host uses SQL, HMAC-authenticated processor routes, and atomic claims. A host-specific background function or scheduler may provide a more reliable wake-up, but disabling it must preserve the portable path. - **No assumed runtime.** The processor endpoint is a normal H3 handler under `/_agent-native/`. It runs wherever the rest of the framework runs. - **No persistent in-memory state.** The dedup map in the webhook handler is best-effort only; the SQL queue is the source of truth. Any cold start loses the dedup map but the queue stays consistent. - **Postgres concurrency.** `claimPendingTask` uses one conditional update with `RETURNING`, so the claim and read are one database operation. - **Self-webhook URL resolution.** The processor URL is built from `WEBHOOK_BASE_URL`, `APP_URL`, or `URL` env vars (with `localhost:3000` as the dev fallback). Templates that change their public URL must keep one of these set. ## Why Fire-and-Forget on Serverless Is Unreliable The portable webhook path gives the outbound processor request a short head start but does not wait for the response body, so that initial dispatch is **not** guaranteed to complete before the function freezes. On a long-lived host, the in-process retry loop is a best-effort safety net. On Netlify with durable dispatch enabled, the initial handoff is acknowledged and the external scheduled function supplies the recovery wake-up for cases where: - The serverless platform froze the handler before the outbound `fetch` flushed its bytes. - The processor function 502'd or cold-started slow enough to time out. - The processor itself was killed mid-agent-loop (function timeout, container shutdown, deploy mid-run). Tasks stuck in `pending` for >90s or `processing` for >5min get re-fired up to 3 times. After 3 attempts they're marked `failed` permanently so we stop spamming the processor. **Never assume the initial dispatch succeeded.** Rely on the SQL queue, atomic claim, and a recovery mechanism whose lifetime is independent of the original request. An in-process timer alone is not that mechanism on serverless. ## Debugging Checklist 1. **Platform sent the webhook?** Check the platform's delivery logs (Slack admin, Telegram `getWebhookInfo`). 2. **Webhook handler returned 200?** If not, the platform retries — look for duplicate task rows. Signature failures return 401. 3. **Task in the queue?** `SELECT * FROM integration_pending_tasks WHERE external_thread_id = '...' ORDER BY created_at DESC LIMIT 5`. 4. **Status?** `pending` means the processor never picked it up — check that `_process-task` is reachable from the box itself (the self-fetch must work over the public URL). `processing` for over 5 minutes means the processor died mid-run — a live recovery sweep will pick it up. 5. **Failed?** Check `error_message` and `attempts`. After 3 attempts the row is parked at `failed` and won't be retried.