-Receive inbound over **webhook · WebSocket · REST · MCP**. Send through an **HTTP API**. Inbound mail includes structured **SPF · DKIM · DMARC** evidence.
+**[Try hosted e2a — start free →](https://e2a.dev)**
-A [Token Canopy](https://tokencanopy.com) product
+We run the email infrastructure. You connect your app or agent.
-[](https://github.com/tokencanopy/e2a/actions/workflows/test.yml)
-[](https://github.com/tokencanopy/e2a/actions/workflows/build-image.yml)
-[](LICENSE)
-[](https://www.npmjs.com/package/@e2a/sdk)
-[](https://pypi.org/project/e2a/)
-[](https://mcptoplist.com/server/dev.e2a%2Fmcp-server)
-[](https://github.com/tokencanopy/e2a/releases/latest)
+[Self-host with Docker](#self-host-docker) · [Documentation](#api) · [Agent quickstart](#quickstart) · [Examples](#working-examples)
-**`/v1` is now generally available** — shipped in [**v1.5.0**](https://github.com/tokencanopy/e2a/releases/tag/v1.5.0).
-
-[Hosted (e2a.dev)](https://e2a.dev) · [Transactional email API](https://e2a.dev/transactional-email-api) · [Agent quickstart](#quickstart) · [Examples](#working-examples) · [Concepts](#concepts) · [API](#api) · [SDKs](#sdks) · [MCP](#mcp-server) · [Deploy](#deployment) · [FAQ](#faq)
-
-
+A [Token Canopy](https://tokencanopy.com) product · Apache 2.0
----
+**Using a coding agent? Paste this prompt into its chat:**
-> [!IMPORTANT]
-> **The core `/v1` API and SDKs are stable and generally available (GA) as of [v1.5.0](https://github.com/tokencanopy/e2a/releases/tag/v1.5.0): no breaking changes within `/v1`.** That tag is the compatibility baseline — every later release is audited against it. A small, explicitly enumerated surface is still **beta** and may change before it is declared stable — contacts & outreach, scheduled sending (`send_at`), email templates & starter templates, the reviews (HITL) queue, agent protection config, agent-scoped suppressions, managed unsubscribe, message lifecycle diagnostics, delivery metrics, and the `thread_id` message-read field. Beta surface is marked `x-stability-level: beta` in the OpenAPI spec and `(beta)` in the docs; where only specific *values* of a stable field are beta (the `scheduled` send status, the screening/review-hold event types, the `blocked_by_policy` error code), the field carries `x-experimental-values` naming exactly those values. Everything else is covered by the GA freeze. See the full matrix in [docs/api.md → Stability: GA and beta surface](docs/api.md#stability-ga-and-beta-surface). Existing `v1.0.x` application/cherry-pick tags predate the API freeze and are not `/v1` compatibility baselines.
+```text
+Connect this coding agent to hosted e2a MCP at https://api.e2a.dev/mcp and help me sign in via browser OAuth.
+```
-e2a is the **open-source email API for applications and AI agents**. Any product can send transactional email over HTTP, TypeScript, or Python; agent-native systems can also use real two-way inboxes. Inbound mail arrives with structured SPF, DKIM, and DMARC evidence, and outbound mail can use an optional human-in-the-loop approval gate. Use the hosted service or run the Apache-2.0 stack yourself. No AI agent or agent framework is required for application-triggered sending.
+Supports coding agents with remote MCP and browser OAuth. [Client setup guide](https://e2a.dev/setup.md).
-**Four ways to plug an agent in:**
+
-- **MCP** — point any MCP-aware runtime at the hosted server (`https://api.e2a.dev/mcp`) and your agent gets an inbox toolset (`list_messages`, `send_message`, `reply_to_message`, …). The fastest path for agent frameworks. → [MCP server](#mcp-server)
-- **SDKs** — TypeScript (`@e2a/sdk`) and Python (`e2a`) clients with one-call webhook verification and a WebSocket `listen()` stream. → [SDKs](#sdks)
-- **Raw delivery** — subscribe a **webhook**, open a **WebSocket**, or **poll** the REST API directly. → [Delivery channels](#delivery-channels)
-- **CLI** — `e2a listen` bridges inbound mail to a local HTTP handler (including an OpenAI Responses auto-reply mode). → [CLI](#cli)
+## Choose how to start
-What you get on top of bare SMTP:
+- **Hosted — recommended for getting started.** Sign up at [e2a.dev](https://e2a.dev). Includes the shared `agents.e2a.dev` domain for instant slug-based onboarding (no DNS setup), a dashboard, the hosted MCP server, and managed deliverability.
+- **Self-host — run your own infrastructure.** See [Self-host (Docker)](#self-host-docker) and [Deployment](#deployment). Nearly every feature works the same (content screening is currently self-host-only — see the [note below](#content-screening)); the shared-domain slug shortcut just needs you to point a mail domain at your relay and set `shared_domain` in `config.yaml`.
-- **Authenticated inbound identity** — normalized SPF, DKIM, and DMARC evidence, with an explicit aligned DMARC verdict
-- **No public URL required** — WebSocket, REST polling, and MCP all work from a laptop or behind a firewall
-- **Outbound API** — agents send to other agents (SMTP relay) or humans (upstream SMTP, e.g. SES, Resend)
-- **Human in the loop** — opt-in approval gate that holds outbound mail until a reviewer approves via dashboard, magic-link email, the MCP tools, or the API
-- **Inbound threat screening** — opt-in content scan flags **prompt-injection** payloads (hidden HTML, Unicode-tag smuggling, encoded text) — and, with the LLM detector, **phishing** — then routes each message to *allow · review · block*, feeding the same review queue as HITL → [Content screening](#content-screening). *Available on self-hosted deployments; not yet enabled on the hosted service.*
-- **Email reply topology** — standards-compliant reply headers plus optional beta `thread_id` metadata on message reads; caller-owned `conversation_id` remains application correlation
-- **Email templates (beta)** — reusable `{{variable}}` templates rendered server-side at send time, plus a pre-built starter catalog → [docs/templates.md](docs/templates.md)
-- **Contacts & outreach (beta)** — account-level contact identity (CRUD + bulk import with safe reversal) and per-agent outreach state with server-derived reply/delivery facts, plus the `contact.due` due-queue notification event → [docs/api.md](docs/api.md#contacts--outreach-v1contacts-v1agentsemailcontacts-beta)
-- **Scheduled sending (beta)** — `send_at` on send/reply/forward defers submission up to 90 days ahead; a scheduled send is durable acceptance (`status=scheduled`) and can be canceled by trashing the message before submission
+For application email, start with the [transactional email guide](https://e2a.dev/transactional-email-api). For coding agents, use the prompt above or the setup instructions below.
## Quickstart
-The fastest path is to give your AI agent an inbox directly. Install the e2a plugin — it registers the hosted [MCP server](#mcp-server) and an operate-well skill, so your agent can send, receive, reply in-thread, and hold mail for review out of the box. On first tool use it runs an OAuth flow in your browser — no API key to paste.
+### Connect your agent to hosted e2a
+
+Give your AI agent an inbox directly. Install the e2a plugin — it registers the hosted [MCP server](#mcp-server) and an operate-well skill, so your agent can send, receive, reply in-thread, and hold mail for review out of the box. On first tool use it runs an OAuth flow in your browser — no API key to paste.
**Claude Code**
@@ -88,12 +71,42 @@ Then launch `codex`, run `/plugins`, and install **e2a**.
**Other MCP clients** (Zed, Goose, Windsurf, Claude Desktop, raw `mcp.json`) — point straight at `https://api.e2a.dev/mcp`; ready-to-paste configs are in [plugins/e2a/clients/](plugins/e2a/clients). See [plugins/e2a/README.md](plugins/e2a/README.md) for the full per-client guide.
-## Use it
+
-You can either use the hosted instance or self-host.
+page body
survey body
survey body
page body
page body
page body
+ Loading... +
+html
", nil, "send", "", "", "", 60) + if err != nil { + t.Fatal(err) + } + backdateExpiry(t, pool, msg.ID) + + w.RunOnce(ctx) + if msgs := smtpDone(); len(msgs) != 0 { + t.Fatalf("paused account must not send inline, got %d SMTP messages", len(msgs)) + } + if len(enq.calls) != 1 { + t.Fatalf("enqueue attempts = %v, want exactly one", enq.calls) + } + var status string + var expiresAt time.Time + if err := pool.QueryRow(ctx, `SELECT status, approval_expires_at FROM messages WHERE id=$1`, msg.ID).Scan(&status, &expiresAt); err != nil { + t.Fatal(err) + } + if status != identity.MessageStatusPendingReview { + t.Fatalf("status = %q, want pending_review (held, not rejected)", status) + } + if expiresAt.Before(time.Now().Add(50 * time.Minute)) { + t.Fatalf("approval_expires_at = %v, want deferred about an hour ahead", expiresAt) + } + // Deferred out of the window: the next sweep leaves it alone. + w.RunOnce(ctx) + if len(enq.calls) != 1 { + t.Fatalf("enqueue attempts after deferral = %v, want still one", enq.calls) + } +} diff --git a/internal/hitlworker/worker.go b/internal/hitlworker/worker.go index ca4fbe5ec..00a1467e2 100644 --- a/internal/hitlworker/worker.go +++ b/internal/hitlworker/worker.go @@ -27,6 +27,7 @@ import ( "github.com/tokencanopy/e2a/internal/loopback" "github.com/tokencanopy/e2a/internal/messagelifecycle" "github.com/tokencanopy/e2a/internal/outbound" + "github.com/tokencanopy/e2a/internal/outboundsend" "github.com/tokencanopy/e2a/internal/piguard" "github.com/tokencanopy/e2a/internal/usage" "github.com/tokencanopy/e2a/internal/webhookpub" @@ -61,6 +62,11 @@ const DefaultBatchSize = 100 // Worker runs the TTL sweep. Construct with New; its RunOnce is driven on a // schedule by the River maintenance periodic (see maintenance.go). +// pausedReviewRetry is how far a TTL-expired review on a paused account is +// deferred before the sweep looks at it again. Long enough not to churn, short +// enough that a resume is picked up within the hour. +const pausedReviewRetry = time.Hour + type Worker struct { store *identity.Store sender *outbound.Sender @@ -406,6 +412,17 @@ func (w *Worker) autoApproveAsync(ctx context.Context, agent *identity.AgentIden if errors.Is(err, identity.ErrNotPendingApproval) { return true // resolved between load and transition } + if errors.Is(err, outboundsend.ErrSendingPaused) { + // The account is paused for sending. The draft stays pending_review + // — that is the held queue the pause promises — but it must not + // stay the sweep's oldest candidate, or it is re-picked first every + // cycle and starves every other expired review. Defer its TTL; the + // sweep after resume resolves it. + if derr := w.store.DeferReviewExpiry(ctx, c.MessageID, time.Now().Add(pausedReviewRetry)); derr != nil { + log.Printf("[hitl-worker] auto-approve %s: defer while account is paused: %v", c.MessageID, derr) + } + return true + } // Transient tx/enqueue failure: leave the row pending_review for the next // cycle. Do NOT autoReject — no send happened, so this is not a "stuck" send. log.Printf("[hitl-worker] auto-approve %s: accept+enqueue: %v", c.MessageID, err) diff --git a/internal/httpapi/error_catalog.go b/internal/httpapi/error_catalog.go index 558bab232..e71b00587 100644 --- a/internal/httpapi/error_catalog.go +++ b/internal/httpapi/error_catalog.go @@ -21,6 +21,7 @@ var errorCodeCatalog = []errorCodeContract{ {Code: "unauthorized", Status: "401", Family: "auth"}, {Code: "forbidden", Status: "403", Family: "auth"}, {Code: "blocked_by_policy", Status: "403", Family: "auth"}, + {Code: "sending_paused", Status: "403", Family: "auth"}, {Code: "invalid_request", Status: "400 / 422", Family: "validation", DetailsSchema: "ValidationErrorDetails"}, {Code: "invalid_cursor", Status: "400", Family: "validation"}, {Code: "invalid_filter", Status: "400", Family: "validation"}, diff --git a/internal/httpapi/errors.go b/internal/httpapi/errors.go index 50b3c6470..049c48b54 100644 --- a/internal/httpapi/errors.go +++ b/internal/httpapi/errors.go @@ -54,7 +54,7 @@ type ErrorEnvelope struct { // ErrorBody is the inner object of the envelope. type ErrorBody struct { - Code string `json:"code" doc:"Machine-branchable error code — the stable discriminator clients switch on. Open set: treat it as a string and tolerate unknown values, since new codes may be added over time (branch on the ones you handle, fall back to the HTTP status otherwise). Exact current vocabulary (machine-checked): unauthorized, forbidden, blocked_by_policy, invalid_request, invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope, reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty, recipient_suppressed, not_found, attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found, gone, conflict, precondition_failed, agent_taken, domain_taken, alias_taken, address_in_trash, message_held, message_not_pending, message_not_yet_delivered, not_in_trash, purge_in_progress, send_in_progress, webhook_disabled, webhook_cooldown, domain_not_registered, domain_has_agents, domain_not_verified, inbound_mx_missing, limit_exceeded, rate_limited, contact_limit_reached, template_limit_reached, webhook_limit_reached, idempotency_in_flight, idempotency_key_reuse, payload_too_large, attachment_too_large, not_implemented, events_log_disabled, limits_unavailable, inbound_mx_check_failed, auth_unavailable, internal_error, method_not_allowed, unsupported_media_type, error. Grouped semantics: auth: unauthorized (401), forbidden (403), blocked_by_policy (403, outbound policy gate; experimental). Validation: invalid_request is the single canonical code for input-validation failures whether they arrive as 400 (malformed) or 422 (semantically invalid); field/resource-specific invalid_* refinements (invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope), reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty (all 400); recipient_suppressed (422). Not found: not_found (404) plus the *_not_found family (attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found); gone (410, past retention). Conflict/state: conflict (409, generic), precondition_failed (412, optimistic-concurrency validator is stale), the *_taken family — the requested identifier is already claimed — (agent_taken, domain_taken, alias_taken, all 409), address_in_trash (409), message_held (409), message_not_pending (409), message_not_yet_delivered (409, retry after the source outbound message is sent), not_in_trash (409), purge_in_progress (409, permanent delete already claimed), send_in_progress (409), webhook_disabled (409), webhook_cooldown (409), domain_not_registered (400), domain_has_agents (400), domain_not_verified (400 on create-agent, 403 on send), inbound_mx_missing (400). Capacity: limit_exceeded (402, plan quota — see LimitExceededDetails), rate_limited (429, request rate — see RateLimitedDetails), contact_limit_reached, template_limit_reached and webhook_limit_reached (400, fixed per-account caps). Idempotency: idempotency_in_flight (409, wait then retry the byte-identical request), idempotency_key_reuse (422, caller bug — do not retry as-is). Size: payload_too_large (413, request body), attachment_too_large (413, inline fetch over the cap — use download_url). Availability: not_implemented (501, feature not available on this deployment), events_log_disabled (501), limits_unavailable (503), inbound_mx_check_failed (503), auth_unavailable (503, an auth backend — e.g. a delegated-token verifier or the identity store — could not judge the credential; retry). Server/fallback: internal_error (5xx), method_not_allowed (405), unsupported_media_type (415), and the generic code error for any otherwise-unmapped status."` + Code string `json:"code" doc:"Machine-branchable error code — the stable discriminator clients switch on. Open set: treat it as a string and tolerate unknown values, since new codes may be added over time (branch on the ones you handle, fall back to the HTTP status otherwise). Exact current vocabulary (machine-checked): unauthorized, forbidden, blocked_by_policy, sending_paused, invalid_request, invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope, reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty, recipient_suppressed, not_found, attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found, gone, conflict, precondition_failed, agent_taken, domain_taken, alias_taken, address_in_trash, message_held, message_not_pending, message_not_yet_delivered, not_in_trash, purge_in_progress, send_in_progress, webhook_disabled, webhook_cooldown, domain_not_registered, domain_has_agents, domain_not_verified, inbound_mx_missing, limit_exceeded, rate_limited, contact_limit_reached, template_limit_reached, webhook_limit_reached, idempotency_in_flight, idempotency_key_reuse, payload_too_large, attachment_too_large, not_implemented, events_log_disabled, limits_unavailable, inbound_mx_check_failed, auth_unavailable, internal_error, method_not_allowed, unsupported_media_type, error. Grouped semantics: auth: unauthorized (401), forbidden (403), blocked_by_policy (403, outbound policy gate; experimental), sending_paused (403, outbound sending is paused for the account by the platform abuse controls; queued mail is held, new sends are refused until an operator resumes; experimental). Validation: invalid_request is the single canonical code for input-validation failures whether they arrive as 400 (malformed) or 422 (semantically invalid); field/resource-specific invalid_* refinements (invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope), reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty (all 400); recipient_suppressed (422). Not found: not_found (404) plus the *_not_found family (attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found); gone (410, past retention). Conflict/state: conflict (409, generic), precondition_failed (412, optimistic-concurrency validator is stale), the *_taken family — the requested identifier is already claimed — (agent_taken, domain_taken, alias_taken, all 409), address_in_trash (409), message_held (409), message_not_pending (409), message_not_yet_delivered (409, retry after the source outbound message is sent), not_in_trash (409), purge_in_progress (409, permanent delete already claimed), send_in_progress (409), webhook_disabled (409), webhook_cooldown (409), domain_not_registered (400), domain_has_agents (400), domain_not_verified (400 on create-agent, 403 on send), inbound_mx_missing (400). Capacity: limit_exceeded (402, plan quota — see LimitExceededDetails), rate_limited (429, request rate — see RateLimitedDetails), contact_limit_reached, template_limit_reached and webhook_limit_reached (400, fixed per-account caps). Idempotency: idempotency_in_flight (409, wait then retry the byte-identical request), idempotency_key_reuse (422, caller bug — do not retry as-is). Size: payload_too_large (413, request body), attachment_too_large (413, inline fetch over the cap — use download_url). Availability: not_implemented (501, feature not available on this deployment), events_log_disabled (501), limits_unavailable (503), inbound_mx_check_failed (503), auth_unavailable (503, an auth backend — e.g. a delegated-token verifier or the identity store — could not judge the credential; retry). Server/fallback: internal_error (5xx), method_not_allowed (405), unsupported_media_type (415), and the generic code error for any otherwise-unmapped status."` Message string `json:"message" doc:"Human-readable explanation. Not for branching — use code."` Details any `json:"details,omitempty" doc:"Optional structured context, polymorphic by code. Treat it as an open object keyed off code; unknown codes and fields must be preserved."` RequestID string `json:"request_id" doc:"Echoes the X-Request-Id response header so a failing call is greppable in logs."` diff --git a/internal/httpapi/spec_review_test.go b/internal/httpapi/spec_review_test.go index abf340c3f..13a9c62f7 100644 --- a/internal/httpapi/spec_review_test.go +++ b/internal/httpapi/spec_review_test.go @@ -263,6 +263,8 @@ func assertMessageLifecycleContractSchema(t *testing.T, doc map[string]any) { "suppression.recipient_blocked", "suppression.hard_bounce_applied", "suppression.complaint_applied", "queue.inbound_processing", "queue.outbound_submission", "submission.upstream_accepted", "submission.local_loopback_accepted", "submission.temporary_failure", "submission.provider_rejected", "submission.local_retries_exhausted", "submission.cancelled", + "submission.policy_budget_expired", + "submission.sending_setup_expired", "delivery.recipient_server_accepted", "delivery.temporary_delay", "delivery.permanent_bounce", "delivery.transient_bounce", "delivery.undetermined_bounce", "complaint.recipient_reported", }, diff --git a/internal/httpapi/stability.go b/internal/httpapi/stability.go index 7eea7ad91..7d30d3b26 100644 --- a/internal/httpapi/stability.go +++ b/internal/httpapi/stability.go @@ -244,9 +244,10 @@ func (s *Server) applyEvolutionStance() { for _, schema := range []string{"HoldReasonView", "ProtectionFindingView", "ThreatCategoryView"} { markSchema(schemas, schema, extStabilityLevel, stabilityBeta) } - // ErrorBody.code is a stable open discriminator; only the outbound - // gate-policy value remains experimental. - markProperty(schemas, "ErrorBody", "code", extExperimentalValues, []string{"blocked_by_policy"}) + // ErrorBody.code is a stable open discriminator; the outbound gate-policy + // value and the sending-abuse pause value remain experimental — both are + // produced by controls that ship disabled. + markProperty(schemas, "ErrorBody", "code", extExperimentalValues, []string{"blocked_by_policy", "sending_paused"}) // // The template hooks on send are beta (templates are beta) even though // sendMessage itself is stable. diff --git a/internal/httpapi/stability_test.go b/internal/httpapi/stability_test.go index dfdb3dd0f..1f37e285d 100644 --- a/internal/httpapi/stability_test.go +++ b/internal/httpapi/stability_test.go @@ -436,12 +436,13 @@ func TestSpecBetaMarkers(t *testing.T) { } } - // The error discriminator remains stable; only the gate-policy value is - // experimental. + // The error discriminator remains stable; only the two values produced by + // controls that ship disabled — the outbound gate policy and the sending + // abuse pause — are experimental. errorCode, _ := schemaProps(t, doc, "ErrorBody")["code"].(map[string]any) rawErrorValues, _ := errorCode["x-experimental-values"].([]any) - if len(rawErrorValues) != 1 || rawErrorValues[0] != "blocked_by_policy" { - t.Errorf("ErrorBody.code x-experimental-values = %v, want [blocked_by_policy]", rawErrorValues) + if len(rawErrorValues) != 2 || rawErrorValues[0] != "blocked_by_policy" || rawErrorValues[1] != "sending_paused" { + t.Errorf("ErrorBody.code x-experimental-values = %v, want [blocked_by_policy sending_paused]", rawErrorValues) } // Managed unsubscribe is a beta opt-in nested inside otherwise-stable diff --git a/internal/identity/acquisition.go b/internal/identity/acquisition.go new file mode 100644 index 000000000..721cc390c --- /dev/null +++ b/internal/identity/acquisition.go @@ -0,0 +1,38 @@ +package identity + +import "errors" + +// AcquisitionSources is the closed answer set for the onboarding survey +// ("Where did you hear about e2a?"). It must match the CHECK constraint +// in migrations/120_users_acquisition_survey.sql exactly — the values +// are the analytics enum, so they are code, not config. +var AcquisitionSources = []string{ + "search", + "ai_assistant", + "github", + "x_twitter", + "hn_reddit", + "content", + "mcp_directory", + "word_of_mouth", + "other", + AcquisitionSourceSkipped, +} + +// AcquisitionSourceSkipped records "asked, declined". It counts as +// answered so the survey never reappears. +const AcquisitionSourceSkipped = "skipped" + +// ErrAcquisitionSurveyAnswered is returned by RecordAcquisitionSurvey when +// the user already has an answer on file. The first answer is kept. +var ErrAcquisitionSurveyAnswered = errors.New("acquisition survey already answered") + +// IsAcquisitionSource reports whether s is exactly one of AcquisitionSources. +func IsAcquisitionSource(s string) bool { + for _, v := range AcquisitionSources { + if v == s { + return true + } + } + return false +} diff --git a/internal/identity/acquisition_test.go b/internal/identity/acquisition_test.go new file mode 100644 index 000000000..717e0affd --- /dev/null +++ b/internal/identity/acquisition_test.go @@ -0,0 +1,44 @@ +package identity_test + +import ( + "regexp" + "strings" + "testing" + + "github.com/tokencanopy/e2a/internal/identity" + "github.com/tokencanopy/e2a/migrations" +) + +// The Go enum and the SQL CHECK are two copies of one list. Read the +// CHECK back out of the migration so adding a value to only one side +// fails here, without a database. +func TestAcquisitionSourcesMatchMigrationEnum(t *testing.T) { + sql, err := migrations.FS.ReadFile("120_users_acquisition_survey.sql") + if err != nil { + t.Fatal(err) + } + m := regexp.MustCompile(`(?s)acquisition_source IN \((.*?)\)`).FindStringSubmatch(string(sql)) + if m == nil { + t.Fatal("no `acquisition_source IN (...)` CHECK found in migration 120") + } + var fromSQL []string + for _, q := range regexp.MustCompile(`'([a-z_]+)'`).FindAllStringSubmatch(m[1], -1) { + fromSQL = append(fromSQL, q[1]) + } + if got, want := strings.Join(identity.AcquisitionSources, ","), strings.Join(fromSQL, ","); got != want { + t.Fatalf("Go enum %q != migration CHECK %q", got, want) + } + for _, s := range identity.AcquisitionSources { + if !identity.IsAcquisitionSource(s) { + t.Errorf("IsAcquisitionSource(%q) = false", s) + } + } + for _, bad := range []string{"", "Search", "carrier_pigeon", " github"} { + if identity.IsAcquisitionSource(bad) { + t.Errorf("IsAcquisitionSource(%q) = true", bad) + } + } + if identity.AcquisitionSourceSkipped != "skipped" || !identity.IsAcquisitionSource("skipped") { + t.Errorf("AcquisitionSourceSkipped = %q", identity.AcquisitionSourceSkipped) + } +} diff --git a/internal/identity/delivery_store.go b/internal/identity/delivery_store.go index 3be22a369..4d53d0c8d 100644 --- a/internal/identity/delivery_store.go +++ b/internal/identity/delivery_store.go @@ -319,6 +319,7 @@ func (s *Store) RecordDeliveryOutcomeTx(ctx context.Context, tx pgx.Tx, messageI if _, err := tx.Exec(ctx, `UPDATE messages SET delivery_status = 'failed', + local_hold_class = NULL, local_hold_anchor = NULL, delivery_failure_source = COALESCE(delivery_failure_source, 'provider') WHERE id = $1`, messageID, ); err != nil { @@ -373,7 +374,7 @@ func (s *Store) MarkMessageSent(ctx context.Context, messageID, sentAs string, t defer tx.Rollback(ctx) if _, err := tx.Exec(ctx, - `UPDATE messages SET delivery_status = 'sent', sent_as = $2 WHERE id = $1`, + `UPDATE messages SET delivery_status = 'sent', sent_as = $2, local_hold_class = NULL, local_hold_anchor = NULL WHERE id = $1`, messageID, nullIfEmpty(sentAs), ); err != nil { return err @@ -444,6 +445,18 @@ type OutboundSendPayload struct { ReviewedAt *time.Time // ProviderMessageID is the evidence-repaired provider id ('' when none). ProviderMessageID string + // LocalHoldClass / LocalHoldAnchor are the durable finite-hold state the + // worker persisted on an earlier execution ('' / nil when the message has + // never entered a finite hold). The absolute deadline is always derived + // from this pair, never stored. + LocalHoldClass string + LocalHoldAnchor *time.Time + // LastResumedAt is account_sending_controls.last_resumed_at for the owning + // account; TenantReadyAt is its ses_tenant_ready_at (nil until the SES + // tenant is ready). Both feed the worker's hold-anchor and setup→rate + // transition rules. nil when the account has no control row yet. + LastResumedAt *time.Time + TenantReadyAt *time.Time } // OutboundSentInfo carries the fields the async worker's MarkSent/MarkFailed @@ -590,6 +603,10 @@ func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobI failureAttempt *int scheduledAt *time.Time reviewedAt *time.Time + holdClass string + holdAnchor *time.Time + lastResumedAt *time.Time + tenantReadyAt *time.Time ) var userID, registeredDomain string // Lock agent first to match permanent agent deletion's lock order, then @@ -613,14 +630,18 @@ func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobI m.to_recipients, m.cc, m.bcc, m.raw_message, m.created_at, m.deleted_at, m.send_job_id, m.provider_accepted_at, COALESCE(m.provider_message_id,''), COALESCE(m.delivery_failure_source,''),COALESCE(m.delivery_failure_reason_code,''), - m.delivery_failure_occurred_at,m.delivery_failure_attempt,m.scheduled_at,m.reviewed_at + m.delivery_failure_occurred_at,m.delivery_failure_attempt,m.scheduled_at,m.reviewed_at, + COALESCE(m.local_hold_class,''), m.local_hold_anchor, + c.last_resumed_at, c.ses_tenant_ready_at FROM messages m + LEFT JOIN account_sending_controls c ON c.user_id = $3 WHERE m.id = $1 AND m.agent_id = $2 AND m.direction = 'outbound' FOR UPDATE OF m`, - messageID, agentID, + messageID, agentID, userID, ).Scan(&deliveryStatus, &envelopeFrom, &sentAs, &messageType, &to, &cc, &bcc, &raw, &createdAt, &deletedAt, &stampedJobID, &providerAcceptedAt, &providerMessageID, - &failureSource, &failureReason, &failureOccurredAt, &failureAttempt, &scheduledAt, &reviewedAt) + &failureSource, &failureReason, &failureOccurredAt, &failureAttempt, &scheduledAt, &reviewedAt, + &holdClass, &holdAnchor, &lastResumedAt, &tenantReadyAt) if errors.Is(err, pgx.ErrNoRows) { if err := tx.Commit(ctx); err != nil { return nil, err @@ -665,6 +686,7 @@ func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobI if _, err := tx.Exec(ctx, `UPDATE messages SET delivery_status = 'failed', + local_hold_class = NULL, local_hold_anchor = NULL, delivery_detail = 'send canceled because the message or agent is in trash', delivery_failure_source = 'local', delivery_failure_reason_code = 'submission.cancelled', @@ -710,6 +732,10 @@ func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobI ProviderMessageID: providerMessageID, ScheduledAt: scheduledAt, ReviewedAt: reviewedAt, + LocalHoldClass: holdClass, + LocalHoldAnchor: holdAnchor, + LastResumedAt: lastResumedAt, + TenantReadyAt: tenantReadyAt, } if err := tx.Commit(ctx); err != nil { return nil, err @@ -717,6 +743,28 @@ func (s *Store) ClaimOutboundForSend(ctx context.Context, messageID string, jobI return p, nil } +// RecordOutboundHold persists a message's finite-hold class and anchor. +// +// The worker owns the transition rules (first finite hold, setup→rate, +// monotonic promotion to policy_budget); this writes exactly the pair it was +// given and only while the message is still pre-terminal. Terminal writes +// clear the pair, so a stale hold can never outlive its message's outcome. +func (s *Store) RecordOutboundHold(ctx context.Context, messageID, class string, anchor time.Time) error { + if class == "" || anchor.IsZero() { + return fmt.Errorf("record outbound hold: class and anchor are required") + } + _, err := s.pool.Exec(ctx, ` + UPDATE messages + SET local_hold_class = $2, local_hold_anchor = $3 + WHERE id = $1 AND direction = 'outbound' + AND delivery_status IN ('accepted', 'sending')`, + messageID, class, anchor.UTC()) + if err != nil { + return fmt.Errorf("record outbound hold: %w", err) + } + return nil +} + func isCompleteTerminalFallback(source, reason string, occurredAt *time.Time, attempt *int) bool { if occurredAt == nil || occurredAt.IsZero() || attempt == nil || *attempt < 0 { return false @@ -724,7 +772,8 @@ func isCompleteTerminalFallback(source, reason string, occurredAt *time.Time, at switch messagelifecycle.ReasonCode(reason) { case messagelifecycle.ReasonSubmissionProviderRejected: return delivery.FailureSource(source) == delivery.FailureSourceProvider - case messagelifecycle.ReasonSubmissionLocalRetriesExhausted, messagelifecycle.ReasonSubmissionCancelled: + case messagelifecycle.ReasonSubmissionLocalRetriesExhausted, messagelifecycle.ReasonSubmissionCancelled, + messagelifecycle.ReasonSubmissionPolicyBudgetExpired, messagelifecycle.ReasonSubmissionSendingSetupExpired: return delivery.FailureSource(source) == delivery.FailureSourceLocal default: return false @@ -808,6 +857,7 @@ func (s *Store) MarkOutboundSentTx(ctx context.Context, tx pgx.Tx, messageID, pr err := tx.QueryRow(ctx, `UPDATE messages m SET delivery_status = 'sent', provider_message_id = $2, send_claimed_at = NULL, + local_hold_class = NULL, local_hold_anchor = NULL, rfc_message_id_key = CASE WHEN rfc_message_id_key IS NULL AND $3 <> '' THEN $3 ELSE rfc_message_id_key @@ -891,7 +941,7 @@ func (s *Store) ResolveOutboundProviderAcceptedTx(ctx context.Context, tx pgx.Tx m := &Message{ID: messageID, Direction: "outbound", DeliveryStatus: "sent"} err = tx.QueryRow(ctx, `UPDATE messages m - SET delivery_status = 'sent', send_claimed_at = NULL, delivery_failure_source = NULL, delivery_failure_reason_code = NULL, delivery_detail = NULL, + SET delivery_status = 'sent', send_claimed_at = NULL, local_hold_class = NULL, local_hold_anchor = NULL, delivery_failure_source = NULL, delivery_failure_reason_code = NULL, delivery_detail = NULL, delivery_failure_occurred_at=NULL, delivery_failure_attempt=NULL, delivery_failure_blocked_recipients=NULL FROM agent_identities a WHERE m.id = $1 AND m.direction = 'outbound' @@ -981,6 +1031,7 @@ func (s *Store) MarkOutboundFailedTx(ctx context.Context, tx pgx.Tx, messageID, err := tx.QueryRow(ctx, `UPDATE messages m SET delivery_status = 'failed', + local_hold_class = NULL, local_hold_anchor = NULL, delivery_detail = COALESCE(NULLIF(m.delivery_detail, ''), $2), delivery_failure_source = $3, send_claimed_at = NULL diff --git a/internal/identity/migrate_test.go b/internal/identity/migrate_test.go index 089377497..5d5d94010 100644 --- a/internal/identity/migrate_test.go +++ b/internal/identity/migrate_test.go @@ -1762,3 +1762,43 @@ func TestRunMigrations_PartialState(t *testing.T) { } }) } + +// TestUsersAcquisitionSurveyMigrationIsNullableIdempotentAndConstrained +// exercises migration 120 (the plan's "108" slot was already taken by +// 108_external_principal_mappings.sql by the time this landed; 120 is +// the next free slot per TestEmbeddedMigrationNumbersAreUniqueFrom108). +func TestUsersAcquisitionSurveyMigrationIsNullableIdempotentAndConstrained(t *testing.T) { + ctx := context.Background() + pool := testutil.TestDB(t) + sql, err := migrations.FS.ReadFile("120_users_acquisition_survey.sql") + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, string(sql)); err != nil { + t.Fatalf("second migration application: %v", err) + } + for _, column := range []string{"acquisition_source", "acquisition_detail", "acquisition_answered_at"} { + var nullable, defaultValue string + if err := pool.QueryRow(ctx, `SELECT is_nullable,COALESCE(column_default,'') FROM information_schema.columns WHERE table_schema='public' AND table_name='users' AND column_name=$1`, column).Scan(&nullable, &defaultValue); err != nil { + t.Fatalf("column %s: %v", column, err) + } + if nullable != "YES" || defaultValue != "" { + t.Fatalf("column %s nullable=%q default=%q", column, nullable, defaultValue) + } + } + if _, err := pool.Exec(ctx, `INSERT INTO users (id, email, name, google_subject) VALUES ('usr_mig120', 'mig120@example.test', 'M', 'sub-mig120')`); err != nil { + t.Fatal(err) + } + // Unknown source is rejected by the CHECK. + if _, err := pool.Exec(ctx, `UPDATE users SET acquisition_source='carrier_pigeon', acquisition_answered_at=now() WHERE id='usr_mig120'`); err == nil { + t.Fatal("unknown acquisition_source was accepted") + } + // Source without timestamp is rejected (both-null-or-both-set). + if _, err := pool.Exec(ctx, `UPDATE users SET acquisition_source='github' WHERE id='usr_mig120'`); err == nil { + t.Fatal("acquisition_source without acquisition_answered_at was accepted") + } + // Valid pair is accepted. + if _, err := pool.Exec(ctx, `UPDATE users SET acquisition_source='github', acquisition_answered_at=now() WHERE id='usr_mig120'`); err != nil { + t.Fatalf("valid pair rejected: %v", err) + } +} diff --git a/internal/identity/outbound_hold_test.go b/internal/identity/outbound_hold_test.go new file mode 100644 index 000000000..88f8df06d --- /dev/null +++ b/internal/identity/outbound_hold_test.go @@ -0,0 +1,110 @@ +package identity_test + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/tokencanopy/e2a/internal/identity" + "github.com/tokencanopy/e2a/internal/testutil" +) + +// The finite-hold pair rides the claim payload so every worker execution +// re-derives the same deadline, and it is cleared by the terminal write so a +// stale hold can never outlive its message's outcome. +func TestOutboundHoldRidesTheClaimAndClearsOnTerminal(t *testing.T) { + pool := testutil.TestDB(t) + store := identity.NewStore(pool) + ctx := context.Background() + agentID := convoTestSetup(t, store, "hold-claim") + + var userID string + if err := pool.QueryRow(ctx, `SELECT user_id FROM agent_identities WHERE id = $1`, agentID).Scan(&userID); err != nil { + t.Fatal(err) + } + resumed := time.Date(2026, 9, 1, 8, 0, 0, 0, time.UTC) + ready := time.Date(2026, 9, 2, 9, 30, 0, 0, time.UTC) + if _, err := pool.Exec(ctx, ` + INSERT INTO account_sending_controls (user_id, last_resumed_at, ses_tenant_name, ses_tenant_ready, ses_tenant_ready_at) + VALUES ($1, $2, 'tenant_hold_test', true, $3) + ON CONFLICT (user_id) DO UPDATE SET last_resumed_at = $2, ses_tenant_ready = true, ses_tenant_ready_at = $3`, + userID, resumed, ready, + ); err != nil { + t.Fatal(err) + } + + var msgID string + if err := store.WithTx(ctx, func(tx pgx.Tx) error { + m, err := store.CreateOutboundMessageTx(ctx, tx, agentID, + []string{"one@example.test"}, nil, nil, "Hold", "send", "smtp", "", "conv-hold", + []byte("From: bot\r\n\r\nbody"), "accepted", "agent@test.e2a.dev", "relay") + if err != nil { + return err + } + msgID = m.ID + return store.StampSendJobIDTx(ctx, tx, m.ID, 4242) + }); err != nil { + t.Fatalf("seed: %v", err) + } + + p, err := store.ClaimOutboundForSend(ctx, msgID, 4242) + if err != nil || p == nil { + t.Fatalf("claim: payload=%v err=%v", p, err) + } + if p.LocalHoldClass != "" || p.LocalHoldAnchor != nil { + t.Fatalf("fresh claim carries a hold: %q %v", p.LocalHoldClass, p.LocalHoldAnchor) + } + if p.LastResumedAt == nil || !p.LastResumedAt.Equal(resumed) || p.TenantReadyAt == nil || !p.TenantReadyAt.Equal(ready) { + t.Fatalf("control timestamps = %v / %v, want %v / %v", p.LastResumedAt, p.TenantReadyAt, resumed, ready) + } + if err := store.ReleaseOutboundSendClaim(ctx, msgID, 4242); err != nil { + t.Fatal(err) + } + + anchor := time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC) + if err := store.RecordOutboundHold(ctx, msgID, "policy_budget", anchor); err != nil { + t.Fatalf("record hold: %v", err) + } + p, err = store.ClaimOutboundForSend(ctx, msgID, 4242) + if err != nil || p == nil { + t.Fatalf("re-claim: payload=%v err=%v", p, err) + } + if p.LocalHoldClass != "policy_budget" || p.LocalHoldAnchor == nil || !p.LocalHoldAnchor.Equal(anchor) { + t.Fatalf("hold on re-claim = %q %v, want policy_budget @ %v", p.LocalHoldClass, p.LocalHoldAnchor, anchor) + } + + if err := store.WithTx(ctx, func(tx pgx.Tx) error { + _, err := store.MarkOutboundSentTx(ctx, tx, msgID, "+ Loading... +
+- Loading... -
-- {subtitle} -
- {children} -page body
survey body
survey body
page body
page body
page body