feat(api): fold EmailEffects into EmailApi

EmailApi.sendOnce(request, effectKey) with EmailApiEffectOutcome and
EmailApiEffectResult, lookups by idempotency key and RFC Message-ID, and
rfcMessageId on EmailApiMessage, so other qbits have one public surface.
EmailEffects and its types stay as internal, deprecated delegates; the
durable-jobs guide, README and DESIGN 2.5 point at EmailApi.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -6,6 +6,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and
 ## [Unreleased]
 
 ### Added
+- `EmailApi.sendOnce` as the public durable-effect handoff, with API-only outcome/result types, committed enqueue/replay semantics and suppression/transaction refusal parity. Exact idempotency-key and RFC Message-ID lookups return `EmailApiMessage`, which now includes `rfcMessageId` across all lookup routes. Public requests support ordinary caller transactions; `EmailEffects`, `EmailEffectOutcome` and `EmailMessageRef` are deprecated internals. H2 parity, reconciliation, storage-failure and API-boundary coverage; no new dependencies.
+- Stable `com.kingsrook.qbits.email.api` contract for cross-qbit sends, create-once templates, metadata-only message lookups, enabled sender addresses, default provider selection and secret-safe readiness (including marketing and PHI prerequisites). JDK/QQQ-only signatures, complete request/status mapping tests and a reflection package-boundary guard; no new dependencies.
+- Documented both supported cross-qbit routes: core-only `SendMessageAction` via `qbitEmail`, and the public API for richer sends and lookups.
+- Persisted unique RFC Message-IDs, stable across delivery attempts, with SMTP/Gmail SMTP, SES raw MIME and Postmark/SendGrid/Mailgun request propagation (SES replaces the header downstream).
+- Core-independent `EmailEffects` handoff, reference lookups by idempotency key or RFC Message-ID, refusal mapping and an RC.7 binding design sketch; offline retry and provider-contract tests.
+- Checkstyle now checks test sources as well as main sources; existing test imports and formatting corrected.
+- Provider sender addresses: child table `email_provider_sender_address` (validated lowercase address, display name, one default, enabled, note; unique per provider; domain bound to the provider's sending domain; admin-only), shown on the provider record. `EmailRequest.withFromAddressId`, `from_address_id` on messages, refusal of a from address that is not an enabled row of the resolved provider, and the provider `from_address` / `from_name` kept as the fallback.
+- Contingent possible-value filters on the new from-address pickers (send processes by `providerId`, Send Test Email by the acted-on provider, message `fromAddressId` by `providerId`) and on the attempt and event message pickers (by `providerId`) and the suppression source event (by `sourceMessageId`); the send processes also take `providerId`.
 - Phase (c): authenticated SNS, Postmark, SendGrid and Mailgun webhooks, durable inbox deduplication and Mailgun nonce replay protection; configurable Javalin routes.
 - Atomic scheduled event processing, per-recipient event history, monotonic delivery state, category-aware suppression, optional daily complaint alerts and FAILED inbox Reprocess with Review/Result.
 - One-click unsubscribe and escaped self-contained light/dark confirmation pages using the existing envelope-bound signer; account/security tokens refused.

README.md

@@ -52,6 +52,7 @@ in a static variable.
 | `email_brand` | Brands | `email.manage`; template edit grants also apply |
 | `email_suppression` | Suppressions | `email.manage`; changing/removing a complaint needs `email.admin` |
 | `email_provider` | Providers, with child Messages | `email.admin`, including reads and test sends |
+| `email_provider_sender_address` | Sender Addresses on a provider | `email.admin`, including reads |
 
 `email.view` reads messages, templates, versions, brands and suppressions.
 `email.send` permits the Java service, bridge and backend send process. Reading
@@ -87,6 +88,30 @@ enabled, exactly one must be the enabled default. Switch defaults in a batch upd
 clearing the old default first. A nullable unique `default_slot` backs the validation.
 Providers with message history cannot be deleted; disable them instead.
 
+**Sender addresses.** A provider can own any number of verified from addresses, kept in
+`email_provider_sender_address` and shown as a Sender Addresses list on the provider record
+(not in the nav; `email.admin` reads and writes, like the provider). Each row has an
+`address` (validated, stored lowercase, line breaks refused), an optional `displayName`,
+`isDefault`, `enabled` and a `note`. An address is unique per provider, at most one enabled
+row per provider is the default, and when the provider has a `sendingDomain` every address must
+be in that domain (changing the domain with other-domain rows is refused). Deleting a provider
+deletes its rows. Existing databases add the table and the nullable
+`email_message.from_address_id` column from `docs/ddl/email-postgres.sql`.
+
+The provider's own `fromAddress` / `fromName` stay the fallback, so nothing changes until a
+row exists. A row marked default wins over the fallback. On a send, `withFromAddressId(id)`
+(or `fromAddressId` on the send processes) must name an enabled row of the resolved provider,
+and once a provider has rows an explicit `withFrom(...)` must match one of them by address;
+either failure is `REFUSED` with "That from address is not set up for provider <name>". With
+no rows, `withFrom(...)` behaves as before. The chosen row is recorded on the message as
+`fromAddressId`, and a resend reuses it.
+
+The from address pickers are contingent: `fromAddressId` on the send processes lists only the
+rows of the chosen `providerId` (`providerName` is still accepted for API callers, and the two
+must agree), and the Send Test Email action lists only the rows of the provider being tested.
+That action's form has no record id, so a prepare step publishes the single selected provider id
+as the process value `possibleValueFilterValueId`, which the picker's filter reads after `${input.id}`.
+
 For an unauthenticated relay, use kind `SMTP`, host/port, security `NONE`, and a
 `fromAddress`. Leave credential names empty. Authenticated SMTP uses `usernameEnvName`
 and `secretEnvName`; security is `NONE`, `STARTTLS` (required) or `TLS`. Timeouts are
@@ -132,7 +157,99 @@ and credential failures pause the provider. Tracking is explicitly controlled by
 `trackOpens` / `trackClicks`, off by default. The producer registers the phase-(c)
 receiver idempotently when the provided Javalin middleware is present.
 
-## Sending
+## Sending from another qbit
+
+There are two supported routes. Both queue through the durable outbox; a successful
+call acknowledges enqueueing, not delivery.
+
+**1. QQQ core only, for simple subject/body email.** The host installs qbit-email;
+the consuming qbit needs no qbit-email dependency or imports. Use
+`SendMessageAction` with `messagingProviderName` set to `qbitEmail` (the default
+bridge name; hosts can configure another name):
+
+```java
+import com.kingsrook.qqq.backend.core.actions.messaging.SendMessageAction;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Content;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Party;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.SendMessageInput;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.email.EmailContentRole;
+
+new SendMessageAction().execute(new SendMessageInput()
+   .withMessagingProviderName("qbitEmail")
+   .withTo(new Party().withAddress("pat@example.com"))
+   .withSubject("Ready")
+   .withContent(new Content().withContentRole(EmailContentRole.TEXT).withBody("Your export is ready.")));
+```
+
+**2. The published `com.kingsrook.qbits.email.api` package**, for templates,
+idempotency, categories, linked records, sender selection and delivery lookups.
+Consumers with a qbit-email dependency import only this package:
+
+```java
+import com.kingsrook.qbits.email.api.EmailApi;
+import com.kingsrook.qbits.email.api.EmailApiRequest;
+import com.kingsrook.qbits.email.api.EmailApiResult;
+
+EmailApiResult result = EmailApi.send(new EmailApiRequest()
+   .withTo("pat@example.com")
+   .withTemplateName("order_ready")
+   .withVariable("order_number", "A-1")
+   .withVariable("customer_name", "Pat")
+   .withIdempotencyKey("orders:42:ready")
+   .withLinkedRecord("orders", 42)
+   .withSender("orders", "ready"));
+```
+
+`EmailApi.ensureTemplate(EmailApiTemplate)` creates a template once; see
+[Templates from another qbit](#templates-from-another-qbit).
+
+For durable job effects, use `EmailApi.sendOnce(request, effectKey)`. It assigns
+the effect key as the request's idempotency key and returns `EmailApiEffectResult`
+with `EmailApiEffectOutcome.QUEUED_OR_DUPLICATE` or `REFUSED` and a nullable
+`EmailApiMessage`. Success means the outbox has committed; duplicate handoffs return
+the same row and RFC Message-ID. Null requests, null/blank/over-500-character keys,
+conflicting request keys and caller transactions are refused. Suppressed rows remain
+refused on replay and include their audit metadata; refusals without a row have no
+message. Database failures throw `QException` for retry or reconciliation. Ordinary
+`EmailApi.send` accepts `request.withTransaction(transaction, backendName)` when a
+caller-owned or cross-backend deferred enqueue is needed. See [durable job effects](docs/DURABLE-JOBS.md).
+
+`EmailApi.isInstalled(instance)` detects whether the qbit was produced.
+`defaultProviderName(instance)` resolves an enabled configured/default provider;
+`listSenderAddresses(providerName)` lists enabled sender rows (null selects the
+default). `readiness(instance, category)` reports configuration prerequisites,
+including MARKETING's public URL, unsubscribe key and default brand postal address.
+For a PHI-approved template, use `readiness(instance, category, true)` to also check
+the body key and provider PHI/tracking policy. PHI is a template approval state,
+not a category. Readiness is a preflight, not proof that a particular send will pass.
+
+`getMessage(id)`, `getMessageByIdempotencyKey(key)`, `getMessageByRfcMessageId(rfcMessageId)`
+and `findMessagesByIdempotencyKeyPrefix(prefix, limit)` expose only
+id, delivery status, category, idempotency key, RFC Message-ID, provider message id, template name,
+linked table/id and create/sent/delivered timestamps. They never return addresses,
+subjects, bodies or variable snapshots. Prefixes are literal and nonblank; limits
+must be positive; results are ordered by ascending message id. Missing messages
+return an empty Optional/list. Exact-key lookups also return empty for null/blank
+values; RFC lookup takes the stored bracketed header, not a provider response id.
+Ordinary enqueue results have `QUEUED`, `DUPLICATE`,
+`SUPPRESSED` or `REFUSED`; delivery status is a separate string on `EmailApiMessage`.
+
+Database calls require the caller's initialized `QContext`; methods accepting an
+installed instance require it to be the current context instance. Lookups run as
+the email system, so callers must authorize any exposure to their own users.
+Sending retains the existing permission, schema, PHI, suppression and recipient
+rules. Database failures propagate as core `QException`; invalid template definitions
+and lookup arguments throw `IllegalArgumentException`. No secret values are exposed.
+
+The API uses only JDK, QQQ core and its own API types in published signatures.
+Changes within a `0.x` minor release line are additive only. Internal packages
+(including runtime, configuration, outbox, metadata, support and template classes)
+are not API and may change. `EmailEffects`, `EmailEffectOutcome` and `EmailMessageRef`
+are deprecated internals; `EmailApi` is the sole public qbit-email surface.
+The internal service details below describe host and
+qbit-email implementation behavior, not additional cross-qbit contracts.
+
+## Internal sending service
 
 ```java
 EmailSendResult result = EmailService.send(new EmailRequest()
@@ -246,14 +363,14 @@ preview never stores the result or changes the version.
 ### Templates from another qbit
 
 A qbit that sends its own mail ships the template with its code instead of asking an
-administrator to create it. Describe it with `EmailTemplateDefinition` and call
-`EmailTemplates.ensure` at startup or before first use:
+administrator to create it. Describe it with `EmailApiTemplate` and call
+`EmailApi.ensureTemplate` at startup or before first use:
 
 ```java
-boolean created = EmailTemplates.ensure(new EmailTemplateDefinition()
+boolean created = EmailApi.ensureTemplate(new EmailApiTemplate()
    .withName("order_ready")
    .withLabel("Order ready")
-   .withCategory(EmailCategory.TRANSACTIONAL)
+   .withCategory(EmailApiCategory.TRANSACTIONAL)
    .withSubject("Order {{order_number}} is ready")
    .withBody("Hi {{customer_name}}, order {{order_number}} is ready for pickup.")
    .withVariable("order_number", "Order number", "STRING", true, "PUBLIC")
@@ -265,7 +382,7 @@ boolean created = EmailTemplates.ensure(new EmailTemplateDefinition()
 
 `kind` defaults to MARKDOWN (MJML and HTML are also accepted), `withTextBody` is
 optional, and `withVariablesSchema(json)` replaces `withVariable` when you already have
-the JSON. When no template of that name exists, `ensure` validates the definition like
+the JSON. When no template of that name exists, `ensureTemplate` validates the definition like
 Activate Version (syntax, schema, a render with the samples, MJML compile), creates the
 template and an ACTIVE version 1 as the qbit, and returns true. When one exists it does
 nothing and returns false, so people's edits and later versions are never overwritten;
@@ -424,17 +541,10 @@ The bridge recursively maps Party/MultiParty TO, CC, BCC, FROM and REPLY_TO role
 TEXT/HTML alternatives. REFUSED throws `QException`; successful calls acknowledge
 queueing, not delivery. The built-in EMAIL and SES types are unchanged.
 
-For schema-aware core bridge calls, use the typed extension:
-
-```java
-new SendMessageAction().execute(new EmailTemplatedMessageInput()
-   .withEmailRequest(new EmailRequest().withTemplateName("email_secure_message")
-      .withTo("pat@example.com").withVariable("link", "https://app.example.com/inbox"))
-   .withMessagingProviderName("qbitEmail"));
-```
-
-Ordinary core `SendMessageInput` remains ad-hoc and cannot infer classifications
-from arbitrary pre-rendered text. Use the typed extension for classified variables.
+For schema-aware calls from another qbit, use `EmailApi.send` with an
+`EmailApiRequest` and declared template variables as described above. Ordinary
+core `SendMessageInput` is ad-hoc and cannot infer classifications from arbitrary
+pre-rendered text. The existing typed bridge extension is an internal interface.
 
 ## Configuration
 
@@ -476,6 +586,26 @@ concurrent idempotency, shared rate budgets, health recovery and process screens
 The tests also exercise template lifecycle, parent-scoped possible values, rendering,
 PHI canary egress, encryption, preview widgets and home data.
 
+## Durable jobs
+
+`EmailEffects.send(request, effectIdempotencyKey)` hands a message to the committed
+email outbox exactly once per key and returns an `EmailEffects.Result` containing
+`EmailEffectOutcome` and an `EmailMessageRef`. `QUEUED_OR_DUPLICATE` means the outbox
+accepted responsibility; `REFUSED` means a definite refusal, including suppression.
+Use `findByIdempotencyKey` or `findByRfcMessageId` to reconcile a handoff. These are
+trusted system reads returning only references. Database failures propagate.
+Requests must not carry a caller transaction; the effect dispatcher calls after its
+own chunk commit. Existing `EmailService` transaction support is unchanged.
+
+Each new message stores a unique `rfcMessageId` once. SMTP (including the Gmail preset),
+Postmark, SendGrid and Mailgun carry it on every attempt. SES receives it in the raw
+request but replaces it downstream; every transport retains `X-QQQ-Email-Id`.
+Message-ID enables correlation and receiver deduplication where supported; it does
+not promise exactly-once inbox delivery.
+
+See [the handoff contract, provider references, schema upgrade and RC.7 binding sketch](docs/DURABLE-JOBS.md).
+The build remains on released QQQ **4.1.0-RC.5**, without durable-jobs dependencies.
+
 ## License
 
 Apache License 2.0. See [LICENSE](LICENSE) and [NOTICE](NOTICE).

docs/DESIGN.md

@@ -44,7 +44,7 @@ unsubscribe and PHI rules.
 
 ## 2. Sending API
 
-### 2.1 `EmailService` (Java, for every consumer)
+### 2.1 `EmailService` (internal Java enqueue service)
 
 ```java
 EmailSendResult result = EmailService.send(new EmailRequest()
@@ -54,6 +54,7 @@ EmailSendResult result = EmailService.send(new EmailRequest()
    .withRecordVariable("order", "order", 1234)     // a record, read as the sender; classification known
    .withCategory(EmailCategory.NOTIFICATION)       // default: the template's
    .withProviderName("pmg")                        // optional; else the default provider
+   .withFromAddressId(4)                           // optional enabled email_provider_sender_address row of that provider
    .withBrandName("kof22")                         // optional; else the template's, else the default brand
    .withIdempotencyKey("alerts:notif:991")         // optional; a repeat returns the existing message
    .withLinkedRecord("alerts_alert", 991)          // shown as a link on the message record
@@ -73,7 +74,8 @@ EmailSendResult result = EmailService.send(new EmailRequest()
 
 ### 2.2 Process `email_send_templated`
 The same request as a backend process (template, to, cc, bcc, variables JSON, linked table and
-id, idempotency key, provider, brand) for scripts, the API and other qbits. Not in the nav;
+id, idempotency key, provider by `providerId` or `providerName`, `fromAddressId` filtered by the
+chosen `providerId`, brand) for scripts, the API and other qbits. Not in the nav;
 callable through the API and `/qqq/v1/processes`.
 
 ### 2.3 Permissions
@@ -94,9 +96,58 @@ template may set `edit_permission_name` to restrict who edits it (e.g. marketing
 - Attachments are stored through qbit-object-storage when it is installed, else refused with a
   plain message (no blobs in the outbox, nothing on local disk: D11).
 - Callers (qbit-alerts `EmailChannel` and `NotificationEmailer`, qbit-marketing
-  `QqqMessagingEmailTransport`) can switch by configuration at once, then move to `EmailService`
+  `QqqMessagingEmailTransport`) can switch by configuration at once, then move to `EmailApi`
   in phase (d) for templates, categories, idempotency and linked records.
 
+### 2.5 Public API
+
+`com.kingsrook.qbits.email.api.EmailApi` is the sole public qbit-email surface for consuming qbits.
+Changes within a `0.x` minor release line are additive only. Public/protected
+signatures contain only JDK, QQQ core or API-package types; internal packages are
+not API and may change. A reflection test discovers every compiled API class and
+checks signatures, including generic arguments, bounds and nested types.
+
+- `EmailApi.send(EmailApiRequest)` maps to the existing outbox service, preserving
+  permissions, schema/PHI validation, suppression and one-recipient rules. Requests
+  cover recipients/sender selection, body alternatives, template variables and record
+  provenance, category/provider/brand/locale, idempotency, linked records, producer
+  identity, scheduling and caller transactions (`withTransaction(transaction, backendName)`).
+  Results expose message id, enqueue status and reasons.
+- `sendOnce(EmailApiRequest, effectKey)` provides durable handoff through the existing
+  internal effects implementation. The key becomes the request's idempotency key;
+  null requests, missing/blank/over-500-character keys, mismatches and caller transactions
+  are refused. `EmailApiEffectResult` carries `EmailApiEffectOutcome.QUEUED_OR_DUPLICATE`
+  or `REFUSED` and nullable `EmailApiMessage`. Success means committed outbox storage,
+  and replays retain the row and RFC Message-ID. Suppressed audit rows remain refused
+  on replay and return metadata; refusals without a row return null. Operational
+  failures throw `QException`. The deprecated `EmailEffects`, `EmailEffectOutcome` and
+  `EmailMessageRef` remain internal; consumers use this API. See [durable jobs](DURABLE-JOBS.md).
+- `ensureTemplate(EmailApiTemplate)` delegates create-once template registration;
+  `EmailApiVariable` carries the variable contract. Existing content is never
+  overwritten, and the API does not confer PHI approval.
+- `getMessage`, exact `getMessageByIdempotencyKey`/`getMessageByRfcMessageId` and
+  bounded, literal `findMessagesByIdempotencyKeyPrefix` perform
+  system reads of an explicit metadata projection: id/status/category, idempotency
+  key, RFC Message-ID and provider id, template name, linked record and create/sent/delivered dates.
+  Exact-key lookups return empty for null/blank/missing values. RFC lookup requires
+  the exact bracketed header stored at enqueue time, distinct from provider response ids.
+  No recipients, subjects, bodies or snapshots are fetched or returned by lookups.
+  Callers own authorization before presenting metadata to users.
+- `isInstalled`, `defaultProviderName`, `listSenderAddresses` and `readiness` replace
+  consumer access to runtime/configuration/tables. Readiness checks an enabled
+  default provider; MARKETING also needs a valid publicBaseUrl, available unsubscribe
+  key and default brand postal address. The PHI overload checks body key availability
+  and provider PHI/tracking policy. Reasons never contain resolved secret values.
+  PHI is a template state, so it is an explicit option rather than a fifth category.
+- Database operations use the current `QContext`; an installed instance argument
+  must match that context. Database errors remain core `QException`. Readiness does
+  not validate an individual request or guarantee delivery.
+
+For simple subject/body sends, consumers need no qbit-email dependency: use core
+`SendMessageAction` with `messagingProviderName = "qbitEmail"` (or the host's
+configured bridge name). Use the API package for templates, idempotency, links and
+lookups. Both routes enqueue; neither sends inline.
+
 ## 3. Providers and transports
 
 ### 3.1 `email_provider`
@@ -122,7 +173,16 @@ They are never stored, logged, audited, put in process output or returned. The r
 whether each named variable is set (yes or no), never the value.
 
 **Record action "Send Test Email"** (`email.admin`): queues a fixed test message to a given
-address through that provider and reports the outcome on the standard result screen.
+address through that provider and reports the outcome on the standard result screen. An optional
+`fromAddressId` picker lists only that provider's sender addresses.
+
+**Sender addresses** (`email_provider_sender_address`, a child list on the provider, `email.admin`):
+provider_id, address (validated, lowercase), display_name, is_default (at most one enabled default
+per provider), enabled, note; unique (provider_id, address); the domain must equal `sending_domain`
+when that is set. The provider's `from_address` / `from_name` remain the fallback; an enabled default
+row wins when present. A chosen id, or an explicit from text once the provider has rows, must match an
+enabled row of the resolved provider or the send is `REFUSED` ("That from address is not set up for
+provider <name>"). The row used is kept as `email_message.from_address_id`.
 
 ### 3.2 `EmailTransport` SPI (in the qbit)
 ```java
@@ -174,6 +234,7 @@ transient; other 4xx are permanent.
 | variables_snapshot | JSON, redacted per section 7.3 |
 | html_body, text_body | rendered at queue time so retries send the same bytes; encrypted for `phi_approved`; cleared after `bodyRetentionDays` (default 30) once final, keeping `body_sha256` |
 | idempotency_key | unique when present |
+| rfc_message_id | STRING(500), unique; stored once at enqueue and reused on every attempt; see [durable job effects](DURABLE-JOBS.md) |
 | provider_id, provider_message_id | |
 | attempt_count, next_attempt_date, lease_expires_date, last_error, sent_date, delivered_date, dead_letter_date | |
 | linked_table_name, linked_record_id | rendered as a link on the record screen |
@@ -480,6 +541,7 @@ section's description explains the split with qbit-alerts' Notifications section
 | `email_brand` | yes | people |
 | `email_suppression` | yes | people and the qbit |
 | `email_provider` | yes | admins |
+| `email_provider_sender_address` | relation (on the provider) | admins |
 | `email_webhook_inbox` | relation (on the provider) | the qbit |
 | `email_inbound`, `email_inbound_attachment` | phase (e) | the qbit |
 

docs/DURABLE-JOBS.md

@@ -0,0 +1,129 @@
+# Durable job email effects
+
+This implements the email-side contract for QRun-IO/qqq#1191, spec section 2.13,
+without importing core jobs types. QQQ stays at released 4.1.0-RC.5. The
+`EmailJobEffectTarget implements EffectTarget` binding is deferred until RC.7 ships.
+
+## Stored identity and schema upgrade
+
+`email_message.rfc_message_id` (`rfcMessageId` in metadata) is a unique STRING(500)
+shown in the message T2 details. Before insertion, the service stores
+`<token@domain>`, where `token` is the first 32 lowercase hex characters of SHA-256
+of the UTF-8 idempotency key. Without a key, the seed is `qbit-email:` plus a fresh
+UUID, since the row id is not yet allocated. The domain is the normalized From
+mailbox domain, excluding display names; unavailable or unsuitable domains use
+`qbit-email.local`. The key itself is never included in the header.
+
+The dispatcher reads this stored value on every attempt. JavaMail's
+`updateMessageID()` is overridden so repeated `saveChanges()` calls cannot replace
+it. A manual **Resend** creates a new outbox row and a new identity; a delivery retry
+keeps its identity. `providerMessageId` remains the provider's response identifier,
+distinct from `rfcMessageId` and the numeric outbox `id`.
+
+New installations use [email-postgres.sql](ddl/email-postgres.sql). Existing
+PostgreSQL installations must pause enqueue and dispatch, then apply
+[email-message-id-upgrade.sql](ddl/email-message-id-upgrade.sql) before starting
+this version. The migration backfills old rows using the existing key, or
+`qbit-email:` plus the already allocated row id, and the stored From domain.
+It cannot recover previously generated random transport Message-IDs. Review host
+schema adaptations first; host security columns are unaffected.
+
+## Provider behavior
+
+| Transport | Request mechanism | Identity preservation |
+|---|---|---|
+| SMTP / Gmail SMTP preset | MIME `Message-ID` | Sent unchanged by this transport; downstream relay policy still applies |
+| SES v2 | Raw MIME `Message-ID` | **SES replaces Message-ID downstream** |
+| Postmark | `Headers: [{Name: "Message-ID", Value: "<...>"}]` | Custom Message-ID supported |
+| SendGrid | `headers: {"Message-ID": "<...>"}` | Sent as a custom header |
+| Mailgun | MIME Message-ID plus multipart `h:Message-Id` | Sent in the `/messages.mime` request |
+
+All transports retain `X-QQQ-Email-Id` with the numeric outbox id. Postmark's
+`MessageID`, SendGrid's response `X-Message-Id`, and SES's returned id are provider
+references, not our RFC header. A fixed header enables recognition and receiver
+deduplication where supported, but is not an exactly-once inbox guarantee.
+
+Provider references: [Postmark custom Message-ID](https://postmarkapp.com/developer/user-guide/send-email-with-api/send-a-single-email),
+[SendGrid custom headers](https://www.twilio.com/docs/sendgrid/api-reference/mail-send/mail-send),
+[Mailgun sending parameters](https://documentation.mailgun.com/docs/mailgun/api-reference/send/mailgun/messages/post-v3--domain-name--messages),
+and [SES header replacement](https://docs.aws.amazon.com/ses/latest/APIReference/API_SendRawEmail.html).
+
+## Handoff and reconcile
+
+`EmailApi.sendOnce(EmailApiRequest request, String effectKey)` sets the request's key, refusing
+a different pre-existing key, a missing/blank/over-500-character effect key, a null request
+or a caller transaction. The service performs its usual validation and send
+permission checks. Success is returned only after the outbox transaction commits;
+there is no transport call here. Use ordinary `EmailApi.send` with
+`request.withTransaction(transaction, backendName)` for caller-owned or cross-backend
+deferred transactions. `sendOnce` refuses either kind of caller transaction.
+
+`EmailApiEffectResult(outcome, message)` contains an `EmailApiEffectOutcome` and a
+nullable `EmailApiMessage` with the outbox id, current delivery status, RFC Message-ID,
+provider reference and other metadata. All these types are in
+`com.kingsrook.qbits.email.api`, the sole public qbit-email surface:
+
+- `QUEUED_OR_DUPLICATE`: one committed row accepted the handoff, including replays.
+- `REFUSED`: invalid or suppressed request. A suppressed audit row has a reference;
+  refusals without a stored row have a null reference. Replaying a suppressed row
+  stays refused even though `EmailService` returns `DUPLICATE`.
+- Database/query failures throw `QException`; callers must retry or reconcile,
+  never turn an uncertain outcome into a permanent refusal.
+
+`EmailApi.getMessageByIdempotencyKey(key)` and `EmailApi.getMessageByRfcMessageId(messageId)`
+return `Optional<EmailApiMessage>` from system reads of an explicit metadata projection.
+They return empty for null/blank/missing values. These are trusted backend APIs,
+not viewer-scoped public endpoints. Lookups reflect current status and provider
+reference and do not expose bodies, recipients or variables. `EmailApi.getMessage(id)`
+and `EmailApi.findMessagesByIdempotencyKeyPrefix(prefix, limit)` return the same projection,
+including `rfcMessageId`. Callers must authorize exposing this metadata to their users.
+`EmailEffects`, `EmailEffectOutcome` and `EmailMessageRef` are deprecated internal
+implementation types; consuming qbits use `EmailApi`.
+
+The effect's `SENT` means successful durable handoff, not mailbox delivery. The
+email dispatcher owns later delivery/retry failures. Reconcile acknowledges an
+existing outbox row (including an audit row); it must never create another email
+merely because that row is terminal. Retain message rows and idempotency keys for
+at least the maximum effect retry/reconciliation horizon.
+
+## RC.7 binding design note (not compiled)
+
+Mapping: `deliver -> QUEUED_OR_DUPLICATE => SENT(ref.id)`, `REFUSED => FAILED`;
+`reconcile -> getMessageByIdempotencyKey present => FOUND(ref.id), absent => NOT_FOUND`.
+Exceptions escape to core's retry/unknown handling. Payload decoding must preserve
+the existing PHI/variable policies and must not attach a caller transaction.
+
+This 25-line sketch is documentation-only. Match core names and signatures to the
+published RC.7 contract when it ships; `decode` represents payload conversion.
+
+```java
+// Future RC.7 binding; no core jobs imports in the RC.5 implementation.
+final class EmailJobEffectTarget implements EffectTarget
+{
+   // SENT acknowledges the committed qbit outbox row.
+   @Override
+   public DeliveryResult deliver(Effect effect) throws Exception
+   {
+      EmailApiRequest request = decode(effect.payload());
+      EmailApiEffectResult result = EmailApi.sendOnce(request, effect.idempotencyKey());
+      if(result.outcome() == EmailApiEffectOutcome.REFUSED)
+      {
+         return (DeliveryResult.failed("Email refused"));
+      }
+      return (DeliveryResult.sent(result.message().id().toString()));
+   }
+
+   // Existence reconciles handoff independently of mailbox delivery.
+   @Override
+   public ReconcileResult reconcile(Effect effect) throws Exception
+   {
+      return (EmailApi.getMessageByIdempotencyKey(effect.idempotencyKey())
+         .map(ref -> ReconcileResult.found(ref.id().toString()))
+         .orElseGet(ReconcileResult::notFound));
+   }
+}
+```
+
+Tests use H2, the existing GreenMail SMTP sink, fake SES and HTTP request seams;
+no vendor endpoints are called. The PostgreSQL migration is supplied for host
+deployment and is not executed by the offline Java test gate.

docs/ddl/email-message-id-upgrade.sql

@@ -0,0 +1,11 @@
+-- Existing PostgreSQL installations: pause enqueue and dispatch before applying.
+-- SHA-256 is PostgreSQL's built-in bytea function; no extension is required.
+BEGIN;
+ALTER TABLE email_message ADD COLUMN rfc_message_id VARCHAR(500);
+UPDATE email_message
+SET rfc_message_id = '<' || left(encode(sha256(convert_to(
+   coalesce(idempotency_key, 'qbit-email:' || id::text), 'UTF8')), 'hex'), 32)
+   || '@' || coalesce(lower(substring(from_address from
+      '@([A-Za-z0-9]+(?:[.-][A-Za-z0-9]+)*)(?:>|$)')), 'qbit-email.local') || '>';
+ALTER TABLE email_message ADD CONSTRAINT email_message_rfc_message_id_unique UNIQUE (rfc_message_id);
+COMMIT;

docs/ddl/email-postgres.sql

@@ -38,6 +38,19 @@ CREATE TABLE email_provider (
    UNIQUE (default_slot)
 );
 
+CREATE TABLE email_provider_sender_address (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   provider_id INTEGER NOT NULL,
+   address VARCHAR(500) NOT NULL,
+   display_name VARCHAR(500),
+   is_default BOOLEAN,
+   enabled BOOLEAN,
+   note TEXT,
+   UNIQUE (provider_id, address)
+);
+
 CREATE TABLE email_message (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    create_date TIMESTAMP,
@@ -47,6 +60,7 @@ CREATE TABLE email_message (
    cc_addresses TEXT,
    bcc_addresses TEXT,
    from_address VARCHAR(500),
+   from_address_id INTEGER,
    reply_to VARCHAR(500),
    subject TEXT,
    template_id INTEGER,
@@ -61,6 +75,7 @@ CREATE TABLE email_message (
    phi_approved BOOLEAN,
    unsubscribe_url TEXT,
    idempotency_key VARCHAR(500),
+   rfc_message_id VARCHAR(500),
    provider_id INTEGER,
    provider_message_id VARCHAR(500),
    attempt_count INTEGER,
@@ -79,7 +94,8 @@ CREATE TABLE email_message (
    sender_process VARCHAR(500),
    suppressed_addresses TEXT,
    send_after TIMESTAMP,
-   UNIQUE (idempotency_key)
+   UNIQUE (idempotency_key),
+   UNIQUE (rfc_message_id)
 );
 
 CREATE TABLE email_message_attempt (

pom.xml

@@ -168,7 +168,7 @@
                <failsOnError>false</failsOnError>
                <failOnViolation>true</failOnViolation>
                <violationSeverity>warning</violationSeverity>
-               <includeTestSourceDirectory>false</includeTestSourceDirectory>
+               <includeTestSourceDirectory>true</includeTestSourceDirectory>
             </configuration>
             <executions>
                <execution>

src/main/java/com/kingsrook/qbits/email/EmailQBitProducer.java

@@ -143,6 +143,7 @@ public class EmailQBitProducer implements QBitMetaDataProducer<EmailQBitConfig>
       for(QTableMetaData table : EmailTables.tables(qBitConfig))
       {
          Class<?> customizer = EmailTables.PROVIDER.equals(table.getName()) ? EmailTableCustomizers.ProviderCustomizer.class
+            : EmailTables.SENDER.equals(table.getName()) ? EmailTableCustomizers.SenderAddressCustomizer.class
             : EmailTables.SUPPRESSION.equals(table.getName()) ? EmailTableCustomizers.SuppressionCustomizer.class
             : List.of(EmailTables.TEMPLATE, EmailTables.VERSION, EmailTables.BRAND).contains(table.getName()) ? com.kingsrook.qbits.email.template.EmailTemplateCustomizers.class : null;
          if(customizer != null)

src/main/java/com/kingsrook/qbits/email/api/EmailApi.java

@@ -0,0 +1,420 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.metadata.EmailTableCustomizers;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailEffects;
+import com.kingsrook.qbits.email.outbox.EmailService;
+import com.kingsrook.qbits.email.support.EmailApiMapper;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qbits.email.template.EmailTemplateRenderer;
+import com.kingsrook.qbits.email.template.EmailTemplates;
+import com.kingsrook.qbits.email.template.VariableSchema;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+
+
+/*******************************************************************************
+ ** Stable contract for other qbits. Changes within a 0.x minor release line are
+ ** additive only; implementation packages are not API and may change.
+ ** Database operations use the current QContext. Reads run as the email system
+ ** and return only the documented projection; callers must authorize any
+ ** exposure of this metadata to their own users. Sends retain all existing
+ ** permission, validation, PHI, suppression and recipient checks.
+ *******************************************************************************/
+public final class EmailApi
+{
+   private static final Set<String> MESSAGE_FIELDS = Set.of("id", "status", "category", "idempotencyKey", "rfcMessageId", "providerMessageId",
+      "templateId", "linkedTableName", "linkedRecordId", "createDate", "sentDate", "deliveredDate");
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private EmailApi()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** True only when the email qbit was produced in the supplied instance.
+    ** Does not require a current context; null is not installed.
+    *******************************************************************************/
+   public static boolean isInstalled(QInstance instance)
+   {
+      return (EmailRuntime.config(instance) != null);
+   }
+
+
+
+   /*******************************************************************************
+    ** Queue a message. QUEUED is not a delivery acknowledgement. Null requests
+    ** are refused by the same service used by core messaging.
+    *******************************************************************************/
+   public static EmailApiResult send(EmailApiRequest request) throws QException
+   {
+      return (EmailApiMapper.result(EmailService.send(EmailApiMapper.request(request))));
+   }
+
+
+
+   /*******************************************************************************
+    ** Commit one durable handoff with the effect key as the idempotency key.
+    ** Refuse null requests, blank or over-500-character keys, conflicting keys
+    ** and caller transactions before changing the request. Suppression is REFUSED,
+    ** including replays; a retained audit row is returned. Success acknowledges
+    ** committed storage, not delivery. Database failures propagate as QException.
+    *******************************************************************************/
+   public static EmailApiEffectResult sendOnce(EmailApiRequest request, String effectKey) throws QException
+   {
+      if(request == null || effectKey == null || effectKey.isBlank() || effectKey.length() > 500
+         || request.getTransaction() != null
+         || request.getIdempotencyKey() != null && !request.getIdempotencyKey().equals(effectKey))
+      {
+         return (new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+      }
+      request.withIdempotencyKey(effectKey);
+      var result = EmailEffects.send(EmailApiMapper.request(request), effectKey);
+      EmailApiMessage message = result.message() == null ? null : getMessage(result.message().id())
+         .orElseThrow(() -> new QException("Queued email reference is unavailable"));
+      return (new EmailApiEffectResult(EmailApiEffectOutcome.valueOf(result.outcome().name()), message));
+   }
+
+
+
+   /*******************************************************************************
+    ** Create a validated template and active version once. Returns false when
+    ** it already exists. Invalid definitions throw IllegalArgumentException.
+    ** Does not grant PHI approval or overwrite administrator edits.
+    *******************************************************************************/
+   public static boolean ensureTemplate(EmailApiTemplate template) throws QException
+   {
+      return (EmailTemplates.ensure(EmailApiMapper.template(template)));
+   }
+
+
+
+   /*******************************************************************************
+    ** System lookup of delivery metadata. Null and unknown ids return empty.
+    *******************************************************************************/
+   public static Optional<EmailApiMessage> getMessage(Integer messageId) throws QException
+   {
+      if(messageId == null)
+      {
+         return (Optional.empty());
+      }
+      List<QRecord> rows = read(EmailTables.MESSAGE, EmailStore.equal("id", messageId).withLimit(1), MESSAGE_FIELDS);
+      return (rows.isEmpty() ? Optional.empty() : Optional.of(message(rows.getFirst())));
+   }
+
+
+
+   /*******************************************************************************
+    ** System lookup by exact effect or application key. Null, blank and unknown
+    ** keys return empty. Returns current metadata without message content.
+    *******************************************************************************/
+   public static Optional<EmailApiMessage> getMessageByIdempotencyKey(String key) throws QException
+   {
+      return (findMessage("idempotencyKey", key));
+   }
+
+
+
+   /*******************************************************************************
+    ** System lookup by the exact bracketed RFC Message-ID stored at enqueue time.
+    ** Null, blank and unknown ids return empty; provider response ids are distinct.
+    *******************************************************************************/
+   public static Optional<EmailApiMessage> getMessageByRfcMessageId(String rfcMessageId) throws QException
+   {
+      return (findMessage("rfcMessageId", rfcMessageId));
+   }
+
+
+
+   /*******************************************************************************
+    ** Share exact-key lookup semantics and the metadata-only projection.
+    *******************************************************************************/
+   private static Optional<EmailApiMessage> findMessage(String field, String value) throws QException
+   {
+      if(value == null || value.isBlank())
+      {
+         return (Optional.empty());
+      }
+      List<QRecord> rows = read(EmailTables.MESSAGE, EmailStore.equal(field, value).withLimit(1), MESSAGE_FIELDS);
+      return (rows.isEmpty() ? Optional.empty() : Optional.of(message(rows.getFirst())));
+   }
+
+
+
+   /*******************************************************************************
+    ** System lookup by literal prefix, ordered by ascending message id. Prefix
+    ** must be nonblank and limit positive; invalid arguments are rejected.
+    ** Returns at most limit rows, without message content or recipient data.
+    *******************************************************************************/
+   public static List<EmailApiMessage> findMessagesByIdempotencyKeyPrefix(String prefix, int limit) throws QException
+   {
+      if(prefix == null || prefix.isBlank() || limit < 1)
+      {
+         throw (new IllegalArgumentException("A nonblank idempotency key prefix and a positive limit are required"));
+      }
+      ///////////////////////////////////////////////////////////////////////////
+      // QQQ SQL STARTS_WITH uses LIKE without escaping. Narrow by the literal  //
+      // portion before any wildcard, then check the full prefix in Java. Page //
+      // by id so wildcard false positives cannot consume the requested limit. //
+      ///////////////////////////////////////////////////////////////////////////
+      String candidatePrefix = prefix.split("[%_\\\\]", 2)[0];
+      List<EmailApiMessage> messages = new ArrayList<>();
+      int lastId = 0;
+      while(messages.size() < limit)
+      {
+         int batchSize = Math.min(500, limit - messages.size());
+         QQueryFilter filter = new QQueryFilter(new QFilterCriteria("idempotencyKey", QCriteriaOperator.STARTS_WITH, candidatePrefix))
+            .withCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId))
+            .withOrderBy(new QFilterOrderBy("id")).withLimit(batchSize);
+         List<QRecord> rows = read(EmailTables.MESSAGE, filter, MESSAGE_FIELDS);
+         for(QRecord row : rows)
+         {
+            lastId = row.getValueInteger("id");
+            if(row.getValueString("idempotencyKey") != null && row.getValueString("idempotencyKey").startsWith(prefix))
+            {
+               messages.add(message(row));
+            }
+         }
+         if(rows.size() < batchSize)
+         {
+            break;
+         }
+      }
+      return (List.copyOf(messages));
+   }
+
+
+
+   /*******************************************************************************
+    ** Enabled sender rows of an enabled named provider, ordered by id. Null
+    ** selects the effective default provider. No match returns an empty list.
+    ** The provider's legacy from-address fallback is not a sender row.
+    *******************************************************************************/
+   public static List<EmailApiSenderAddress> listSenderAddresses(String providerName) throws QException
+   {
+      QRecord provider = provider(QContext.getQInstance(), providerName);
+      if(provider == null)
+      {
+         return (List.of());
+      }
+      QQueryFilter filter = EmailStore.equal("providerId", provider.getValueInteger("id"))
+         .withCriteria(new QFilterCriteria("enabled", QCriteriaOperator.EQUALS, true)).withOrderBy(new QFilterOrderBy("id"));
+      return (read(EmailTables.SENDER, filter, Set.of("id", "address", "displayName", "isDefault"))
+         .stream().map(row -> new EmailApiSenderAddress(row.getValueInteger("id"), row.getValueString("address"),
+            row.getValueString("displayName"), Boolean.TRUE.equals(row.getValueBoolean("isDefault")))).toList());
+   }
+
+
+
+   /*******************************************************************************
+    ** Effective enabled default: the configured name takes precedence over the
+    ** default row. Empty when absent, disabled or ambiguous. An installed
+    ** instance must be the current QContext instance, otherwise this throws.
+    *******************************************************************************/
+   public static Optional<String> defaultProviderName(QInstance instance) throws QException
+   {
+      QRecord provider = provider(instance, null);
+      return (provider == null ? Optional.empty() : Optional.of(provider.getValueString("name")));
+   }
+
+
+
+   /*******************************************************************************
+    ** Preflight the default sending configuration for a category. Null category
+    ** means TRANSACTIONAL. This does not validate a particular template,
+    ** recipient or permission, or contact a transport. PHI callers use the
+    ** overload with phiApproved set to true.
+    *******************************************************************************/
+   public static EmailApiReadiness readiness(QInstance instance, EmailApiCategory category) throws QException
+   {
+      return (readiness(instance, category, false));
+   }
+
+
+
+   /*******************************************************************************
+    ** Also check prerequisites for a PHI-approved template. PHI is a template
+    ** property, not an email category. Checks body key availability and provider
+    ** PHI/tracking policy; does not approve a template. Resolver failures are
+    ** reported as missing keys without exposing exception details or values.
+    *******************************************************************************/
+   public static EmailApiReadiness readiness(QInstance instance, EmailApiCategory category, boolean phiApproved) throws QException
+   {
+      List<String> reasons = new ArrayList<>();
+      var config = EmailRuntime.config(instance);
+      if(config == null)
+      {
+         return (new EmailApiReadiness(false, List.of("Email qbit is not installed")));
+      }
+      QRecord provider = provider(instance, null);
+      if(provider == null)
+      {
+         reasons.add("An enabled named or default provider is required");
+      }
+      else
+      {
+         if(!EmailTableCustomizers.validateProvider(provider, config).isEmpty())
+         {
+            reasons.add("The default provider configuration is invalid");
+         }
+         if(phiApproved)
+         {
+            try
+            {
+               EmailTemplateRenderer.checkProvider(provider);
+            }
+            catch(IllegalArgumentException e)
+            {
+               reasons.add("PHI email needs a PHI-enabled provider with open and click tracking disabled");
+            }
+         }
+      }
+      if(category == EmailApiCategory.MARKETING)
+      {
+         try
+         {
+            if(config.getPublicBaseUrl() == null || config.getPublicBaseUrl().isBlank())
+            {
+               throw (new IllegalArgumentException());
+            }
+            VariableSchema.url(config.getPublicBaseUrl());
+         }
+         catch(IllegalArgumentException e)
+         {
+            reasons.add("MARKETING email needs a valid publicBaseUrl");
+         }
+         if(!hasSecret(config.getSecretResolver(), "QQQ_EMAIL_UNSUBSCRIBE_KEY"))
+         {
+            reasons.add("MARKETING email needs QQQ_EMAIL_UNSUBSCRIBE_KEY");
+         }
+         List<QRecord> brands = read(EmailTables.BRAND, EmailStore.equal("isDefault", true).withLimit(2), Set.of("id", "postalAddress"));
+         if(brands.size() != 1 || brands.getFirst().getValueString("postalAddress") == null || brands.getFirst().getValueString("postalAddress").isBlank())
+         {
+            reasons.add("MARKETING email needs a default brand with a postal address");
+         }
+      }
+      if(phiApproved && !hasSecret(config.getSecretResolver(), "QQQ_EMAIL_BODY_KEY"))
+      {
+         reasons.add("PHI email needs QQQ_EMAIL_BODY_KEY");
+      }
+      return (new EmailApiReadiness(reasons.isEmpty(), reasons));
+   }
+
+
+
+   /*******************************************************************************
+    ** Presence only; neither retain values nor expose resolver failures.
+    *******************************************************************************/
+   private static boolean hasSecret(Function<String, String> resolver, String name)
+   {
+      try
+      {
+         String value = resolver.apply(name);
+         return (value != null && !value.isBlank());
+      }
+      catch(RuntimeException e)
+      {
+         return (false);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Match the enqueue service's configured-name/default-row precedence.
+    ** Refuse a mismatched context instead of reading another instance's data.
+    *******************************************************************************/
+   private static QRecord provider(QInstance instance, String name) throws QException
+   {
+      var config = EmailRuntime.config(instance);
+      if(config == null)
+      {
+         return (null);
+      }
+      if(instance != QContext.getQInstance())
+      {
+         throw (new IllegalArgumentException("The supplied instance must be the current QContext instance"));
+      }
+      String effectiveName = name == null ? config.getDefaultProviderName() : name;
+      QQueryFilter filter = EmailStore.equal("enabled", true)
+         .withCriteria(new QFilterCriteria(effectiveName == null ? "isDefault" : "name", QCriteriaOperator.EQUALS, effectiveName == null ? true : effectiveName))
+         .withLimit(2);
+      List<QRecord> providers = SystemWrites.query(EmailTables.PROVIDER, filter);
+      return (providers.size() == 1 ? providers.getFirst() : null);
+   }
+
+
+
+   /*******************************************************************************
+    ** Read an explicit projection; message bodies are never fetched here.
+    *******************************************************************************/
+   private static List<QRecord> read(String table, QQueryFilter filter, Set<String> fields) throws QException
+   {
+      return (SystemWrites.call(() -> new QueryAction().execute(new QueryInput(table).withFilter(filter)
+         .withFieldNamesToInclude(fields).withShouldFetchHeavyFields(false).withShouldOmitHiddenFields(false)).getRecords()));
+   }
+
+
+
+   /*******************************************************************************
+    ** Resolve the template name and copy only the approved metadata fields.
+    *******************************************************************************/
+   private static EmailApiMessage message(QRecord row) throws QException
+   {
+      String templateName = null;
+      if(row.getValueInteger("templateId") != null)
+      {
+         List<QRecord> templates = read(EmailTables.TEMPLATE, EmailStore.equal("id", row.getValueInteger("templateId")).withLimit(1), Set.of("id", "name"));
+         if(!templates.isEmpty())
+         {
+            templateName = templates.getFirst().getValueString("name");
+         }
+      }
+      String category = row.getValueString("category");
+      return (new EmailApiMessage(row.getValueInteger("id"), row.getValueString("status"), category == null ? null : EmailApiCategory.valueOf(category),
+         row.getValueString("idempotencyKey"), row.getValueString("rfcMessageId"), row.getValueString("providerMessageId"), templateName, row.getValueString("linkedTableName"),
+         row.getValueString("linkedRecordId"), row.getValueInstant("createDate"), row.getValueInstant("sentDate"), row.getValueInstant("deliveredDate")));
+   }
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiCategory.java

@@ -0,0 +1,33 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** Stable sending categories; PHI approval belongs to the template, not the category.
+ *******************************************************************************/
+public enum EmailApiCategory
+{
+   TRANSACTIONAL,
+   NOTIFICATION,
+   MARKETING,
+   SECURITY
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiEffectOutcome.java

@@ -0,0 +1,31 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** Durable handoff outcomes. Operational failures propagate to the caller.
+ *******************************************************************************/
+public enum EmailApiEffectOutcome
+{
+   QUEUED_OR_DUPLICATE,
+   REFUSED
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiEffectResult.java

@@ -0,0 +1,30 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** Durable handoff result. The message is null when no outbox row was retained.
+ ** Suppressed audit rows carry metadata even though the outcome is REFUSED.
+ *******************************************************************************/
+public record EmailApiEffectResult(EmailApiEffectOutcome outcome, EmailApiMessage message)
+{
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiMessage.java

@@ -0,0 +1,34 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.time.Instant;
+
+
+/*******************************************************************************
+ ** Immutable delivery metadata. No addresses, subject, bodies or variable snapshots.
+ ** Status is the persisted delivery state, distinct from EmailApiStatus enqueue outcomes.
+ *******************************************************************************/
+public record EmailApiMessage(Integer id, String status, EmailApiCategory category, String idempotencyKey, String rfcMessageId, String providerMessageId,
+   String templateName, String linkedTableName, String linkedRecordId, Instant createDate, Instant sentDate, Instant deliveredDate)
+{
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiReadiness.java

@@ -0,0 +1,39 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.util.List;
+
+
+/*******************************************************************************
+ ** Configuration preflight, not a delivery guarantee. Reasons never contain secret values.
+ *******************************************************************************/
+public record EmailApiReadiness(boolean ready, List<String> reasons)
+{
+   /*******************************************************************************
+    ** Copy configuration reasons.
+    *******************************************************************************/
+   public EmailApiReadiness
+   {
+      reasons = List.copyOf(reasons);
+   }
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiRequest.java

@@ -0,0 +1,670 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+
+
+/*******************************************************************************
+ ** Stable enqueue request. One send always goes through the outbox.
+ ** Null category means TRANSACTIONAL for ad-hoc mail; templates select their category.
+ ** Recipient lists and variable maps are mutable request-building state.
+ *******************************************************************************/
+public final class EmailApiRequest
+{
+   private List<String> to = new ArrayList<>();
+   private List<String> cc = new ArrayList<>();
+   private List<String> bcc = new ArrayList<>();
+   private String from;
+   private Integer fromAddressId;
+   private String replyTo;
+   private String subject;
+   private String markdownBody;
+   private String htmlBody;
+   private String textBody;
+   private String providerName;
+   private String templateName;
+   private String brandName;
+   private String locale = "en";
+   private EmailApiCategory category;
+   private String idempotencyKey;
+   private QBackendTransaction transaction;
+   private String transactionBackendName;
+   private Instant sendAfter;
+   private String linkedTableName;
+   private String linkedRecordId;
+   private String senderQbit;
+   private String senderProcess;
+   private Map<String, Serializable> variables = new LinkedHashMap<>();
+   private Map<String, RecordVariable> recordVariables = new LinkedHashMap<>();
+
+
+
+   /*******************************************************************************
+    ** Getter for to.
+    *******************************************************************************/
+   public List<String> getTo()
+   {
+      return (to);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for to.
+    *******************************************************************************/
+   public EmailApiRequest withTo(List<String> to)
+   {
+      this.to = to;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for cc.
+    *******************************************************************************/
+   public List<String> getCc()
+   {
+      return (cc);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for cc.
+    *******************************************************************************/
+   public EmailApiRequest withCc(List<String> cc)
+   {
+      this.cc = cc;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for bcc.
+    *******************************************************************************/
+   public List<String> getBcc()
+   {
+      return (bcc);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for bcc.
+    *******************************************************************************/
+   public EmailApiRequest withBcc(List<String> bcc)
+   {
+      this.bcc = bcc;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for from.
+    *******************************************************************************/
+   public String getFrom()
+   {
+      return (from);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for from.
+    *******************************************************************************/
+   public EmailApiRequest withFrom(String from)
+   {
+      this.from = from;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for fromAddressId.
+    *******************************************************************************/
+   public Integer getFromAddressId()
+   {
+      return (fromAddressId);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for fromAddressId.
+    *******************************************************************************/
+   public EmailApiRequest withFromAddressId(Integer fromAddressId)
+   {
+      this.fromAddressId = fromAddressId;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for replyTo.
+    *******************************************************************************/
+   public String getReplyTo()
+   {
+      return (replyTo);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for replyTo.
+    *******************************************************************************/
+   public EmailApiRequest withReplyTo(String replyTo)
+   {
+      this.replyTo = replyTo;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for subject.
+    *******************************************************************************/
+   public String getSubject()
+   {
+      return (subject);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for subject.
+    *******************************************************************************/
+   public EmailApiRequest withSubject(String subject)
+   {
+      this.subject = subject;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for markdownBody.
+    *******************************************************************************/
+   public String getMarkdownBody()
+   {
+      return (markdownBody);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for markdownBody.
+    *******************************************************************************/
+   public EmailApiRequest withMarkdownBody(String markdownBody)
+   {
+      this.markdownBody = markdownBody;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for htmlBody.
+    *******************************************************************************/
+   public String getHtmlBody()
+   {
+      return (htmlBody);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for htmlBody.
+    *******************************************************************************/
+   public EmailApiRequest withHtmlBody(String htmlBody)
+   {
+      this.htmlBody = htmlBody;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for textBody.
+    *******************************************************************************/
+   public String getTextBody()
+   {
+      return (textBody);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for textBody.
+    *******************************************************************************/
+   public EmailApiRequest withTextBody(String textBody)
+   {
+      this.textBody = textBody;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for providerName.
+    *******************************************************************************/
+   public String getProviderName()
+   {
+      return (providerName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for providerName.
+    *******************************************************************************/
+   public EmailApiRequest withProviderName(String providerName)
+   {
+      this.providerName = providerName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for templateName.
+    *******************************************************************************/
+   public String getTemplateName()
+   {
+      return (templateName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for templateName.
+    *******************************************************************************/
+   public EmailApiRequest withTemplateName(String templateName)
+   {
+      this.templateName = templateName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for brandName.
+    *******************************************************************************/
+   public String getBrandName()
+   {
+      return (brandName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for brandName.
+    *******************************************************************************/
+   public EmailApiRequest withBrandName(String brandName)
+   {
+      this.brandName = brandName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for locale.
+    *******************************************************************************/
+   public String getLocale()
+   {
+      return (locale);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for locale.
+    *******************************************************************************/
+   public EmailApiRequest withLocale(String locale)
+   {
+      this.locale = locale;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for category.
+    *******************************************************************************/
+   public EmailApiCategory getCategory()
+   {
+      return (category);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for category.
+    *******************************************************************************/
+   public EmailApiRequest withCategory(EmailApiCategory category)
+   {
+      this.category = category;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for idempotencyKey.
+    *******************************************************************************/
+   public String getIdempotencyKey()
+   {
+      return (idempotencyKey);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for idempotencyKey.
+    *******************************************************************************/
+   public EmailApiRequest withIdempotencyKey(String idempotencyKey)
+   {
+      this.idempotencyKey = idempotencyKey;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for the optional caller transaction. sendOnce refuses this option.
+    *******************************************************************************/
+   public QBackendTransaction getTransaction()
+   {
+      return (transaction);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for the declared backend of the caller transaction.
+    *******************************************************************************/
+   public String getTransactionBackendName()
+   {
+      return (transactionBackendName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Join a caller transaction for ordinary send; another backend defers enqueue
+    ** until commit. Declare its backend because QBackendTransaction does not expose
+    ** it. sendOnce refuses any caller transaction to guarantee committed handoff.
+    *******************************************************************************/
+   public EmailApiRequest withTransaction(QBackendTransaction transaction, String backendName)
+   {
+      this.transaction = transaction;
+      this.transactionBackendName = backendName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for sendAfter.
+    *******************************************************************************/
+   public Instant getSendAfter()
+   {
+      return (sendAfter);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for sendAfter.
+    *******************************************************************************/
+   public EmailApiRequest withSendAfter(Instant sendAfter)
+   {
+      this.sendAfter = sendAfter;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for linkedTableName.
+    *******************************************************************************/
+   public String getLinkedTableName()
+   {
+      return (linkedTableName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for linkedTableName.
+    *******************************************************************************/
+   public EmailApiRequest withLinkedTableName(String linkedTableName)
+   {
+      this.linkedTableName = linkedTableName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for linkedRecordId.
+    *******************************************************************************/
+   public String getLinkedRecordId()
+   {
+      return (linkedRecordId);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for linkedRecordId.
+    *******************************************************************************/
+   public EmailApiRequest withLinkedRecordId(String linkedRecordId)
+   {
+      this.linkedRecordId = linkedRecordId;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for senderQbit.
+    *******************************************************************************/
+   public String getSenderQbit()
+   {
+      return (senderQbit);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for senderQbit.
+    *******************************************************************************/
+   public EmailApiRequest withSenderQbit(String senderQbit)
+   {
+      this.senderQbit = senderQbit;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for senderProcess.
+    *******************************************************************************/
+   public String getSenderProcess()
+   {
+      return (senderProcess);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for senderProcess.
+    *******************************************************************************/
+   public EmailApiRequest withSenderProcess(String senderProcess)
+   {
+      this.senderProcess = senderProcess;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for variables.
+    *******************************************************************************/
+   public Map<String, Serializable> getVariables()
+   {
+      return (variables);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for variables.
+    *******************************************************************************/
+   public EmailApiRequest withVariables(Map<String, Serializable> variables)
+   {
+      this.variables = variables;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for recordVariables.
+    *******************************************************************************/
+   public Map<String, RecordVariable> getRecordVariables()
+   {
+      return (recordVariables);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for recordVariables.
+    *******************************************************************************/
+   public EmailApiRequest withRecordVariables(Map<String, RecordVariable> recordVariables)
+   {
+      this.recordVariables = recordVariables;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Append to recipients. Sending still enforces category and PHI recipient rules.
+    *******************************************************************************/
+   public EmailApiRequest withTo(String... addresses)
+   {
+      to.addAll(List.of(addresses));
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Append cc recipients. Sending still enforces category and PHI recipient rules.
+    *******************************************************************************/
+   public EmailApiRequest withCc(String... addresses)
+   {
+      cc.addAll(List.of(addresses));
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Append bcc recipients. Sending still enforces category and PHI recipient rules.
+    *******************************************************************************/
+   public EmailApiRequest withBcc(String... addresses)
+   {
+      bcc.addAll(List.of(addresses));
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Link the message to an application record.
+    *******************************************************************************/
+   public EmailApiRequest withLinkedRecord(String table, Object id)
+   {
+      linkedTableName = table;
+      linkedRecordId = id == null ? null : String.valueOf(id);
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Identify the producing qbit and process.
+    *******************************************************************************/
+   public EmailApiRequest withSender(String qbit, String process)
+   {
+      senderQbit = qbit;
+      senderProcess = process;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Bind one variable; template schema and classification checks still apply.
+    *******************************************************************************/
+   public EmailApiRequest withVariable(String name, Serializable value)
+   {
+      variables.put(name, value);
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Bind record provenance without reading or snapshotting the record.
+    *******************************************************************************/
+   public EmailApiRequest withRecordVariable(String name, String table, Serializable pk)
+   {
+      recordVariables.put(name, new RecordVariable(table, pk));
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** A record reference; values are resolved and classified by the sending service.
+    *******************************************************************************/
+   public record RecordVariable(String table, Serializable id)
+   {
+   }
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiResult.java

@@ -0,0 +1,39 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.util.List;
+
+
+/*******************************************************************************
+ ** Immutable enqueue result with plain refusal reasons.
+ *******************************************************************************/
+public record EmailApiResult(Integer messageId, EmailApiStatus status, List<String> reasons)
+{
+   /*******************************************************************************
+    ** Copy reasons so callers cannot mutate an outcome.
+    *******************************************************************************/
+   public EmailApiResult
+   {
+      reasons = List.copyOf(reasons);
+   }
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiSenderAddress.java

@@ -0,0 +1,29 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** An enabled provider sender address; no provider credentials.
+ *******************************************************************************/
+public record EmailApiSenderAddress(Integer id, String address, String displayName, boolean isDefault)
+{
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiStatus.java

@@ -0,0 +1,33 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** Enqueue outcomes. QUEUED acknowledges storage, not delivery.
+ *******************************************************************************/
+public enum EmailApiStatus
+{
+   QUEUED,
+   DUPLICATE,
+   SUPPRESSED,
+   REFUSED
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiTemplate.java

@@ -0,0 +1,319 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+/*******************************************************************************
+ ** A create-once template definition. Existing templates and versions are never overwritten.
+ ** New templates are not PHI approved; approval remains a separately authorized operation.
+ *******************************************************************************/
+public final class EmailApiTemplate
+{
+   private String name;
+   private String label;
+   private EmailApiCategory category;
+   private String kind = "MARKDOWN";
+   private String subject;
+   private String body;
+   private String textBody;
+   private String variablesSchema;
+   private List<EmailApiVariable> variables = new ArrayList<>();
+   private String sampleVariables;
+   private String ownerQbit;
+   private String editPermissionName;
+
+
+
+   /*******************************************************************************
+    ** Getter for name.
+    *******************************************************************************/
+   public String getName()
+   {
+      return (name);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for name.
+    *******************************************************************************/
+   public EmailApiTemplate withName(String name)
+   {
+      this.name = name;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for label.
+    *******************************************************************************/
+   public String getLabel()
+   {
+      return (label);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for label.
+    *******************************************************************************/
+   public EmailApiTemplate withLabel(String label)
+   {
+      this.label = label;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for category.
+    *******************************************************************************/
+   public EmailApiCategory getCategory()
+   {
+      return (category);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for category.
+    *******************************************************************************/
+   public EmailApiTemplate withCategory(EmailApiCategory category)
+   {
+      this.category = category;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for kind.
+    *******************************************************************************/
+   public String getKind()
+   {
+      return (kind);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for kind.
+    *******************************************************************************/
+   public EmailApiTemplate withKind(String kind)
+   {
+      this.kind = kind;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for subject.
+    *******************************************************************************/
+   public String getSubject()
+   {
+      return (subject);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for subject.
+    *******************************************************************************/
+   public EmailApiTemplate withSubject(String subject)
+   {
+      this.subject = subject;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for body.
+    *******************************************************************************/
+   public String getBody()
+   {
+      return (body);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for body.
+    *******************************************************************************/
+   public EmailApiTemplate withBody(String body)
+   {
+      this.body = body;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for textBody.
+    *******************************************************************************/
+   public String getTextBody()
+   {
+      return (textBody);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for textBody.
+    *******************************************************************************/
+   public EmailApiTemplate withTextBody(String textBody)
+   {
+      this.textBody = textBody;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for variablesSchema.
+    *******************************************************************************/
+   public String getVariablesSchema()
+   {
+      return (variablesSchema);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for variablesSchema.
+    *******************************************************************************/
+   public EmailApiTemplate withVariablesSchema(String variablesSchema)
+   {
+      this.variablesSchema = variablesSchema;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for variables.
+    *******************************************************************************/
+   public List<EmailApiVariable> getVariables()
+   {
+      return (variables);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for variables.
+    *******************************************************************************/
+   public EmailApiTemplate withVariables(List<EmailApiVariable> variables)
+   {
+      this.variables = variables;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for sampleVariables.
+    *******************************************************************************/
+   public String getSampleVariables()
+   {
+      return (sampleVariables);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for sampleVariables.
+    *******************************************************************************/
+   public EmailApiTemplate withSampleVariables(String sampleVariables)
+   {
+      this.sampleVariables = sampleVariables;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for ownerQbit.
+    *******************************************************************************/
+   public String getOwnerQbit()
+   {
+      return (ownerQbit);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for ownerQbit.
+    *******************************************************************************/
+   public EmailApiTemplate withOwnerQbit(String ownerQbit)
+   {
+      this.ownerQbit = ownerQbit;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for editPermissionName.
+    *******************************************************************************/
+   public String getEditPermissionName()
+   {
+      return (editPermissionName);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for editPermissionName.
+    *******************************************************************************/
+   public EmailApiTemplate withEditPermissionName(String editPermissionName)
+   {
+      this.editPermissionName = editPermissionName;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Append a typed variable declaration. Use this or variablesSchema, not both.
+    *******************************************************************************/
+   public EmailApiTemplate withVariable(EmailApiVariable variable)
+   {
+      variables.add(variable);
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Append a variable declaration using the template schema vocabulary.
+    *******************************************************************************/
+   public EmailApiTemplate withVariable(String name, String label, String type, boolean required, String classification)
+   {
+      return (withVariable(new EmailApiVariable(name, label, type, required, classification)));
+   }
+}

src/main/java/com/kingsrook/qbits/email/api/EmailApiVariable.java

@@ -0,0 +1,30 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+/*******************************************************************************
+ ** Template variable declaration. Types: STRING, NUMBER, DATE, DATE_TIME, BOOLEAN,
+ ** URL, LIST, RECORD. Classification: PUBLIC, CONFIDENTIAL, PII, PHI or null.
+ *******************************************************************************/
+public record EmailApiVariable(String name, String label, String type, boolean required, String classification)
+{
+}

src/main/java/com/kingsrook/qbits/email/delivery/EmailDispatcher.java

@@ -451,7 +451,7 @@ public class EmailDispatcher
     *******************************************************************************/
    private static EmailEnvelope envelope(QRecord message)
    {
-      return (new EmailEnvelope().withMessageId(message.getValueString("id")).withFrom(message.getValueString("fromAddress"))
+      return (new EmailEnvelope().withMessageId(message.getValueString("id")).withRfcMessageId(message.getValueString("rfcMessageId")).withFrom(message.getValueString("fromAddress"))
          .withTo(split(message.getValueString("toAddresses"))).withCc(split(message.getValueString("ccAddresses"))).withBcc(split(message.getValueString("bccAddresses")))
          .withReplyTo(message.getValueString("replyTo")).withSubject(content(message, "subject"))
          .withHtmlBody(content(message, "htmlBody"))

src/main/java/com/kingsrook/qbits/email/metadata/EmailTableCustomizers.java

@@ -21,23 +21,32 @@
 package com.kingsrook.qbits.email.metadata;
 
 
+import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import com.kingsrook.qbits.email.EmailQBitConfig;
 import com.kingsrook.qbits.email.EmailRuntime;
 import com.kingsrook.qbits.email.permissions.EmailPermissionChecker;
 import com.kingsrook.qbits.email.permissions.EmailPermissions;
+import com.kingsrook.qbits.email.support.EmailStore;
 import com.kingsrook.qbits.email.support.SystemWrites;
 import com.kingsrook.qbits.email.transport.EmailProviderSettings;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
 import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.QueryOrGetInputInterface;
 import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
 import jakarta.mail.internet.InternetAddress;
@@ -197,6 +206,13 @@ public final class EmailTableCustomizers
                   record.setValue(field, null);
                }
             }
+            String sendingDomain = merged.getValueString("sendingDomain");
+            if(sendingDomain != null && !sendingDomain.isBlank() && merged.getValue("id") != null
+               && SystemWrites.query(EmailTables.SENDER, EmailStore.equal("providerId", merged.getValueInteger("id")), null).stream()
+                  .anyMatch(s -> !sendingDomain.equalsIgnoreCase(s.getValueString("address").substring(s.getValueString("address").lastIndexOf('@') + 1))))
+            {
+               record.addError(new BadInputStatusMessage("Sender addresses must match the sending domain"));
+            }
             resulting.removeIf(r -> Objects.equals(r.getValue("id"), record.getValue("id")));
             resulting.add(merged);
             record.setValue("defaultSlot", Boolean.TRUE.equals(merged.getValueBoolean("enabled")) && Boolean.TRUE.equals(merged.getValueBoolean("isDefault")) ? 1 : null);
@@ -230,6 +246,13 @@ public final class EmailTableCustomizers
          {
             records.forEach(r -> r.addError(new BadInputStatusMessage("Choose another enabled default before deleting this provider")));
          }
+         for(QRecord row : records)
+         {
+            if(!preview && row.getErrors().isEmpty())
+            {
+               SystemWrites.delete(EmailTables.SENDER, EmailStore.equal("providerId", row.getValueInteger("id")), input.getTransaction());
+            }
+         }
          return (records);
       }
 
@@ -272,6 +295,106 @@ public final class EmailTableCustomizers
       }
    }
 
+   /*******************************************************************************
+    ** Sender addresses: canonical lowercase address, the provider's sending domain,
+    ** unique per provider and at most one enabled default per provider.
+    *******************************************************************************/
+   public static class SenderAddressCustomizer implements TableCustomizerInterface
+   {
+      /*******************************************************************************
+       ** Validate each row against its provider and the resulting set of rows.
+       *******************************************************************************/
+      @Override
+      public List<QRecord> preInsertOrUpdate(AbstractActionInput input, List<QRecord> records, boolean preview, Optional<List<QRecord>> oldRecords) throws QException
+      {
+         if(SystemWrites.active())
+         {
+            return (records);
+         }
+         QBackendTransaction transaction = input instanceof InsertInput insert ? insert.getTransaction() : input instanceof UpdateInput update ? update.getTransaction() : null;
+         List<QRecord> resulting = new ArrayList<>();
+         Set<Serializable> providerIds = new LinkedHashSet<>();
+         for(QRecord record : records)
+         {
+            QRecord merged = effective(record, oldRecords);
+            if(record.getValueString("address") != null)
+            {
+               try
+               {
+                  String text = record.getValueString("address");
+                  if(text.contains("\r") || text.contains("\n"))
+                  {
+                     throw (new IllegalArgumentException("Sender address must not contain line breaks"));
+                  }
+                  InternetAddress address = new InternetAddress(text, true);
+                  address.validate();
+                  record.setValue("address", address.getAddress().toLowerCase(Locale.ROOT));
+                  merged.setValue("address", record.getValue("address"));
+               }
+               catch(Exception e)
+               {
+                  record.addError(new BadInputStatusMessage("A valid sender address is required"));
+               }
+            }
+            String name = merged.getValueString("displayName");
+            if(name != null && (name.contains("\r") || name.contains("\n")))
+            {
+               record.addError(new BadInputStatusMessage("Display name must not contain line breaks"));
+            }
+            QRecord provider = merged.getValue("providerId") == null ? null : EmailStore.get(EmailTables.PROVIDER, merged.getValueInteger("providerId"), transaction);
+            if(provider == null)
+            {
+               record.addError(new BadInputStatusMessage("An existing provider is required"));
+            }
+            else if(merged.getValueString("address") != null && record.getErrors().isEmpty())
+            {
+               String domain = provider.getValueString("sendingDomain");
+               String address = merged.getValueString("address");
+               if(domain != null && !domain.isBlank() && !domain.equalsIgnoreCase(address.substring(address.lastIndexOf('@') + 1)))
+               {
+                  record.addError(new BadInputStatusMessage("Sender address domain must match the provider sending domain"));
+               }
+            }
+            if(Boolean.TRUE.equals(merged.getValueBoolean("isDefault")) && !Boolean.TRUE.equals(merged.getValueBoolean("enabled")))
+            {
+               record.addError(new BadInputStatusMessage("A default sender address must be enabled"));
+            }
+            if(merged.getValue("providerId") != null)
+            {
+               providerIds.add(merged.getValue("providerId"));
+            }
+            resulting.add(merged);
+         }
+         if(!providerIds.isEmpty())
+         {
+            List<QRecord> existing = SystemWrites.query(EmailTables.SENDER, new QQueryFilter(new QFilterCriteria("providerId", QCriteriaOperator.IN, new ArrayList<>(providerIds))), transaction);
+            existing.removeIf(r -> records.stream().anyMatch(n -> Objects.equals(n.getValue("id"), r.getValue("id"))));
+            for(int i = 0; i < records.size(); i++)
+            {
+               QRecord merged = resulting.get(i);
+               List<QRecord> peers = new ArrayList<>(existing);
+               for(int j = 0; j < resulting.size(); j++)
+               {
+                  if(j != i)
+                  {
+                     peers.add(resulting.get(j));
+                  }
+               }
+               peers.removeIf(r -> !Objects.equals(r.getValue("providerId"), merged.getValue("providerId")));
+               if(peers.stream().anyMatch(r -> Objects.equals(r.getValueString("address"), merged.getValueString("address"))))
+               {
+                  records.get(i).addError(new BadInputStatusMessage("That sender address already exists for this provider"));
+               }
+               if(Boolean.TRUE.equals(merged.getValueBoolean("isDefault")) && peers.stream().anyMatch(r -> Boolean.TRUE.equals(r.getValueBoolean("isDefault"))))
+               {
+                  records.get(i).addError(new BadInputStatusMessage("Only one sender address per provider can be the default"));
+               }
+            }
+         }
+         return (records);
+      }
+   }
+
    /*******************************************************************************
     ** Canonical addresses and administrator protection for complaint removal.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/email/metadata/EmailTables.java

@@ -26,6 +26,7 @@ import java.util.List;
 import com.kingsrook.qbits.email.EmailQBitConfig;
 import com.kingsrook.qbits.email.outbox.EmailCategory;
 import com.kingsrook.qbits.email.permissions.EmailPermissionChecker;
+import com.kingsrook.qbits.email.support.EmailStore;
 import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
 import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
@@ -53,6 +54,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.tables.UniqueKey;
 public final class EmailTables
 {
    public static final String PROVIDER = "email_provider";
+   public static final String SENDER = "email_provider_sender_address";
    public static final String MESSAGE = "email_message";
    public static final String ATTEMPT = "email_message_attempt";
    public static final String TEMPLATE = "email_template";
@@ -61,7 +63,7 @@ public final class EmailTables
    public static final String SUPPRESSION = "email_suppression";
    public static final String INBOX = "email_webhook_inbox";
    public static final String EVENT = "email_event";
-   public static final List<String> ALL = List.of(PROVIDER, MESSAGE, ATTEMPT, TEMPLATE, VERSION, BRAND, SUPPRESSION, INBOX, EVENT);
+   public static final List<String> ALL = List.of(PROVIDER, SENDER, MESSAGE, ATTEMPT, TEMPLATE, VERSION, BRAND, SUPPRESSION, INBOX, EVENT);
    public static final List<String> MAIN = List.of(MESSAGE, TEMPLATE, BRAND, SUPPRESSION, PROVIDER);
    public static final List<String> SYSTEM_WRITTEN = List.of(MESSAGE, ATTEMPT, INBOX, EVENT);
 
@@ -103,9 +105,10 @@ public final class EmailTables
       ALL.forEach(name -> result.add(QPossibleValueSource.newForTable(name)));
       for(QPossibleValueSource source : result)
       {
-         if(List.of(PROVIDER, MESSAGE, VERSION).contains(source.getName()))
+         if(List.of(PROVIDER, SENDER, MESSAGE, VERSION).contains(source.getName()))
          {
-            source.withSearchFields(PROVIDER.equals(source.getName()) ? List.of("label", "name") : MESSAGE.equals(source.getName()) ? List.of("subject") : List.of("versionNumber", "locale"))
+            source.withSearchFields(PROVIDER.equals(source.getName()) ? List.of("label", "name") : SENDER.equals(source.getName()) ? List.of("address", "displayName")
+               : MESSAGE.equals(source.getName()) ? List.of("subject") : List.of("versionNumber", "locale"))
                .withOrderByField("id");
          }
       }
@@ -119,6 +122,7 @@ public final class EmailTables
    {
       List<QTableMetaData> tables = new ArrayList<>();
       tables.add(provider(config));
+      tables.add(sender(config));
       tables.add(message(config));
       tables.add(messageattempt(config));
       tables.add(template(config));
@@ -210,12 +214,35 @@ public final class EmailTables
       table.withSection(SectionFactory.defaultT1("id", "name", "label", "kind", "enabled", "isDefault"));
       table.withSection(SectionFactory.defaultT2("fromAddress", "fromName", "replyTo", "sendingDomain", "host", "port", "security", "region", "usernameEnvName", "secretEnvName", "webhookSecretEnvName", "snsTopicArn", "configurationSet", "messageStream", "maxPerMinute", "maxPerDay", "trackOpens", "trackClicks", "phiAllowed", "healthStatus", "lastSuccessDate", "lastFailureDate", "lastFailureMessage"));
       table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
+      table.withSection(SectionFactory.customT2("senders", new QIcon("list")).withLabel("Sender Addresses").withWidgetName(SENDER + "_children"));
       table.withSection(SectionFactory.customT2("children", new QIcon("list")).withLabel("Messages").withWidgetName(PROVIDER + "_children"));
       table.withSection(SectionFactory.customT2("inbox", new QIcon("list")).withLabel("Webhook Inbox").withWidgetName(INBOX + "_children"));
       return (table);
    }
 
 
+   /*******************************************************************************
+    ** Metadata for email_provider_sender_address. The provider list in the
+    ** nav is the only entry point; rows are managed as a child list on the provider.
+    *******************************************************************************/
+   private static QTableMetaData sender(EmailQBitConfig config)
+   {
+      QTableMetaData table = base(config, SENDER, "Provider Sender Address")
+         .withField(field("providerId", QFieldType.INTEGER).withPossibleValueSourceName(PROVIDER).withIsRequired(true))
+         .withField(field("address", QFieldType.STRING).withIsRequired(true).withMaxLength(500))
+         .withField(field("displayName", QFieldType.STRING).withMaxLength(500))
+         .withField(field("isDefault", QFieldType.BOOLEAN).withDefaultValue(false))
+         .withField(field("enabled", QFieldType.BOOLEAN).withDefaultValue(true))
+         .withField(field("note", QFieldType.TEXT));
+      table.withRecordLabelFormat(null).withRecordLabelFields(List.of());
+      table.withUniqueKey(new UniqueKey("providerId", "address"));
+      table.withSection(SectionFactory.defaultT1("id", "providerId", "address", "displayName", "isDefault", "enabled"));
+      table.withSection(SectionFactory.defaultT2("note"));
+      table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
+      return (table);
+   }
+
+
    /*******************************************************************************
     ** Metadata for email_message.
     *******************************************************************************/
@@ -227,10 +254,11 @@ public final class EmailTables
          .withField(field("ccAddresses", QFieldType.TEXT).withIsSensitive(true))
          .withField(field("bccAddresses", QFieldType.TEXT).withIsSensitive(true))
          .withField(field("fromAddress", QFieldType.STRING).withIsSensitive(true).withMaxLength(500))
+         .withField(field("fromAddressId", QFieldType.INTEGER).withPossibleValueSourceName(SENDER).withPossibleValueSourceFilter(EmailStore.equal("providerId", "${input.providerId}")))
          .withField(field("replyTo", QFieldType.STRING).withIsSensitive(true).withMaxLength(500))
          .withField(field("subject", QFieldType.TEXT).withIsSensitive(true))
          .withField(field("templateId", QFieldType.INTEGER).withPossibleValueSourceName(TEMPLATE))
-         .withField(field("templateVersionId", QFieldType.INTEGER).withPossibleValueSourceName(VERSION).withPossibleValueSourceFilter(com.kingsrook.qbits.email.support.EmailStore.equal("templateId", "${input.templateId}")))
+         .withField(field("templateVersionId", QFieldType.INTEGER).withPossibleValueSourceName(VERSION).withPossibleValueSourceFilter(EmailStore.equal("templateId", "${input.templateId}")))
          .withField(field("brandId", QFieldType.INTEGER).withPossibleValueSourceName(BRAND))
          .withField(field("category", QFieldType.STRING).withPossibleValueSourceName("email_category").withMaxLength(500))
          .withField(field("locale", QFieldType.STRING).withDefaultValue("en").withMaxLength(500))
@@ -241,6 +269,7 @@ public final class EmailTables
          .withField(field("unsubscribeUrl", QFieldType.TEXT))
          .withField(field("bodySha256", QFieldType.STRING).withMaxLength(500))
          .withField(field("idempotencyKey", QFieldType.STRING).withMaxLength(500))
+         .withField(field("rfcMessageId", QFieldType.STRING).withMaxLength(500))
          .withField(field("providerId", QFieldType.INTEGER).withPossibleValueSourceName(PROVIDER))
          .withField(field("providerMessageId", QFieldType.STRING).withMaxLength(500))
          .withField(field("attemptCount", QFieldType.INTEGER).withDefaultValue(0))
@@ -263,8 +292,9 @@ public final class EmailTables
       table.withRecordLabelFormat(null).withRecordLabelFields(List.of());
       table.getField("subject").setIsHeavy(false);
       table.withUniqueKey(new UniqueKey("idempotencyKey"));
+      table.withUniqueKey(new UniqueKey("rfcMessageId"));
       table.withSection(SectionFactory.defaultT1("id", "status", "toAddresses", "ccAddresses", "bccAddresses", "fromAddress"));
-      table.withSection(SectionFactory.defaultT2("replyTo", "subject", "templateId", "templateVersionId", "brandId", "category", "locale", "variablesSnapshot", "htmlBody", "textBody", "phiApproved", "unsubscribeUrl", "bodySha256", "idempotencyKey", "providerId", "providerMessageId", "attemptCount", "nextAttemptDate", "leaseExpiresDate", "lastError", "finishedDate", "sentDate", "deliveredDate", "bouncedDate", "complainedDate", "deadLetterDate", "linkedTableName", "linkedRecordId", "senderQbit", "senderProcess", "suppressedAddresses", "sendAfter"));
+      table.withSection(SectionFactory.defaultT2("fromAddressId", "replyTo", "subject", "templateId", "templateVersionId", "brandId", "category", "locale", "variablesSnapshot", "htmlBody", "textBody", "phiApproved", "unsubscribeUrl", "bodySha256", "idempotencyKey", "rfcMessageId", "providerId", "providerMessageId", "attemptCount", "nextAttemptDate", "leaseExpiresDate", "lastError", "finishedDate", "sentDate", "deliveredDate", "bouncedDate", "complainedDate", "deadLetterDate", "linkedTableName", "linkedRecordId", "senderQbit", "senderProcess", "suppressedAddresses", "sendAfter"));
       table.withSection(SectionFactory.defaultT3("createDate", "modifyDate"));
       table.withSection(SectionFactory.customT2("children", new QIcon("list")).withLabel("Attempts").withWidgetName(MESSAGE + "_children"));
       table.withSection(SectionFactory.customT2("events", new QIcon("list")).withLabel("Events").withWidgetName(EVENT + "_children"));
@@ -278,7 +308,7 @@ public final class EmailTables
    private static QTableMetaData messageattempt(EmailQBitConfig config)
    {
       QTableMetaData table = base(config, ATTEMPT, "Message Attempt")
-         .withField(field("messageId", QFieldType.INTEGER).withPossibleValueSourceName(MESSAGE).withIsRequired(true))
+         .withField(field("messageId", QFieldType.INTEGER).withPossibleValueSourceName(MESSAGE).withIsRequired(true).withPossibleValueSourceFilter(EmailStore.equal("providerId", "${input.providerId}")))
          .withField(field("attemptNumber", QFieldType.INTEGER).withIsRequired(true))
          .withField(field("outcome", QFieldType.STRING).withPossibleValueSourceName("email_attempt_outcome").withMaxLength(500))
          .withField(field("providerId", QFieldType.INTEGER).withPossibleValueSourceName(PROVIDER))
@@ -307,8 +337,8 @@ public final class EmailTables
          .withField(field("category", QFieldType.STRING).withPossibleValueSourceName("email_category").withMaxLength(500))
          .withField(field("kind", QFieldType.STRING).withPossibleValueSourceName("email_template_kind").withIsRequired(true).withMaxLength(500))
          .withField(field("brandId", QFieldType.INTEGER).withPossibleValueSourceName(BRAND))
-         .withField(field("activeVersionId", QFieldType.INTEGER).withPossibleValueSourceName(VERSION).withIsEditable(false).withPossibleValueSourceFilter(com.kingsrook.qbits.email.support.EmailStore.equal("templateId", "${input.id}")))
-         .withField(field("approvedVersionId", QFieldType.INTEGER).withIsEditable(false).withPossibleValueSourceName(VERSION).withPossibleValueSourceFilter(com.kingsrook.qbits.email.support.EmailStore.equal("templateId", "${input.id}")))
+         .withField(field("activeVersionId", QFieldType.INTEGER).withPossibleValueSourceName(VERSION).withIsEditable(false).withPossibleValueSourceFilter(EmailStore.equal("templateId", "${input.id}")))
+         .withField(field("approvedVersionId", QFieldType.INTEGER).withIsEditable(false).withPossibleValueSourceName(VERSION).withPossibleValueSourceFilter(EmailStore.equal("templateId", "${input.id}")))
          .withField(field("versionLock", QFieldType.INTEGER).withIsHidden(true).withIsEditable(false))
          .withField(field("phiApproved", QFieldType.BOOLEAN).withDefaultValue(false).withIsEditable(false))
          .withField(field("phiApprovedBy", QFieldType.STRING).withMaxLength(500).withIsEditable(false))
@@ -399,7 +429,7 @@ public final class EmailTables
          .withField(field("scope", QFieldType.STRING).withPossibleValueSourceName("email_suppression_scope").withIsRequired(true).withMaxLength(500))
          .withField(field("reason", QFieldType.STRING).withPossibleValueSourceName("email_suppression_reason").withIsRequired(true).withMaxLength(500))
          .withField(field("sourceMessageId", QFieldType.INTEGER))
-         .withField(field("sourceEventId", QFieldType.INTEGER).withPossibleValueSourceName(EVENT))
+         .withField(field("sourceEventId", QFieldType.INTEGER).withPossibleValueSourceName(EVENT).withPossibleValueSourceFilter(EmailStore.equal("messageId", "${input.sourceMessageId}")))
          .withField(field("note", QFieldType.TEXT))
          .withField(field("createdDate", QFieldType.DATE_TIME))
          .withField(field("expiresDate", QFieldType.DATE_TIME));
@@ -442,7 +472,7 @@ public final class EmailTables
    private static QTableMetaData event(EmailQBitConfig config)
    {
       QTableMetaData table = base(config, EVENT, "Event")
-         .withField(field("messageId", QFieldType.INTEGER).withIsRequired(true).withPossibleValueSourceName(MESSAGE))
+         .withField(field("messageId", QFieldType.INTEGER).withIsRequired(true).withPossibleValueSourceName(MESSAGE).withPossibleValueSourceFilter(EmailStore.equal("providerId", "${input.providerId}")))
          .withField(field("providerId", QFieldType.INTEGER).withPossibleValueSourceName(PROVIDER))
          .withField(field("eventType", QFieldType.STRING).withMaxLength(500).withIsRequired(true).withPossibleValueSourceName("email_event_type"))
          .withField(field("bounceType", QFieldType.STRING).withMaxLength(500).withPossibleValueSourceName("email_bounce_type"))
@@ -462,7 +492,7 @@ public final class EmailTables
     *******************************************************************************/
    public static List<QJoinMetaData> joins()
    {
-      return (List.of(join(PROVIDER, MESSAGE, "providerId"), join(MESSAGE, ATTEMPT, "messageId"), join(TEMPLATE, VERSION, "templateId"), join(PROVIDER, INBOX, "providerId"), join(MESSAGE, EVENT, "messageId")));
+      return (List.of(join(PROVIDER, MESSAGE, "providerId"), join(MESSAGE, ATTEMPT, "messageId"), join(TEMPLATE, VERSION, "templateId"), join(PROVIDER, INBOX, "providerId"), join(PROVIDER, SENDER, "providerId"), join(MESSAGE, EVENT, "messageId")));
    }
 
    /*******************************************************************************
@@ -482,8 +512,8 @@ public final class EmailTables
       List<QWidgetMetaData> widgets = new ArrayList<>();
       for(QJoinMetaData join : joins())
       {
-         widgets.add(ChildRecordListRenderer.widgetMetaDataBuilder(join).withName(List.of(INBOX, EVENT).contains(join.getRightTable()) ? join.getRightTable() + "_children" : join.getLeftTable() + "_children")
-            .withLabel(join.getRightTable().equals(INBOX) ? "Webhook Inbox" : join.getRightTable().equals(EVENT) ? "Events" : join.getRightTable().equals(ATTEMPT) ? "Attempts" : join.getRightTable().equals(VERSION) ? "Versions" : "Messages")
+         widgets.add(ChildRecordListRenderer.widgetMetaDataBuilder(join).withName(List.of(INBOX, EVENT, SENDER).contains(join.getRightTable()) ? join.getRightTable() + "_children" : join.getLeftTable() + "_children")
+            .withLabel(join.getRightTable().equals(INBOX) ? "Webhook Inbox" : join.getRightTable().equals(SENDER) ? "Sender Addresses" : join.getRightTable().equals(EVENT) ? "Events" : join.getRightTable().equals(ATTEMPT) ? "Attempts" : join.getRightTable().equals(VERSION) ? "Versions" : "Messages")
             .withMaxRows(50).getWidgetMetaData().withGridColumns(12).withPermissionRules(permissionRules()));
       }
       return (widgets);

src/main/java/com/kingsrook/qbits/email/outbox/EmailEffectOutcome.java

@@ -0,0 +1,33 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.outbox;
+
+
+/*******************************************************************************
+ ** Internal; use EmailApi.
+ ** Durable handoff outcomes. Operational failures propagate to the caller.
+ *******************************************************************************/
+@Deprecated(forRemoval = false)
+public enum EmailEffectOutcome
+{
+   QUEUED_OR_DUPLICATE,
+   REFUSED
+}

src/main/java/com/kingsrook/qbits/email/outbox/EmailEffects.java

@@ -0,0 +1,116 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.outbox;
+
+
+import java.util.List;
+import java.util.Optional;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Internal; use EmailApi.
+ ** Outbox handoff and system-read reconciliation for trusted effect dispatchers.
+ ** This API has no dependency on core durable-job types. Sending retains the
+ ** EmailService permission checks; callers must supply a request without a transaction.
+ *******************************************************************************/
+@Deprecated(forRemoval = false)
+public final class EmailEffects
+{
+   /*******************************************************************************
+    ** A refusal can have a reference when suppression retained an audit row.
+    *******************************************************************************/
+   public record Result(EmailEffectOutcome outcome, EmailMessageRef message)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private EmailEffects()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Hand off once with the effect's identity. Refuse invalid or conflicting keys
+    ** and caller transactions: success must mean the outbox has committed already.
+    ** Database failures throw QException and must not become a permanent refusal.
+    *******************************************************************************/
+   public static Result send(EmailRequest request, String effectIdempotencyKey) throws QException
+   {
+      if(request == null || effectIdempotencyKey == null || effectIdempotencyKey.isBlank() || effectIdempotencyKey.length() > 500
+         || request.getTransaction() != null
+         || request.getIdempotencyKey() != null && !request.getIdempotencyKey().equals(effectIdempotencyKey))
+      {
+         return (new Result(EmailEffectOutcome.REFUSED, null));
+      }
+      request.withIdempotencyKey(effectIdempotencyKey);
+      EmailSendResult result = EmailService.send(request);
+      EmailMessageRef message = result.messageId() == null ? null : find("id", result.messageId()).orElseThrow(() -> new QException("Queued email reference is unavailable"));
+      boolean refused = result.status() == EmailSendResult.Status.REFUSED || result.status() == EmailSendResult.Status.SUPPRESSED
+         || message != null && "SUPPRESSED".equals(message.status());
+      return (new Result(refused ? EmailEffectOutcome.REFUSED : EmailEffectOutcome.QUEUED_OR_DUPLICATE, message));
+   }
+
+
+
+   /*******************************************************************************
+    ** Look up a committed outbox row by its exact effect or application key.
+    *******************************************************************************/
+   public static Optional<EmailMessageRef> findByIdempotencyKey(String key) throws QException
+   {
+      return (key == null || key.isBlank() ? Optional.empty() : find("idempotencyKey", key));
+   }
+
+
+
+   /*******************************************************************************
+    ** Look up the exact bracketed RFC Message-ID stored at enqueue time.
+    *******************************************************************************/
+   public static Optional<EmailMessageRef> findByRfcMessageId(String messageId) throws QException
+   {
+      return (messageId == null || messageId.isBlank() ? Optional.empty() : find("rfcMessageId", messageId));
+   }
+
+
+
+   /*******************************************************************************
+    ** Use the same authorized read scope as the dispatcher; expose only references.
+    *******************************************************************************/
+   private static Optional<EmailMessageRef> find(String field, java.io.Serializable value) throws QException
+   {
+      List<QRecord> rows = SystemWrites.query(EmailTables.MESSAGE, EmailStore.equal(field, value).withLimit(1));
+      if(rows.isEmpty())
+      {
+         return (Optional.empty());
+      }
+      QRecord row = rows.getFirst();
+      return (Optional.of(new EmailMessageRef(row.getValueInteger("id"), row.getValueString("status"), row.getValueString("rfcMessageId"), row.getValueString("providerMessageId"))));
+   }
+}

src/main/java/com/kingsrook/qbits/email/outbox/EmailMessageRef.java

@@ -0,0 +1,31 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.outbox;
+
+
+/*******************************************************************************
+ ** Internal; use EmailApi.
+ ** Reconciliation reference without message bodies, recipients or credentials.
+ *******************************************************************************/
+@Deprecated(forRemoval = false)
+public record EmailMessageRef(Integer id, String status, String rfcMessageId, String providerMessageId)
+{
+}

src/main/java/com/kingsrook/qbits/email/outbox/EmailRequest.java

@@ -39,6 +39,7 @@ public class EmailRequest
    private List<String> cc = new ArrayList<>();
    private List<String> bcc = new ArrayList<>();
    private String from;
+   private Integer fromAddressId;
    private String replyTo;
    private String subject;
    private String markdownBody;
@@ -213,6 +214,26 @@ public class EmailRequest
    }
 
 
+   /*******************************************************************************
+    ** Getter for fromAddressId.
+    *******************************************************************************/
+   public Integer getFromAddressId()
+   {
+      return (fromAddressId);
+   }
+
+
+   /*******************************************************************************
+    ** Fluent setter for fromAddressId, a row of email_provider_sender_address that
+    ** must belong to the resolved provider and be enabled.
+    *******************************************************************************/
+   public EmailRequest withFromAddressId(Integer fromAddressId)
+   {
+      this.fromAddressId = fromAddressId;
+      return (this);
+   }
+
+
    /*******************************************************************************
     ** Getter for replyTo.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/email/outbox/EmailService.java

@@ -30,6 +30,7 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import java.util.UUID;
 import com.kingsrook.qbits.email.EmailQBitConfig;
 import com.kingsrook.qbits.email.EmailRuntime;
 import com.kingsrook.qbits.email.metadata.EmailTableCustomizers;
@@ -164,16 +165,29 @@ public final class EmailService
       List<String> bcc;
       String from;
       String replyTo;
+      QRecord senderRow;
+      try
+      {
+         senderRow = senderRow(provider, SystemWrites.query(EmailTables.SENDER, EmailStore.equal("providerId", provider.getValue("id")), queryTransaction), request);
+      }
+      catch(IllegalArgumentException e)
+      {
+         return (refused(e.getMessage()));
+      }
       try
       {
          to = addresses(request.getTo());
          cc = addresses(request.getCc());
          bcc = addresses(request.getBcc());
-         String sender = request.getFrom() == null ? provider.getValueString("fromAddress") : request.getFrom();
+         String sender = request.getFrom() != null ? request.getFrom() : senderRow != null ? senderRow.getValueString("address") : provider.getValueString("fromAddress");
          from = single(sender);
-         if(request.getFrom() == null && provider.getValueString("fromName") != null)
+         if(request.getFrom() == null)
          {
-            from = new InternetAddress(new InternetAddress(from).getAddress(), provider.getValueString("fromName"), "UTF-8").toString();
+            String name = senderRow != null ? senderRow.getValueString("displayName") : provider.getValueString("fromName");
+            if(name != null && !name.isBlank())
+            {
+               from = new InternetAddress(new InternetAddress(from).getAddress(), name, "UTF-8").toString();
+            }
          }
          String reply = request.getReplyTo() == null ? provider.getValueString("replyTo") : request.getReplyTo();
          replyTo = reply == null ? null : single(reply);
@@ -244,10 +258,11 @@ public final class EmailService
       QRecord row = new QRecord().withValue("providerId", provider.getValue("id"))
          .withValue("status", allSuppressed ? "SUPPRESSED" : "QUEUED").withValue("finishedDate", allSuppressed ? now : null).withValue("attemptCount", 0)
          .withValue("toAddresses", String.join(",", to)).withValue("ccAddresses", String.join(",", cc)).withValue("bccAddresses", String.join(",", bcc))
-         .withValue("fromAddress", from).withValue("replyTo", replyTo).withValue("subject", request.getSubject())
+         .withValue("fromAddress", from).withValue("fromAddressId", senderRow == null ? null : senderRow.getValue("id")).withValue("replyTo", replyTo).withValue("subject", request.getSubject())
          .withValue("htmlBody", rendered.html()).withValue("textBody", rendered.text()).withValue("bodySha256", digest(rendered))
          .withValue("brandId", adHocBrand == null ? null : adHocBrand.getValue("id"))
          .withValue("category", category.name()).withValue("locale", request.getLocale()).withValue("idempotencyKey", request.getIdempotencyKey())
+         .withValue("rfcMessageId", messageId(request.getIdempotencyKey(), from))
          .withValue("linkedTableName", request.getLinkedTableName()).withValue("linkedRecordId", request.getLinkedRecordId())
          .withValue("senderQbit", request.getSenderQbit()).withValue("senderProcess", request.getSenderProcess())
          .withValue("sendAfter", request.getSendAfter()).withValue("nextAttemptDate", request.getSendAfter() == null ? now : request.getSendAfter())
@@ -304,7 +319,7 @@ public final class EmailService
          return (refused("Provider is unavailable"));
       }
       EmailRequest request = new EmailRequest().withSubject(row.getValueString("subject")).withHtmlBody(row.getValueString("htmlBody"))
-         .withTextBody(row.getValueString("textBody")).withFrom(row.getValueString("fromAddress")).withReplyTo(row.getValueString("replyTo"))
+         .withTextBody(row.getValueString("textBody")).withFrom(row.getValueString("fromAddress")).withFromAddressId(row.getValueInteger("fromAddressId")).withReplyTo(row.getValueString("replyTo"))
          .withProviderName(provider.getValueString("name")).withCategory(EmailCategory.valueOf(row.getValueString("category")))
          .withLinkedRecord(EmailTables.MESSAGE, id).withSender("email", "email_resend");
       if(row.getValueString("toAddresses") != null && !row.getValueString("toAddresses").isBlank())
@@ -378,18 +393,10 @@ public final class EmailService
       {
          return (new EmailSendResult(existing.getValueInteger("id"), EmailSendResult.Status.DUPLICATE, List.of()));
       }
-      QRecord inserted;
-      try
-      {
-         inserted = SystemWrites.insert(EmailTables.MESSAGE, List.of(row), transaction).getRecords().get(0);
-      }
-      catch(QException e)
-      {
-         return (refused("Could not queue the email message"));
-      }
+      QRecord inserted = SystemWrites.insert(EmailTables.MESSAGE, List.of(row), transaction).getRecords().get(0);
       if(inserted.getErrors() != null && !inserted.getErrors().isEmpty())
       {
-         return (refused("Could not queue the email message"));
+         throw (new QException("Could not queue the email message"));
       }
       try
       {
@@ -479,6 +486,57 @@ public final class EmailService
       return (result);
    }
 
+   /*******************************************************************************
+    ** The provider sender-address row a request uses, or null for the provider's own
+    ** from address. A chosen id, or a from text, must name an enabled row of this
+    ** provider; once a provider has rows, an explicit from text must match one.
+    ** With neither, an enabled default row wins over the provider's from address.
+    *******************************************************************************/
+   private static QRecord senderRow(QRecord provider, List<QRecord> rows, EmailRequest request)
+   {
+      String refusal = "That from address is not set up for provider " + provider.getValueString("name");
+      String wanted = null;
+      if(request.getFrom() != null && (request.getFromAddressId() != null || !rows.isEmpty()))
+      {
+         try
+         {
+            wanted = new InternetAddress(request.getFrom()).getAddress().toLowerCase(Locale.ROOT);
+         }
+         catch(Exception e)
+         {
+            throw (new IllegalArgumentException("One or more email addresses are invalid"));
+         }
+      }
+      for(QRecord row : rows)
+      {
+         boolean enabled = Boolean.TRUE.equals(row.getValueBoolean("enabled"));
+         if(request.getFromAddressId() != null)
+         {
+            if(request.getFromAddressId().equals(row.getValueInteger("id")))
+            {
+               if(!enabled || wanted != null && !wanted.equals(row.getValueString("address")))
+               {
+                  throw (new IllegalArgumentException(refusal));
+               }
+               return (row);
+            }
+         }
+         else if(wanted != null && enabled && wanted.equals(row.getValueString("address")))
+         {
+            return (row);
+         }
+         else if(request.getFrom() == null && enabled && Boolean.TRUE.equals(row.getValueBoolean("isDefault")))
+         {
+            return (row);
+         }
+      }
+      if(request.getFromAddressId() != null || wanted != null)
+      {
+         throw (new IllegalArgumentException(refusal));
+      }
+      return (null);
+   }
+
    /*******************************************************************************
     ** Strict address parsing, rejecting header injection and local-only addresses.
     *******************************************************************************/
@@ -532,6 +590,42 @@ public final class EmailService
       }
    }
 
+
+
+   /*******************************************************************************
+    ** Create once before insertion. A UUID seeds messages without an idempotency key.
+    ** The mailbox parser ignores display names; malformed domains use a local fallback.
+    *******************************************************************************/
+   static String messageId(String key, String from)
+   {
+      String domain = "qbit-email.local";
+      try
+      {
+         String address = new InternetAddress(from).getAddress();
+         String candidate = address.substring(address.lastIndexOf('@') + 1);
+         if(address.contains("@") && candidate.matches("[A-Za-z0-9]+(?:[.-][A-Za-z0-9]+)*"))
+         {
+            domain = candidate.toLowerCase(Locale.ROOT);
+         }
+      }
+      catch(Exception e)
+      {
+         // A fallback also supports callers preparing legacy message rows.
+      }
+      try
+      {
+         String seed = key == null ? "qbit-email:" + UUID.randomUUID() : key;
+         String token = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(seed.getBytes(StandardCharsets.UTF_8))).substring(0, 32);
+         return ("<" + token + "@" + domain + ">");
+      }
+      catch(java.security.NoSuchAlgorithmException e)
+      {
+         throw (new IllegalStateException("SHA-256 unavailable"));
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Refusal with no stored row.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/email/permissions/EmailPermissionChecker.java

@@ -74,7 +74,7 @@ public class EmailPermissionChecker implements CustomPermissionChecker
       boolean write = actionInput instanceof InsertInput || actionInput instanceof UpdateInput || actionInput instanceof DeleteInput;
       if(metaData instanceof QTableMetaData table)
       {
-         if(EmailTables.PROVIDER.equals(table.getName()))
+         if(EmailTables.PROVIDER.equals(table.getName()) || EmailTables.SENDER.equals(table.getName()))
          {
             return (EmailPermissions.ADMIN);
          }

src/main/java/com/kingsrook/qbits/email/processes/EmailProcesses.java

@@ -90,6 +90,9 @@ public final class EmailProcesses
    public static final String PROCESS_WEBHOOKS = "email_process_webhooks";
    public static final String RETENTION = "email_retention";
 
+   /** Process value the send-test prepare step sets to the selected provider id; the from address filter falls back to it. */
+   public static final String PROVIDER_FILTER_VALUE = "possibleValueFilterValueId";
+
    /*******************************************************************************
     ** Utility class.
     *******************************************************************************/
@@ -116,9 +119,12 @@ public final class EmailProcesses
    public static List<QProcessMetaData> processes(EmailQBitConfig config)
    {
       QProcessMetaData test = action(SEND_TEST, "Send Test Email", EmailTables.PROVIDER);
-      test.withStep(0, new QFrontendStepMetaData().withName("input").withLabel("Recipient")
+      test.withStep(0, new QBackendStepMetaData().withName("prepare").withCode(new QCodeReference(SendTestPrepareStep.class)));
+      test.withStep(1, new QFrontendStepMetaData().withName("input").withLabel("Recipient")
          .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM))
-         .withFormField(new QFieldMetaData("toAddress", QFieldType.STRING).withLabel("To address").withIsRequired(true)));
+         .withFormField(new QFieldMetaData("toAddress", QFieldType.STRING).withLabel("To address").withIsRequired(true))
+         .withFormField(new QFieldMetaData("fromAddressId", QFieldType.INTEGER).withLabel("From address").withPossibleValueSourceName(EmailTables.SENDER)
+            .withPossibleValueSourceFilter(EmailStore.equal("providerId", "${input.id}??${processValues." + PROVIDER_FILTER_VALUE + "}"))));
       QFunctionInputMetaData inputs = new QFunctionInputMetaData();
       QFunctionInputMetaData adHocInputs = new QFunctionInputMetaData();
       for(String field : List.of("templateName", "to", "cc", "bcc", "replyTo", "subject", "markdownBody", "htmlBody", "textBody", "providerName", "brandName",
@@ -128,6 +134,12 @@ public final class EmailProcesses
          adHocInputs.withField(new QFieldMetaData(field, QFieldType.STRING));
       }
       adHocInputs.withField(new QFieldMetaData("category", QFieldType.STRING));
+      for(QFunctionInputMetaData target : List.of(inputs, adHocInputs))
+      {
+         target.withField(new QFieldMetaData("providerId", QFieldType.INTEGER).withPossibleValueSourceName(EmailTables.PROVIDER));
+         target.withField(new QFieldMetaData("fromAddressId", QFieldType.INTEGER).withPossibleValueSourceName(EmailTables.SENDER)
+            .withPossibleValueSourceFilter(EmailStore.equal("providerId", "${input.providerId}")));
+      }
       QProcessMetaData adHoc = new QProcessMetaData().withName(SEND_AD_HOC).withLabel("Send Ad Hoc Email").withIsHidden(true)
          .withPermissionRules(EmailTables.permissionRules())
          .withStep(new QBackendStepMetaData().withName("send").withCode(new QCodeReference(SendStep.class)).withInputData(adHocInputs));
@@ -294,7 +306,7 @@ public final class EmailProcesses
                   case SEND_NOW -> new EmailDispatcher().sendNow(id).name();
                   case CANCEL -> cancel(id) ? "CANCELLED" : "Skipped";
                   case RESEND -> resend(current).status().name();
-                  case SEND_TEST -> sendTest(current, input.getValueString("toAddress"));
+                  case SEND_TEST -> sendTest(current, input.getValueString("toAddress"), input.getValueInteger("fromAddressId"));
                   default -> "Skipped";
                };
             }
@@ -348,17 +360,45 @@ public final class EmailProcesses
    /*******************************************************************************
     ** Queue and dispatch a fixed provider test message.
     *******************************************************************************/
-   private static String sendTest(QRecord provider, String to) throws QException
+   private static String sendTest(QRecord provider, String to, Integer fromAddressId) throws QException
    {
       if(to == null)
       {
          return ("REFUSED: a to address is required");
       }
-      EmailSendResult result = EmailService.send(new EmailRequest().withProviderName(provider.getValueString("name")).withTo(to)
+      EmailSendResult result = EmailService.send(new EmailRequest().withProviderName(provider.getValueString("name")).withTo(to).withFromAddressId(fromAddressId)
          .withSubject("Email delivery test").withMarkdownBody("This is a test of your email provider.").withSender("email", SEND_TEST));
       return (result.status() == EmailSendResult.Status.QUEUED ? new EmailDispatcher().sendNow(result.messageId()).name() : result.status().name());
    }
 
+   /*******************************************************************************
+    ** Seeds the from address possible-value filter: a record action's form has no
+    ** record id, so a single selected provider id becomes a process value that the
+    ** filter reads after the form's own ${input.id}.
+    *******************************************************************************/
+   public static class SendTestPrepareStep implements BackendStep
+   {
+      /*******************************************************************************
+       ** Expose the selected provider id when exactly one provider is selected.
+       *******************************************************************************/
+      @Override
+      public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         String ids = input.getValueString("recordIds");
+         if(ids != null && !ids.contains(","))
+         {
+            try
+            {
+               output.addValue(PROVIDER_FILTER_VALUE, Integer.valueOf(ids.trim()));
+            }
+            catch(NumberFormatException e)
+            {
+               // Not a single id: the filter stays open and the review step still validates the selection.
+            }
+         }
+      }
+   }
+
    /*******************************************************************************
     ** Backend sending API with inputs declared as QQQ fields.
     *******************************************************************************/
@@ -370,10 +410,20 @@ public final class EmailProcesses
       @Override
       public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
       {
-         EmailRequest request = new EmailRequest().withTemplateName(input.getValueString("templateName"))
+         String providerName = input.getValueString("providerName");
+         if(input.getValueInteger("providerId") != null)
+         {
+            QRecord provider = EmailStore.get(EmailTables.PROVIDER, input.getValueInteger("providerId"), null);
+            if(provider == null || providerName != null && !providerName.equals(provider.getValueString("name")))
+            {
+               throw (new QException("providerId does not match an existing provider named providerName"));
+            }
+            providerName = provider.getValueString("name");
+         }
+         EmailRequest request = new EmailRequest().withTemplateName(input.getValueString("templateName")).withFromAddressId(input.getValueInteger("fromAddressId"))
             .withSubject(input.getValueString("subject")).withMarkdownBody(input.getValueString("markdownBody"))
             .withHtmlBody(input.getValueString("htmlBody")).withTextBody(input.getValueString("textBody"))
-            .withReplyTo(input.getValueString("replyTo")).withProviderName(input.getValueString("providerName"))
+            .withReplyTo(input.getValueString("replyTo")).withProviderName(providerName)
             .withBrandName(input.getValueString("brandName")).withIdempotencyKey(input.getValueString("idempotencyKey"))
             .withLinkedRecord(input.getValueString("linkedTableName"), input.getValueString("linkedRecordId"));
          if(input.getValueString("to") != null)

src/main/java/com/kingsrook/qbits/email/support/EmailApiMapper.java

@@ -0,0 +1,117 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.support;
+
+
+import com.kingsrook.qbits.email.api.EmailApiRequest;
+import com.kingsrook.qbits.email.api.EmailApiResult;
+import com.kingsrook.qbits.email.api.EmailApiStatus;
+import com.kingsrook.qbits.email.api.EmailApiTemplate;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.template.EmailTemplateDefinition;
+
+
+/*******************************************************************************
+ ** Internal adapter between the stable contract and the implementation.
+ *******************************************************************************/
+public final class EmailApiMapper
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private EmailApiMapper()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Map every published request field without bypassing send validation.
+    *******************************************************************************/
+   public static EmailRequest request(EmailApiRequest request)
+   {
+      if(request == null)
+      {
+         return (null);
+      }
+      EmailRequest mapped = new EmailRequest()
+         .withTo(request.getTo())
+         .withCc(request.getCc())
+         .withBcc(request.getBcc())
+         .withFrom(request.getFrom())
+         .withFromAddressId(request.getFromAddressId())
+         .withReplyTo(request.getReplyTo())
+         .withSubject(request.getSubject())
+         .withMarkdownBody(request.getMarkdownBody())
+         .withHtmlBody(request.getHtmlBody())
+         .withTextBody(request.getTextBody())
+         .withProviderName(request.getProviderName())
+         .withTemplateName(request.getTemplateName())
+         .withBrandName(request.getBrandName())
+         .withLocale(request.getLocale())
+         .withIdempotencyKey(request.getIdempotencyKey())
+         .withTransaction(request.getTransaction(), request.getTransactionBackendName())
+         .withSendAfter(request.getSendAfter())
+         .withLinkedTableName(request.getLinkedTableName())
+         .withLinkedRecordId(request.getLinkedRecordId())
+         .withSenderQbit(request.getSenderQbit())
+         .withSenderProcess(request.getSenderProcess())
+         .withVariables(request.getVariables())
+         .withCategory(request.getCategory() == null ? null : EmailCategory.valueOf(request.getCategory().name()));
+      request.getRecordVariables().forEach((name, reference) -> mapped.withRecordVariable(name, reference.table(), reference.id()));
+      return (mapped);
+   }
+
+
+
+   /*******************************************************************************
+    ** Map the complete create-once template contract.
+    *******************************************************************************/
+   public static EmailTemplateDefinition template(EmailApiTemplate template)
+   {
+      EmailTemplateDefinition mapped = new EmailTemplateDefinition()
+         .withName(template.getName())
+         .withLabel(template.getLabel())
+         .withKind(template.getKind())
+         .withSubject(template.getSubject())
+         .withBody(template.getBody())
+         .withTextBody(template.getTextBody())
+         .withVariablesSchema(template.getVariablesSchema())
+         .withSampleVariables(template.getSampleVariables())
+         .withOwnerQbit(template.getOwnerQbit())
+         .withEditPermissionName(template.getEditPermissionName())
+         .withCategory(template.getCategory() == null ? null : EmailCategory.valueOf(template.getCategory().name()));
+      template.getVariables().forEach(variable -> mapped.withVariable(variable.name(), variable.label(), variable.type(), variable.required(), variable.classification()));
+      return (mapped);
+   }
+
+
+
+   /*******************************************************************************
+    ** Map enqueue outcomes by their stable names.
+    *******************************************************************************/
+   public static EmailApiResult result(EmailSendResult result)
+   {
+      return (new EmailApiResult(result.messageId(), EmailApiStatus.valueOf(result.status().name()), result.reasons()));
+   }
+}

src/main/java/com/kingsrook/qbits/email/transport/EmailEnvelope.java

@@ -31,6 +31,7 @@ import java.util.Map;
 public class EmailEnvelope
 {
    private String messageId;
+   private String rfcMessageId;
    private String from;
    private List<String> to = List.of();
    private List<String> cc = List.of();
@@ -62,6 +63,27 @@ public class EmailEnvelope
    }
 
 
+
+   /*******************************************************************************
+    ** Getter for the persisted RFC 5322 Message-ID, distinct from the outbox row id.
+    *******************************************************************************/
+   public String getRfcMessageId()
+   {
+      return (rfcMessageId);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for the persisted RFC 5322 Message-ID.
+    *******************************************************************************/
+   public EmailEnvelope withRfcMessageId(String rfcMessageId)
+   {
+      this.rfcMessageId = rfcMessageId;
+      return (this);
+   }
+
+
    /*******************************************************************************
     ** Getter for from.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/email/transport/EmailHttpSupport.java

@@ -81,13 +81,17 @@ final class EmailHttpSupport
       Map<String, String> headers = new LinkedHashMap<>();
       envelope.getHeaders().forEach((name, value) ->
       {
-         if(!name.equalsIgnoreCase("Bcc") && !name.equalsIgnoreCase("Auto-Submitted") && !name.equalsIgnoreCase("X-QQQ-Email-Id"))
+         if(!name.equalsIgnoreCase("Bcc") && !name.equalsIgnoreCase("Auto-Submitted") && !name.equalsIgnoreCase("X-QQQ-Email-Id") && !name.equalsIgnoreCase("Message-ID"))
          {
             headers.put(name, value);
          }
       });
       headers.put("Auto-Submitted", "auto-generated");
       headers.put("X-QQQ-Email-Id", envelope.getMessageId());
+      if(envelope.getRfcMessageId() != null)
+      {
+         headers.put("Message-ID", envelope.getRfcMessageId());
+      }
       return (headers);
    }
 

src/main/java/com/kingsrook/qbits/email/transport/EmailMime.java

@@ -53,7 +53,26 @@ final class EmailMime
       noLineBreaks(envelope.getFrom());
       noLineBreaks(envelope.getReplyTo());
       noLineBreaks(envelope.getMessageId());
-      MimeMessage message = new MimeMessage(session);
+      noLineBreaks(envelope.getRfcMessageId());
+      String rfcMessageId = envelope.getRfcMessageId();
+      MimeMessage message = new MimeMessage(session)
+      {
+         /*******************************************************************************
+          ** JavaMail may save again when sending; keep the persisted identity each time.
+          *******************************************************************************/
+         @Override
+         protected void updateMessageID() throws MessagingException
+         {
+            if(rfcMessageId == null)
+            {
+               super.updateMessageID();
+            }
+            else
+            {
+               setHeader("Message-ID", rfcMessageId);
+            }
+         }
+      };
       message.setFrom(new InternetAddress(envelope.getFrom(), true));
       recipients(message, Message.RecipientType.TO, envelope.getTo());
       recipients(message, Message.RecipientType.CC, envelope.getCc());

src/main/java/com/kingsrook/qbits/email/transport/MailgunEmailTransport.java

@@ -96,6 +96,10 @@ public class MailgunEmailTransport implements EmailTransport
          field(body, boundary, "o:tracking-opens", settings.enabled("trackOpens") ? "yes" : "no");
          field(body, boundary, "o:tracking-clicks", settings.enabled("trackClicks") ? "yes" : "no");
          field(body, boundary, "v:qqq_email_id", envelope.getMessageId());
+         if(envelope.getRfcMessageId() != null)
+         {
+            field(body, boundary, "h:Message-Id", envelope.getRfcMessageId());
+         }
          body.writeBytes(("--" + boundary + "\r\nContent-Disposition: form-data; name=\"message\"; filename=\"message.eml\"\r\n"
             + "Content-Type: message/rfc822\r\n\r\n").getBytes(StandardCharsets.UTF_8));
          EmailMime.build(Session.getInstance(new Properties()), envelope).writeTo(body, new String[] {"Bcc"});

src/main/java/com/kingsrook/qbits/email/transport/SesEmailTransport.java

@@ -50,6 +50,9 @@ import software.amazon.awssdk.services.sesv2.model.SesV2Exception;
 
 /*******************************************************************************
  ** SES v2 raw MIME transport; credentials are resolved only for an attempt.
+ ** The raw request carries the stored Message-ID, but SES replaces it downstream.
+ ** Receiver deduplication by that header is unavailable with SES; X-QQQ-Email-Id
+ ** remains the application correlation header. See docs/DURABLE-JOBS.md.
  *******************************************************************************/
 public class SesEmailTransport implements EmailTransport
 {

src/test/java/com/kingsrook/qbits/email/EmailDeliveryTest.java

@@ -21,18 +21,34 @@
 package com.kingsrook.qbits.email;
 
 import java.net.ServerSocket;
-import java.time.*;
-import java.util.*;
-import java.util.concurrent.*;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import com.icegreen.greenmail.util.GreenMail;
 import com.icegreen.greenmail.util.ServerSetup;
-import com.kingsrook.qbits.email.delivery.*;
+import com.kingsrook.qbits.email.delivery.EmailDispatcher;
 import com.kingsrook.qbits.email.metadata.EmailTables;
-import com.kingsrook.qbits.email.outbox.*;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.outbox.EmailService;
 import com.kingsrook.qbits.email.processes.EmailProcesses;
-import com.kingsrook.qbits.email.support.*;
-import com.kingsrook.qbits.email.transport.*;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qbits.email.transport.EmailTransport;
+import com.kingsrook.qbits.email.transport.EmailTransportException;
+import com.kingsrook.qbits.email.transport.EmailTransportRegistry;
+import com.kingsrook.qbits.email.transport.EmailTransportResult;
+import com.kingsrook.qbits.email.transport.SmtpEmailTransport;
 import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
 import com.kingsrook.qqq.backend.core.actions.messaging.SendMessageAction;
 import com.kingsrook.qqq.backend.core.alerts.AlertResolution;
@@ -41,16 +57,23 @@ import com.kingsrook.qqq.backend.core.alerts.AlertSinkInterface;
 import com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
-import com.kingsrook.qqq.backend.core.model.actions.messaging.*;
-import com.kingsrook.qqq.backend.core.model.actions.messaging.email.*;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Content;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.MultiParty;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Party;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.SendMessageInput;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.email.EmailContentRole;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.email.EmailPartyRole;
 import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import jakarta.mail.Multipart;
-import org.junit.jupiter.api.*;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
-import static org.assertj.core.api.Assertions.*;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /*******************************************************************************
  ** Email Delivery Test.
@@ -398,6 +421,39 @@ class EmailDeliveryTest
       assertThat(H2Support.all(EmailTables.ATTEMPT)).hasSize(1);
    }
 
+   /*******************************************************************************
+    ** A lost acknowledgement retries the stored identity through the real SMTP sink.
+    *******************************************************************************/
+   @Test
+   void smtpRetryPreservesStoredMessageId() throws Exception
+   {
+      var queued = EmailService.send(request().withIdempotencyKey("job:1:row:mail"));
+      String stored = message(queued.messageId()).getValueString("rfcMessageId");
+      AtomicInteger calls = new AtomicInteger();
+      var dispatcher = fake((envelope, settings) ->
+      {
+         assertThat(envelope.getRfcMessageId()).isEqualTo(stored);
+         var sent = new SmtpEmailTransport().send(envelope, settings);
+         if(calls.incrementAndGet() == 1)
+         {
+            throw new EmailTransportException("Lost acknowledgement", false);
+         }
+         return (sent);
+      });
+      assertThat(dispatcher.sendNow(queued.messageId())).isEqualTo(EmailDispatcher.SendResult.RETRY_WAIT);
+      EmailRuntime.setClock(Clock.fixed(now.plusSeconds(301), ZoneOffset.UTC));
+      assertThat(dispatcher.sendNow(queued.messageId())).isEqualTo(EmailDispatcher.SendResult.SENT);
+      assertThat(mail.waitForIncomingEmail(2000, 2)).isTrue();
+      assertThat(mail.getReceivedMessages()).hasSize(2);
+      for(var received : mail.getReceivedMessages())
+      {
+         assertThat(received.getMessageID()).isEqualTo(stored);
+         assertThat(received.getHeader("X-QQQ-Email-Id")).containsExactly(queued.messageId().toString());
+      }
+      assertThat(message(queued.messageId()).getValueString("rfcMessageId")).isEqualTo(stored);
+      assertThat(H2Support.all(EmailTables.ATTEMPT)).hasSize(2);
+   }
+
    /*******************************************************************************
     ** Smtp Authentication And Secret Resolution Only At Send.
     *******************************************************************************/
@@ -477,8 +533,8 @@ class EmailDeliveryTest
       var second = EmailService.send(request());
       fake((e, s) ->
       {
-               throw new EmailTransportException("throttle", false).withRetryAfter(Duration.ofHours(2));
-            })
+         throw new EmailTransportException("throttle", false).withRetryAfter(Duration.ofHours(2));
+      })
             .runOnce();
       assertThat(message(second.messageId()).getValueInstant("nextAttemptDate"))
             .isEqualTo(now.plusSeconds(3960));
@@ -494,8 +550,8 @@ class EmailDeliveryTest
       assertThat(
                   fake((e, s) ->
                   {
-                           throw new EmailTransportException("rejected", true);
-                        })
+                     throw new EmailTransportException("rejected", true);
+                  })
                         .sendNow(queued.messageId()))
             .isEqualTo(EmailDispatcher.SendResult.DEAD_LETTER);
       var future = EmailService.send(request());

src/test/java/com/kingsrook/qbits/email/EmailHomePolishTest.java

@@ -30,9 +30,9 @@ import java.util.Set;
 import com.kingsrook.qbits.email.app.EmailHome;
 import com.kingsrook.qbits.email.metadata.EmailEnums;
 import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
 import com.kingsrook.qbits.email.render.ContrastCheck;
 import com.kingsrook.qbits.email.render.EmailLayout;
-import com.kingsrook.qbits.email.outbox.EmailCategory;
 import com.kingsrook.qbits.email.support.SystemWrites;
 import com.kingsrook.qqq.backend.core.actions.dashboard.RenderWidgetAction;
 import com.kingsrook.qqq.backend.core.actions.tables.GetAction;
@@ -66,7 +66,9 @@ class EmailHomePolishTest
    @BeforeEach
    void setup() throws Exception
    {
-      H2Support.setup(c -> { });
+      H2Support.setup(c ->
+      {
+      });
       EmailRuntime.setClock(Clock.fixed(NOW, ZoneOffset.UTC));
       H2Support.provider(25);
    }

src/test/java/com/kingsrook/qbits/email/EmailMetadataTest.java

@@ -20,14 +20,24 @@
 
 package com.kingsrook.qbits.email;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
 import com.kingsrook.qbits.email.app.EmailApp;
-import com.kingsrook.qbits.email.metadata.*;
-import com.kingsrook.qbits.email.outbox.*;
-import com.kingsrook.qbits.email.permissions.*;
-import com.kingsrook.qbits.email.processes.*;
-import com.kingsrook.qbits.email.support.*;
-import com.kingsrook.qqq.backend.core.actions.tables.*;
+import com.kingsrook.qbits.email.metadata.EmailTableCustomizers;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.AdHocEmailRenderer;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.outbox.EmailService;
+import com.kingsrook.qbits.email.permissions.EmailPermissionChecker;
+import com.kingsrook.qbits.email.processes.EmailProcesses;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
@@ -36,10 +46,14 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
-import com.kingsrook.qqq.backend.core.model.metadata.fields.*;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.Capability;
-import org.junit.jupiter.api.*;
-import static org.assertj.core.api.Assertions.*;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /*******************************************************************************
  ** Email Metadata Test.
@@ -73,7 +87,7 @@ class EmailMetadataTest
    void producerStandardsAndIntegrationFields() throws Exception
    {
       var instance = QContext.getQInstance();
-      assertThat(instance.getTables()).hasSize(9);
+      assertThat(instance.getTables()).hasSize(10);
       for(var table : instance.getTables().values())
       {
          assertThat(table.getName()).matches("email_[a-z_]+");
@@ -412,7 +426,7 @@ class EmailMetadataTest
                   "MAILGUN",
                   "GMAIL_API",
                   "future"))
-                  {
+      {
          assertThat(
                      new EmailQBitConfig()
                            .withAllowedEnvNames(Set.of("CUSTOM"))

src/test/java/com/kingsrook/qbits/email/EmailProcessesTest.java

@@ -20,20 +20,28 @@
 
 package com.kingsrook.qbits.email;
 
-import java.time.*;
-import java.util.*;
-import com.icegreen.greenmail.util.*;
-import com.kingsrook.qbits.email.delivery.*;
-import com.kingsrook.qbits.email.metadata.*;
-import com.kingsrook.qbits.email.outbox.*;
-import com.kingsrook.qbits.email.processes.*;
-import com.kingsrook.qbits.email.support.*;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.Map;
+import com.icegreen.greenmail.util.GreenMail;
+import com.icegreen.greenmail.util.ServerSetup;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailService;
+import com.kingsrook.qbits.email.processes.EmailProcesses;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
 import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
-import com.kingsrook.qqq.backend.core.model.actions.processes.*;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
-import org.junit.jupiter.api.*;
-import static org.assertj.core.api.Assertions.*;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /*******************************************************************************
  ** Email Processes Test.

src/test/java/com/kingsrook/qbits/email/EmailReviewFixesTest.java

@@ -109,7 +109,7 @@ class EmailReviewFixesTest
       var schema = new VariableSchema("[{\"name\":\"items\",\"type\":\"LIST\"},{\"name\":\"value\",\"type\":\"STRING\"}]");
       var loop = engine.parse("{{#each items}}{{#each items}}{{/each}}{{/each}}", schema);
       assertThatThrownBy(() -> loop.render(Map.of("items", Collections.nCopies(100, "")), EmailTemplateEngine.Context.TEXT)).hasMessage("Template output is too large");
-      assertThatThrownBy(() -> engine.parse("{{value}}", schema).render(Map.of("value", "\u00e9".repeat(600_000)), EmailTemplateEngine.Context.TEXT))
+      assertThatThrownBy(() -> engine.parse("{{value}}", schema).render(Map.of("value", "é".repeat(600_000)), EmailTemplateEngine.Context.TEXT))
          .hasMessage("Template output is too large");
       assertThat(engine.parse("{{#if value}}".repeat(8) + "ok" + "{{/if}}".repeat(8), schema).render(Map.of("value", true), EmailTemplateEngine.Context.TEXT)).isEqualTo("ok");
       assertThatThrownBy(() -> engine.parse("{{#if value}}".repeat(9) + "{{/if}}".repeat(9), schema)).hasMessageContaining("8 levels");
@@ -179,7 +179,10 @@ class EmailReviewFixesTest
       {
          assertThatThrownBy(() -> EmailTemplateEngine.validateHtml(source)).hasMessageContaining("must be quoted");
       }
-      EmailRuntime.config().withMjmlCompiler(source -> { throw new AssertionError("Unsafe MJML reached compiler"); });
+      EmailRuntime.config().withMjmlCompiler(source ->
+      {
+         throw new AssertionError("Unsafe MJML reached compiler");
+      });
       QRecord mjml = EmailTemplatesTest.draft("includes", "MJML", "[]", "{}", "<mjml><mj-include path='secret'/></mjml>");
       assertThatThrownBy(() -> EmailVersions.activate(mjml.getValueInteger("id"))).hasMessage("MJML includes are not allowed");
       assertThat(preview(mjml)).contains("MJML includes are not allowed");
@@ -192,7 +195,10 @@ class EmailReviewFixesTest
    void viewerNeverCompiles() throws Exception
    {
       QRecord version = EmailTemplatesTest.draft("viewer", "MJML", "[]", "{}", "<mjml/>");
-      EmailRuntime.config().withMjmlCompiler(source -> { throw new AssertionError("Viewer compiled MJML"); });
+      EmailRuntime.config().withMjmlCompiler(source ->
+      {
+         throw new AssertionError("Viewer compiled MJML");
+      });
       QContext.getQSession().setPermissions(Set.of("email.view"));
       assertThat(preview(version)).contains("Not compiled yet. Activate the version or use Preview as a manager.");
       QContext.getQSession().setPermissions(Set.of("email.manage"));
@@ -270,7 +276,10 @@ class EmailReviewFixesTest
       QRecord original = EmailStore.get(EmailTables.MESSAGE, first.messageId(), null);
       SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", second.messageId()).withValue("htmlBody", original.getValue("htmlBody")));
       var dispatcher = new com.kingsrook.qbits.email.delivery.EmailDispatcher(new com.kingsrook.qbits.email.transport.EmailTransportRegistry()
-         .withTransport("SMTP", (envelope, settings) -> { throw new AssertionError("Swapped ciphertext reached transport"); }));
+         .withTransport("SMTP", (envelope, settings) ->
+         {
+            throw new AssertionError("Swapped ciphertext reached transport");
+         }));
       assertThat(dispatcher.sendNow(second.messageId())).isEqualTo(com.kingsrook.qbits.email.delivery.EmailDispatcher.SendResult.RETRY_WAIT);
    }
 

src/test/java/com/kingsrook/qbits/email/EmailSenderAddressTest.java

@@ -0,0 +1,489 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.outbox.EmailService;
+import com.kingsrook.qbits.email.permissions.EmailPermissionChecker;
+import com.kingsrook.qbits.email.processes.EmailProcesses;
+import com.kingsrook.qbits.email.support.EmailStore;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.actions.values.SearchPossibleValueSourceAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.actions.values.SearchPossibleValueSourceInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.PossibleValueSearchFilterUseCase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/*******************************************************************************
+ ** Provider sender addresses: validation, sending rules, the fallback to the
+ ** provider's own from address and every contingent possible-value pair.
+ *******************************************************************************/
+class EmailSenderAddressTest
+{
+   private QRecord relay;
+   private QRecord other;
+
+   /*******************************************************************************
+    ** Two providers on an isolated database.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      H2Support.setup(c ->
+      {
+      });
+      relay = H2Support.provider(25);
+      other = H2Support.insert(EmailTables.PROVIDER, new QRecord().withValue("name", "other").withValue("kind", "SMTP").withValue("host", "127.0.0.1")
+         .withValue("port", 25).withValue("security", "NONE").withValue("fromAddress", "other@example.com").withValue("enabled", true).withValue("isDefault", false));
+   }
+
+   /*******************************************************************************
+    ** Clear thread state.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+
+   /*******************************************************************************
+    ** A sender address row for a provider.
+    *******************************************************************************/
+   private static QRecord row(QRecord provider, String address)
+   {
+      return (new QRecord().withValue("providerId", provider.getValue("id")).withValue("address", address).withValue("enabled", true).withValue("isDefault", false));
+   }
+
+   /*******************************************************************************
+    ** Insert through the table action and return the (possibly errored) result.
+    *******************************************************************************/
+   private static QRecord tryInsert(QRecord row) throws Exception
+   {
+      return (new InsertAction().execute(new InsertInput(EmailTables.SENDER).withRecords(List.of(row))).getRecords().get(0));
+   }
+
+   /*******************************************************************************
+    ** A one-recipient request.
+    *******************************************************************************/
+   private static EmailRequest request(String providerName)
+   {
+      return (new EmailRequest().withProviderName(providerName).withTo("person@example.net").withSubject("Hello").withTextBody("Body"));
+   }
+
+   /*******************************************************************************
+    ** Addresses are canonical lowercase and line breaks, bad addresses are refused.
+    *******************************************************************************/
+   @Test
+   void addressIsCanonicalAndSafe() throws Exception
+   {
+      QRecord saved = H2Support.insert(EmailTables.SENDER, row(relay, "Sales@Example.COM"));
+      assertThat(saved.getValueString("address")).isEqualTo("sales@example.com");
+      assertThat(tryInsert(row(relay, "a@example.com\nBcc: x@example.net")).getErrors()).isNotEmpty();
+      assertThat(tryInsert(row(relay, "not an address")).getErrors()).isNotEmpty();
+      assertThat(tryInsert(row(relay, "ok@example.com").withValue("displayName", "Evil\r\nBcc: x@example.net")).getErrors()).isNotEmpty();
+      assertThat(tryInsert(new QRecord().withValue("providerId", 9999).withValue("address", "x@example.com")).getErrors()).isNotEmpty();
+   }
+
+   /*******************************************************************************
+    ** The domain must equal the provider sending domain when one is set.
+    *******************************************************************************/
+   @Test
+   void domainMustMatchSendingDomain() throws Exception
+   {
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", relay.getValue("id")).withValue("sendingDomain", "example.com"));
+      assertThat(tryInsert(row(relay, "x@other.org")).getErrors()).isNotEmpty();
+      assertThat(tryInsert(row(relay, "x@EXAMPLE.com")).getErrors()).isNullOrEmpty();
+      QRecord off = tryInsert(row(other, "x@other.org"));
+      assertThat(off.getErrors()).isNullOrEmpty();
+      var update = new UpdateAction().execute(new UpdateInput(EmailTables.PROVIDER).withRecords(List.of(new QRecord().withValue("id", other.getValue("id")).withValue("sendingDomain", "example.com"))));
+      assertThat(update.getRecords().get(0).getErrors()).isNotEmpty();
+   }
+
+   /*******************************************************************************
+    ** Unique per provider; the same address may belong to another provider.
+    *******************************************************************************/
+   @Test
+   void addressIsUniquePerProvider() throws Exception
+   {
+      H2Support.insert(EmailTables.SENDER, row(relay, "sales@example.com"));
+      assertThat(tryInsert(row(relay, "SALES@example.com")).getErrors()).isNotEmpty();
+      assertThat(tryInsert(row(other, "sales@example.com")).getErrors()).isNullOrEmpty();
+      var both = new InsertAction().execute(new InsertInput(EmailTables.SENDER).withRecords(List.of(row(relay, "twin@example.com"), row(relay, "twin@example.com")))).getRecords();
+      assertThat(both.stream().filter(r -> r.getErrors() != null && !r.getErrors().isEmpty())).isNotEmpty();
+   }
+
+   /*******************************************************************************
+    ** At most one default per provider, and a default must be enabled.
+    *******************************************************************************/
+   @Test
+   void singleEnabledDefault() throws Exception
+   {
+      H2Support.insert(EmailTables.SENDER, row(relay, "a@example.com").withValue("isDefault", true));
+      assertThat(tryInsert(row(relay, "b@example.com").withValue("isDefault", true)).getErrors()).isNotEmpty();
+      assertThat(tryInsert(row(other, "c@example.com").withValue("isDefault", true)).getErrors()).isNullOrEmpty();
+      assertThat(tryInsert(row(relay, "d@example.com").withValue("isDefault", true).withValue("enabled", false)).getErrors()).isNotEmpty();
+      QRecord plain = H2Support.insert(EmailTables.SENDER, row(relay, "e@example.com"));
+      var promote = new UpdateAction().execute(new UpdateInput(EmailTables.SENDER).withRecords(List.of(new QRecord().withValue("id", plain.getValue("id")).withValue("isDefault", true))));
+      assertThat(promote.getRecords().get(0).getErrors()).isNotEmpty();
+   }
+
+   /*******************************************************************************
+    ** Deleting a provider removes its sender addresses.
+    *******************************************************************************/
+   @Test
+   void deletingProviderRemovesSenderAddresses() throws Exception
+   {
+      H2Support.insert(EmailTables.SENDER, row(other, "gone@example.com"));
+      new DeleteAction().execute(new DeleteInput(EmailTables.PROVIDER).withPrimaryKey(other.getValue("id")));
+      assertThat(H2Support.all(EmailTables.SENDER)).isEmpty();
+   }
+
+   /*******************************************************************************
+    ** Sender rows are admin-only, like providers.
+    *******************************************************************************/
+   @Test
+   void senderRowsAreAdminOnly() throws Exception
+   {
+      var table = QContext.getQInstance().getTable(EmailTables.SENDER);
+      assertThat(table.getPermissionRules().getPermissionBaseName()).isEqualTo("email");
+      assertThat(EmailTables.MAIN).doesNotContain(EmailTables.SENDER);
+      assertThat(QContext.getQInstance().getJoin(EmailTables.PROVIDER + "_join_" + EmailTables.SENDER)).isNotNull();
+      assertThat(QContext.getQInstance().getWidget(EmailTables.SENDER + "_children")).isNotNull();
+      QContext.getQSession().setPermissions(java.util.Set.of("email.manage"));
+      var checker = new EmailPermissionChecker();
+      assertThatThrownBy(() -> checker.checkPermissionsThrowing(new InsertInput(EmailTables.SENDER), table)).hasMessageContaining("email.admin");
+      assertThatThrownBy(() -> checker.checkPermissionsThrowing(new QueryInput(EmailTables.SENDER), table)).hasMessageContaining("email.admin");
+      QContext.getQSession().setPermissions(java.util.Set.of("email.admin"));
+      checker.checkPermissionsThrowing(new InsertInput(EmailTables.SENDER), table);
+   }
+
+   /*******************************************************************************
+    ** Sending with a chosen id: ok, another provider's row, and a disabled row.
+    *******************************************************************************/
+   @Test
+   void sendWithFromAddressId() throws Exception
+   {
+      QRecord mine = H2Support.insert(EmailTables.SENDER, row(relay, "Sales@example.com").withValue("displayName", "Sales Desk"));
+      QRecord theirs = H2Support.insert(EmailTables.SENDER, row(other, "help@example.com"));
+      QRecord off = H2Support.insert(EmailTables.SENDER, row(relay, "old@example.com").withValue("enabled", false));
+      EmailSendResult ok = EmailService.send(request("relay").withFromAddressId(mine.getValueInteger("id")));
+      assertThat(ok.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      QRecord stored = EmailStore.get(EmailTables.MESSAGE, ok.messageId(), null);
+      assertThat(stored.getValueString("fromAddress")).contains("sales@example.com").contains("Sales Desk");
+      assertThat(stored.getValueInteger("fromAddressId")).isEqualTo(mine.getValueInteger("id"));
+      EmailSendResult wrong = EmailService.send(request("relay").withFromAddressId(theirs.getValueInteger("id")));
+      assertThat(wrong.status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(wrong.reasons()).containsExactly("That from address is not set up for provider relay");
+      EmailSendResult disabled = EmailService.send(request("relay").withFromAddressId(off.getValueInteger("id")));
+      assertThat(disabled.status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(disabled.reasons()).containsExactly("That from address is not set up for provider relay");
+      assertThat(EmailService.send(request("relay").withFromAddressId(4242)).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(EmailService.send(request("relay").withFromAddressId(mine.getValueInteger("id")).withFrom("other@example.com")).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+   /*******************************************************************************
+    ** Once a provider has rows an explicit from text must match an enabled one.
+    *******************************************************************************/
+   @Test
+   void fromTextMustMatchRowsWhenTheyExist() throws Exception
+   {
+      QRecord listed = H2Support.insert(EmailTables.SENDER, row(relay, "sales@example.com"));
+      H2Support.insert(EmailTables.SENDER, row(relay, "old@example.com").withValue("enabled", false));
+      EmailSendResult ok = EmailService.send(request("relay").withFrom("Sales Team <SALES@example.com>"));
+      assertThat(ok.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      QRecord stored = EmailStore.get(EmailTables.MESSAGE, ok.messageId(), null);
+      assertThat(stored.getValueString("fromAddress")).contains("Sales Team");
+      assertThat(stored.getValueInteger("fromAddressId")).isEqualTo(listed.getValueInteger("id"));
+      EmailSendResult unlisted = EmailService.send(request("relay").withFrom("random@example.com"));
+      assertThat(unlisted.status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(unlisted.reasons()).containsExactly("That from address is not set up for provider relay");
+      assertThat(EmailService.send(request("relay").withFrom("old@example.com")).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(EmailService.send(request("relay").withFrom("sender@example.com")).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+      assertThat(EmailService.send(request("relay").withFrom("not an address")).status()).isEqualTo(EmailSendResult.Status.REFUSED);
+   }
+
+   /*******************************************************************************
+    ** Without rows, behavior is unchanged: any from text, and the provider's from address and name.
+    *******************************************************************************/
+   @Test
+   void fallsBackToProviderFromAddress() throws Exception
+   {
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", relay.getValue("id")).withValue("fromName", "Relay Team"));
+      EmailSendResult plain = EmailService.send(request("relay"));
+      QRecord stored = EmailStore.get(EmailTables.MESSAGE, plain.messageId(), null);
+      assertThat(stored.getValueString("fromAddress")).contains("sender@example.com").contains("Relay Team");
+      assertThat(stored.getValue("fromAddressId")).isNull();
+      EmailSendResult explicit = EmailService.send(request("relay").withFrom("anyone@example.org"));
+      assertThat(explicit.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      assertThat(EmailStore.get(EmailTables.MESSAGE, explicit.messageId(), null).getValueString("fromAddress")).isEqualTo("anyone@example.org");
+      H2Support.insert(EmailTables.SENDER, row(other, "help@example.com"));
+      EmailSendResult unaffected = EmailService.send(request("relay").withFrom("anyone@example.org"));
+      assertThat(unaffected.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+   }
+
+   /*******************************************************************************
+    ** An enabled default row wins over the provider's own from address; a disabled one does not.
+    *******************************************************************************/
+   @Test
+   void defaultRowWins() throws Exception
+   {
+      QRecord preferred = H2Support.insert(EmailTables.SENDER, row(relay, "team@example.com").withValue("displayName", "The Team").withValue("isDefault", true));
+      EmailSendResult result = EmailService.send(request("relay"));
+      QRecord stored = EmailStore.get(EmailTables.MESSAGE, result.messageId(), null);
+      assertThat(stored.getValueString("fromAddress")).contains("team@example.com").contains("The Team");
+      assertThat(stored.getValueInteger("fromAddressId")).isEqualTo(preferred.getValueInteger("id"));
+      SystemWrites.updateOne(EmailTables.SENDER, new QRecord().withValue("id", preferred.getValue("id")).withValue("enabled", false));
+      QRecord back = EmailStore.get(EmailTables.MESSAGE, EmailService.send(request("relay")).messageId(), null);
+      assertThat(back.getValueString("fromAddress")).isEqualTo("sender@example.com");
+      assertThat(back.getValue("fromAddressId")).isNull();
+   }
+
+   /*******************************************************************************
+    ** A resend keeps the original sender row.
+    *******************************************************************************/
+   @Test
+   void resendKeepsSenderRow() throws Exception
+   {
+      QRecord mine = H2Support.insert(EmailTables.SENDER, row(relay, "sales@example.com"));
+      EmailSendResult sent = EmailService.send(request("relay").withFromAddressId(mine.getValueInteger("id")));
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", sent.messageId()).withValue("status", "DEAD_LETTER"));
+      EmailSendResult again = EmailService.resend(sent.messageId());
+      assertThat(again.status()).isEqualTo(EmailSendResult.Status.QUEUED);
+      assertThat(EmailStore.get(EmailTables.MESSAGE, again.messageId(), null).getValueInteger("fromAddressId")).isEqualTo(mine.getValueInteger("id"));
+   }
+
+   /*******************************************************************************
+    ** The templated and ad hoc send processes accept providerId and fromAddressId.
+    *******************************************************************************/
+   @Test
+   void sendProcessAcceptsProviderAndFromAddressIds() throws Exception
+   {
+      QRecord mine = H2Support.insert(EmailTables.SENDER, row(relay, "sales@example.com"));
+      QRecord theirs = H2Support.insert(EmailTables.SENDER, row(other, "help@example.com"));
+      Map<String, Serializable> values = Map.of("to", "person@example.net", "subject", "Hi", "textBody", "Body", "providerId", relay.getValueInteger("id"), "fromAddressId", mine.getValueInteger("id"));
+      var input = new RunProcessInput();
+      input.setProcessName(EmailProcesses.SEND_AD_HOC);
+      values.forEach(input::addValue);
+      assertThat(new RunProcessAction().execute(input).getValues().get("status")).isEqualTo("QUEUED");
+      var wrong = new RunProcessInput();
+      wrong.setProcessName(EmailProcesses.SEND_AD_HOC);
+      values.forEach(wrong::addValue);
+      wrong.addValue("fromAddressId", theirs.getValueInteger("id"));
+      assertThat(new RunProcessAction().execute(wrong).getValues().get("status")).isEqualTo("REFUSED");
+      var named = new RunProcessInput();
+      named.setProcessName(EmailProcesses.SEND_AD_HOC);
+      values.forEach(named::addValue);
+      named.addValue("providerName", "relay");
+      assertThat(new RunProcessAction().execute(named).getValues().get("status")).isEqualTo("QUEUED");
+      var mismatch = new RunProcessInput();
+      mismatch.setProcessName(EmailProcesses.SEND_AD_HOC);
+      values.forEach(mismatch::addValue);
+      mismatch.addValue("providerName", "other");
+      assertThatThrownBy(() -> new RunProcessAction().execute(mismatch)).hasMessageContaining("providerId does not match");
+      var missing = new RunProcessInput();
+      missing.setProcessName(EmailProcesses.SEND_AD_HOC);
+      values.forEach(missing::addValue);
+      missing.addValue("providerId", 4242);
+      assertThatThrownBy(() -> new RunProcessAction().execute(missing)).hasMessageContaining("providerId does not match");
+   }
+
+   /*******************************************************************************
+    ** Send Test Email takes an optional from address that must belong to the provider.
+    *******************************************************************************/
+   @Test
+   void sendTestUsesFromAddress() throws Exception
+   {
+      QRecord mine = H2Support.insert(EmailTables.SENDER, row(relay, "sales@example.com"));
+      QRecord theirs = H2Support.insert(EmailTables.SENDER, row(other, "help@example.com"));
+      var wrong = ProcessRunner.runWithValues(EmailProcesses.SEND_TEST, Map.of("toAddress", "test@example.com", "fromAddressId", theirs.getValueInteger("id")), relay.getValue("id"));
+      assertThat(wrong.result()).anyMatch(s -> s.contains("REFUSED"));
+      var ok = ProcessRunner.runWithValues(EmailProcesses.SEND_TEST, Map.of("toAddress", "test@example.com", "fromAddressId", mine.getValueInteger("id")), relay.getValue("id"));
+      assertThat(ok.result()).anyMatch(s -> s.contains("FAILED") || s.contains("SENT") || s.contains("RETRY") || s.contains("QUEUED"));
+      assertThat(H2Support.all(EmailTables.MESSAGE)).anyMatch(m -> mine.getValue("id").equals(m.getValue("fromAddressId")));
+   }
+
+   // ----- contingent possible-value pairs -----
+
+   /*******************************************************************************
+    ** Assert the filter on a field and that a possible-value search with the parent
+    ** value returns only that parent's rows. The search runs through core's action.
+    *******************************************************************************/
+   private static void assertContingent(QFieldMetaData field, String inputName, Serializable parentValue, List<Serializable> expected, List<Serializable> excluded) throws Exception
+   {
+      assertThat(field.getPossibleValueSourceFilter()).isNotNull();
+      var filter = field.getPossibleValueSourceFilter().clone();
+      assertThat(filter.toString()).contains("${input." + inputName + "}");
+      filter.interpretValues(Map.of("input", Map.of(inputName, parentValue), "processValues", Map.<String, Serializable>of()), PossibleValueSearchFilterUseCase.FORM);
+      var found = new SearchPossibleValueSourceAction().execute(new SearchPossibleValueSourceInput().withPossibleValueSourceName(field.getPossibleValueSourceName()).withDefaultQueryFilter(filter));
+      List<String> ids = found.getResults().stream().map(v -> v.getId().toString()).toList();
+      assertThat(ids).containsAll(expected.stream().map(String::valueOf).toList());
+      assertThat(ids).doesNotContainAnyElementsOf(excluded.stream().map(String::valueOf).toList());
+   }
+
+   /*******************************************************************************
+    ** One row each for the two providers.
+    *******************************************************************************/
+   private QRecord[] senders() throws Exception
+   {
+      return (new QRecord[] { H2Support.insert(EmailTables.SENDER, row(relay, "a@example.com")), H2Support.insert(EmailTables.SENDER, row(other, "b@example.com")) });
+   }
+
+   /*******************************************************************************
+    ** Templated send: fromAddressId follows providerId.
+    *******************************************************************************/
+   @Test
+   void templatedSendFromAddressFollowsProvider() throws Exception
+   {
+      QRecord[] rows = senders();
+      QFieldMetaData field = QContext.getQInstance().getProcess(EmailProcesses.SEND_TEMPLATED).getBackendStep("send").getInputMetaData().getFieldThrowing("fromAddressId");
+      assertThat(field.getPossibleValueSourceName()).isEqualTo(EmailTables.SENDER);
+      assertContingent(field, "providerId", relay.getValueInteger("id"), List.of(rows[0].getValueInteger("id")), List.of(rows[1].getValueInteger("id")));
+      assertContingent(field, "providerId", other.getValueInteger("id"), List.of(rows[1].getValueInteger("id")), List.of(rows[0].getValueInteger("id")));
+      assertThat(QContext.getQInstance().getProcess(EmailProcesses.SEND_TEMPLATED).getBackendStep("send").getInputMetaData().getFieldThrowing("providerId").getPossibleValueSourceName()).isEqualTo(EmailTables.PROVIDER);
+   }
+
+   /*******************************************************************************
+    ** Ad hoc send: fromAddressId follows providerId.
+    *******************************************************************************/
+   @Test
+   void adHocSendFromAddressFollowsProvider() throws Exception
+   {
+      QRecord[] rows = senders();
+      QFieldMetaData field = QContext.getQInstance().getProcess(EmailProcesses.SEND_AD_HOC).getBackendStep("send").getInputMetaData().getFieldThrowing("fromAddressId");
+      assertContingent(field, "providerId", relay.getValueInteger("id"), List.of(rows[0].getValueInteger("id")), List.of(rows[1].getValueInteger("id")));
+   }
+
+   /*******************************************************************************
+    ** Send Test Email: fromAddressId follows the provider being acted on, which a
+    ** record action's form only gets from the prepare step's process value.
+    *******************************************************************************/
+   @Test
+   void sendTestFromAddressFollowsActedOnProvider() throws Exception
+   {
+      QRecord[] rows = senders();
+      QFieldMetaData field = QContext.getQInstance().getProcess(EmailProcesses.SEND_TEST).getFrontendStep("input").getFormFields().stream()
+         .filter(f -> "fromAddressId".equals(f.getName())).findFirst().orElseThrow();
+      assertContingent(field, "id", relay.getValueInteger("id"), List.of(rows[0].getValueInteger("id")), List.of(rows[1].getValueInteger("id")));
+      var input = new RunProcessInput();
+      input.setProcessName(EmailProcesses.SEND_TEST);
+      input.addValue("recordsParam", "recordIds");
+      input.addValue("recordIds", String.valueOf(other.getValue("id")));
+      input.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.BREAK);
+      var seeded = new RunProcessAction().execute(input).getValues().get(EmailProcesses.PROVIDER_FILTER_VALUE);
+      assertThat(seeded).isEqualTo(other.getValue("id"));
+      var filter = field.getPossibleValueSourceFilter().clone();
+      filter.interpretValues(Map.of("input", Map.<String, Serializable>of(), "processValues", Map.of(EmailProcesses.PROVIDER_FILTER_VALUE, (Serializable) seeded)), PossibleValueSearchFilterUseCase.FORM);
+      var found = new SearchPossibleValueSourceAction().execute(new SearchPossibleValueSourceInput().withPossibleValueSourceName(EmailTables.SENDER).withDefaultQueryFilter(filter));
+      assertThat(found.getResults().stream().map(v -> v.getId().toString()).toList()).containsExactly(rows[1].getValueString("id"));
+      var several = new RunProcessInput();
+      several.setProcessName(EmailProcesses.SEND_TEST);
+      several.addValue("recordsParam", "recordIds");
+      several.addValue("recordIds", "1,2");
+      several.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.BREAK);
+      assertThat(new RunProcessAction().execute(several).getValues()).doesNotContainKey(EmailProcesses.PROVIDER_FILTER_VALUE);
+      var none = new RunProcessInput();
+      none.setProcessName(EmailProcesses.SEND_TEST);
+      none.addValue("recordsParam", "recordIds");
+      none.addValue("recordIds", "x");
+      none.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.BREAK);
+      assertThat(new RunProcessAction().execute(none).getValues()).doesNotContainKey(EmailProcesses.PROVIDER_FILTER_VALUE);
+   }
+
+   /*******************************************************************************
+    ** Message: fromAddressId follows providerId, and is read-only.
+    *******************************************************************************/
+   @Test
+   void messageFromAddressFollowsProvider() throws Exception
+   {
+      QRecord[] rows = senders();
+      QFieldMetaData field = QContext.getQInstance().getTable(EmailTables.MESSAGE).getField("fromAddressId");
+      assertThat(field.getIsEditable()).isFalse();
+      assertContingent(field, "providerId", relay.getValueInteger("id"), List.of(rows[0].getValueInteger("id")), List.of(rows[1].getValueInteger("id")));
+   }
+
+   /*******************************************************************************
+    ** Attempt and event: the message follows the provider.
+    *******************************************************************************/
+   @Test
+   void attemptAndEventMessageFollowProvider() throws Exception
+   {
+      QRecord fromRelay = EmailStore.get(EmailTables.MESSAGE, EmailService.send(request("relay")).messageId(), null);
+      QRecord fromOther = EmailStore.get(EmailTables.MESSAGE, EmailService.send(request("other")).messageId(), null);
+      for(String table : List.of(EmailTables.ATTEMPT, EmailTables.EVENT))
+      {
+         QFieldMetaData field = QContext.getQInstance().getTable(table).getField("messageId");
+         assertContingent(field, "providerId", relay.getValueInteger("id"), List.of(fromRelay.getValueInteger("id")), List.of(fromOther.getValueInteger("id")));
+         assertContingent(field, "providerId", other.getValueInteger("id"), List.of(fromOther.getValueInteger("id")), List.of(fromRelay.getValueInteger("id")));
+      }
+   }
+
+   /*******************************************************************************
+    ** Suppression: the source event follows the source message.
+    *******************************************************************************/
+   @Test
+   void suppressionSourceEventFollowsSourceMessage() throws Exception
+   {
+      QRecord first = EmailStore.get(EmailTables.MESSAGE, EmailService.send(request("relay")).messageId(), null);
+      QRecord second = EmailStore.get(EmailTables.MESSAGE, EmailService.send(request("other")).messageId(), null);
+      QRecord e1 = SystemWrites.insertOne(EmailTables.EVENT, new QRecord().withValue("messageId", first.getValue("id")).withValue("eventType", "BOUNCED").withValue("eventKey", "k1"));
+      QRecord e2 = SystemWrites.insertOne(EmailTables.EVENT, new QRecord().withValue("messageId", second.getValue("id")).withValue("eventType", "BOUNCED").withValue("eventKey", "k2"));
+      QFieldMetaData field = QContext.getQInstance().getTable(EmailTables.SUPPRESSION).getField("sourceEventId");
+      assertContingent(field, "sourceMessageId", first.getValueInteger("id"), List.of(e1.getValueInteger("id")), List.of(e2.getValueInteger("id")));
+      assertContingent(field, "sourceMessageId", second.getValueInteger("id"), List.of(e2.getValueInteger("id")), List.of(e1.getValueInteger("id")));
+   }
+
+   /*******************************************************************************
+    ** Fields without a parent scope stay unfiltered.
+    *******************************************************************************/
+   @Test
+   void independentRelationshipsStayUnfiltered() throws Exception
+   {
+      var instance = QContext.getQInstance();
+      assertThat(instance.getTable(EmailTables.TEMPLATE).getField("brandId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.MESSAGE).getField("brandId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.MESSAGE).getField("providerId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.MESSAGE).getField("templateId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.VERSION).getField("templateId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.INBOX).getField("providerId").getPossibleValueSourceFilter()).isNull();
+      assertThat(instance.getTable(EmailTables.SENDER).getField("providerId").getPossibleValueSourceFilter()).isNull();
+   }
+}

src/test/java/com/kingsrook/qbits/email/EmailTemplatesTest.java

@@ -148,7 +148,10 @@ class EmailTemplatesTest
       assertThat(((RawHTML) preview.getWidgetData()).getHtml()).contains("<p>Email</p>");
       assertThat(EmailStore.get(EmailTables.VERSION, version.getValueInteger("id"), null).getValue("compiledHtml")).isNull();
       EmailVersions.activate(version.getValueInteger("id"));
-      EmailRuntime.config().withMjmlCompiler(source -> { throw new AssertionError("Compiler ran at send time"); });
+      EmailRuntime.config().withMjmlCompiler(source ->
+      {
+         throw new AssertionError("Compiler ran at send time");
+      });
       var result = EmailService.send(new EmailRequest().withTemplateName("mjml").withLocale("fr").withTo("p@example.com"));
       assertThat(result.status()).isEqualTo(EmailSendResult.Status.QUEUED);
       QRecord row = EmailStore.get(EmailTables.MESSAGE, result.messageId(), null);
@@ -295,7 +298,10 @@ class EmailTemplatesTest
       EmailRuntime.config().withSecretResolver(name -> null);
       assertThat(com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "address", "NOTIFICATION")).isNull();
       assertThatThrownBy(() -> com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "address", "MARKETING")).hasMessageContaining("QQQ_EMAIL_UNSUBSCRIBE_KEY");
-      EmailRuntime.config().withSecretResolver(name -> { throw new IllegalArgumentException("resolver detail"); });
+      EmailRuntime.config().withSecretResolver(name ->
+      {
+         throw new IllegalArgumentException("resolver detail");
+      });
       assertThatThrownBy(() -> com.kingsrook.qbits.email.render.EmailUnsubscribeLinks.create("1", "person@example.com", "NOTIFICATION"))
          .hasMessage("Opt-out email needs publicBaseUrl and QQQ_EMAIL_UNSUBSCRIBE_KEY");
    }

src/test/java/com/kingsrook/qbits/email/H2Support.java

@@ -85,7 +85,7 @@ public final class H2Support
     *******************************************************************************/
    public static QInstance instance(RDBMSBackendMetaData backend, Consumer<EmailQBitConfig> tweak)
          throws Exception
-         {
+   {
       QInstance instance = new QInstance();
       instance.withInstanceDefaultAuthentication(
             new QAuthenticationMetaData()

src/test/java/com/kingsrook/qbits/email/PhiCanaryEgressTest.java

@@ -81,7 +81,17 @@ class PhiCanaryEgressTest
    @BeforeEach
    void setup() throws Exception
    {
-      canary = "CANARY-PHI-" + UUID.randomUUID();
+      ////////////////////////////////////////////////////////////////////////
+      // the random part is mapped from hex to the letters g-v: a five-char //
+      // window of a hex canary could match any hex digest (body hashes,    //
+      // Message-IDs) by chance; these windows can only match real leakage  //
+      ////////////////////////////////////////////////////////////////////////
+      StringBuilder random = new StringBuilder();
+      for(char c : UUID.randomUUID().toString().replace("-", "").toCharArray())
+      {
+         random.append((char) ('g' + Character.digit(c, 16)));
+      }
+      canary = "CANARY-PHI-" + random;
       H2Support.setup(c -> c.withMaxAttempts(2).withSecretResolver(PhiCanaryEgressTest::secret)
          .withClassificationOverrides(Map.of("test_patient.name", EmailClassifications.Level.PHI)));
       H2Support.provider(25);

src/test/java/com/kingsrook/qbits/email/ProcessRunner.java

@@ -66,7 +66,7 @@ public final class ProcessRunner
    public static Lines runWithValues(
          String process, java.util.Map<String, Serializable> values, Serializable... ids)
          throws Exception
-         {
+   {
       RunProcessInput input = new RunProcessInput();
       input.setProcessName(process);
       input.addValue("recordsParam", "recordIds");

src/test/java/com/kingsrook/qbits/email/api/EmailApiBoundaryTest.java

@@ -0,0 +1,219 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.lang.reflect.GenericArrayType;
+import java.lang.reflect.Modifier;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
+import java.lang.reflect.WildcardType;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashSet;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** No ArchUnit dependency: inspect every compiled API class and nested type,
+ ** including generic arguments, bounds, arrays, exceptions and inheritance.
+ *******************************************************************************/
+class EmailApiBoundaryTest
+{
+   /*******************************************************************************
+    ** Discovery uses production output so new API classes cannot escape review.
+    *******************************************************************************/
+   @Test
+   void publishedSignaturesUseOnlyApiJdkAndQqqCoreTypes() throws Exception
+   {
+      Path root = Path.of(EmailApi.class.getProtectionDomain().getCodeSource().getLocation().toURI());
+      Path packagePath = root.resolve("com/kingsrook/qbits/email/api");
+      int count = 0;
+      try(var files = Files.walk(packagePath))
+      {
+         for(Path path : files.filter(p -> p.toString().endsWith(".class")).toList())
+         {
+            String name = root.relativize(path).toString().replace('\\', '.').replace('/', '.').replaceFirst("\\.class$", "");
+            inspect(Class.forName(name));
+            count++;
+         }
+      }
+      assertThat(count).isGreaterThanOrEqualTo(11);
+   }
+
+
+
+   /*******************************************************************************
+    ** Prove the guard catches indirect leaks, not just raw return classes.
+    *******************************************************************************/
+   @Test
+   void rejectsNestedGenericAndProtectedLeaks() throws Exception
+   {
+      assertThatThrownBy(() -> inspect(LeakingGeneric.class)).isInstanceOf(AssertionError.class);
+      assertThatThrownBy(() -> inspect(LeakingProtected.class)).isInstanceOf(AssertionError.class);
+   }
+
+
+
+   /*******************************************************************************
+    ** Constructors, fields, inherited methods and protected declarations count.
+    *******************************************************************************/
+   private void inspect(Class<?> apiClass)
+   {
+      Set<Type> seen = new HashSet<>();
+      check(apiClass.getGenericSuperclass(), seen);
+      checkAll(apiClass.getGenericInterfaces(), seen);
+      checkAll(apiClass.getTypeParameters(), seen);
+      for(var field : apiClass.getDeclaredFields())
+      {
+         if(visible(field.getModifiers()))
+         {
+            check(field.getGenericType(), seen);
+         }
+      }
+      for(var constructor : apiClass.getDeclaredConstructors())
+      {
+         if(visible(constructor.getModifiers()))
+         {
+            checkAll(constructor.getGenericParameterTypes(), seen);
+            checkAll(constructor.getGenericExceptionTypes(), seen);
+            checkAll(constructor.getTypeParameters(), seen);
+         }
+      }
+      Set<java.lang.reflect.Method> methods = new HashSet<>(java.util.List.of(apiClass.getDeclaredMethods()));
+      methods.addAll(java.util.List.of(apiClass.getMethods()));
+      for(var method : methods)
+      {
+         if(visible(method.getModifiers()))
+         {
+            check(method.getGenericReturnType(), seen);
+            checkAll(method.getGenericParameterTypes(), seen);
+            checkAll(method.getGenericExceptionTypes(), seen);
+            checkAll(method.getTypeParameters(), seen);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Public and protected form the published contract.
+    *******************************************************************************/
+   private boolean visible(int modifiers)
+   {
+      return (Modifier.isPublic(modifiers) || Modifier.isProtected(modifiers));
+   }
+
+
+
+   /*******************************************************************************
+    ** Walk each type without looping on recursive type bounds.
+    *******************************************************************************/
+   private void checkAll(Type[] types, Set<Type> seen)
+   {
+      for(Type type : types)
+      {
+         check(type, seen);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Generic containers cannot hide an implementation or third-party type.
+    *******************************************************************************/
+   private void check(Type type, Set<Type> seen)
+   {
+      if(type == null || !seen.add(type))
+      {
+         return;
+      }
+      if(type instanceof Class<?> clazz)
+      {
+         if(clazz.isArray())
+         {
+            check(clazz.getComponentType(), seen);
+         }
+         else
+         {
+            String name = clazz.getName();
+            String module = clazz.getModule().getName();
+            boolean jdk = module != null && (module.startsWith("java.") || module.startsWith("jdk."));
+            assertThat(clazz.isPrimitive() || jdk
+               || name.startsWith("com.kingsrook.qqq.backend.core.") || name.startsWith("com.kingsrook.qbits.email.api."))
+               .as("Published signature type %s", name).isTrue();
+         }
+      }
+      else if(type instanceof ParameterizedType parameterized)
+      {
+         check(parameterized.getRawType(), seen);
+         check(parameterized.getOwnerType(), seen);
+         checkAll(parameterized.getActualTypeArguments(), seen);
+      }
+      else if(type instanceof TypeVariable<?> variable)
+      {
+         checkAll(variable.getBounds(), seen);
+      }
+      else if(type instanceof WildcardType wildcard)
+      {
+         checkAll(wildcard.getUpperBounds(), seen);
+         checkAll(wildcard.getLowerBounds(), seen);
+      }
+      else if(type instanceof GenericArrayType array)
+      {
+         check(array.getGenericComponentType(), seen);
+      }
+      else
+      {
+         throw (new AssertionError("Unhandled signature type: " + type));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Deliberately invalid fixture, outside production class discovery.
+    *******************************************************************************/
+   private static class LeakingGeneric
+   {
+      /*******************************************************************************
+       ** A wildcard and array must not disguise an internal request type.
+       *******************************************************************************/
+      public java.util.List<? extends com.kingsrook.qbits.email.outbox.EmailRequest[]> leak()
+      {
+         return (null);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Protected fields are also part of the API contract.
+    *******************************************************************************/
+   private static class LeakingProtected
+   {
+      protected com.kingsrook.qbits.email.EmailQBitConfig config;
+   }
+}

src/test/java/com/kingsrook/qbits/email/api/EmailApiTest.java

@@ -0,0 +1,555 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.api;
+
+
+import java.lang.reflect.Method;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import com.kingsrook.qbits.email.EmailQBitConfig;
+import com.kingsrook.qbits.email.EmailRuntime;
+import com.kingsrook.qbits.email.H2Support;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.outbox.EmailCategory;
+import com.kingsrook.qbits.email.outbox.EmailEffects;
+import com.kingsrook.qbits.email.outbox.EmailRequest;
+import com.kingsrook.qbits.email.outbox.EmailSendResult;
+import com.kingsrook.qbits.email.support.EmailApiMapper;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.actions.messaging.SendMessageAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Content;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.Party;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.SendMessageInput;
+import com.kingsrook.qqq.backend.core.model.actions.messaging.email.EmailContentRole;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Stable API mapping and behavior against the real outbox and H2 database.
+ *******************************************************************************/
+class EmailApiTest
+{
+   private QInstance instance;
+   private QRecord provider;
+
+
+
+   /*******************************************************************************
+    ** No environment secrets are used by these tests.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      instance = H2Support.setup(config -> config.withSecretResolver(name -> null));
+      provider = H2Support.provider(25);
+   }
+
+
+
+   /*******************************************************************************
+    ** Release thread-local context.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+
+
+
+   /*******************************************************************************
+    ** Every request property reaches the internal request with the same value.
+    *******************************************************************************/
+   @Test
+   void mapsEveryRequestField() throws Exception
+   {
+      Instant after = Instant.parse("2030-01-02T03:04:05Z");
+      EmailApiRequest request = new EmailApiRequest().withTo(new ArrayList<>(List.of("to@example.com"))).withTo("second@example.com")
+         .withCc(new ArrayList<>(List.of("cc@example.com"))).withCc("cc2@example.com")
+         .withBcc(new ArrayList<>(List.of("bcc@example.com"))).withBcc("bcc2@example.com")
+         .withReplyTo("reply@example.com").withFrom("from@example.com").withFromAddressId(8)
+         .withSubject("Subject").withMarkdownBody("Markdown").withHtmlBody("<p>HTML</p>").withTextBody("Text")
+         .withTemplateName("notice").withVariables(new java.util.LinkedHashMap<>(Map.of("number", 7)))
+         .withVariable("label", "value").withRecordVariables(new java.util.LinkedHashMap<>())
+         .withRecordVariable("order", "orders", 42).withCategory(EmailApiCategory.SECURITY)
+         .withProviderName("named").withBrandName("brand").withLocale("fr").withIdempotencyKey("api:42")
+         .withLinkedTableName("old").withLinkedRecordId("old").withLinkedRecord("orders", 42)
+         .withSenderQbit("old").withSenderProcess("old").withSender("orders", "complete").withSendAfter(after)
+         .withTransaction(new QBackendTransaction(), "rdbms");
+      EmailRequest mapped = EmailApiMapper.request(request);
+      for(Method getter : EmailApiRequest.class.getDeclaredMethods())
+      {
+         if(!getter.getName().startsWith("get"))
+         {
+            continue;
+         }
+         Object value = getter.invoke(request);
+         assertThat(value).as(getter.getName()).isNotNull();
+         if(getter.getName().equals("getCategory"))
+         {
+            assertThat(mapped.getCategory().name()).isEqualTo(request.getCategory().name());
+         }
+         else if(getter.getName().equals("getRecordVariables"))
+         {
+            assertThat(mapped.getRecordVariables()).hasSize(1);
+            assertThat(mapped.getRecordVariables().get("order").table()).isEqualTo("orders");
+            assertThat(mapped.getRecordVariables().get("order").id()).isEqualTo(42);
+         }
+         else
+         {
+            assertThat(EmailRequest.class.getMethod(getter.getName()).invoke(mapped)).as(getter.getName()).isEqualTo(value);
+         }
+      }
+      assertThat(mapped.getTo()).containsExactly("to@example.com", "second@example.com");
+      assertThat(mapped.getLinkedRecordId()).isEqualTo("42");
+      assertThat(mapped.getSenderQbit()).isEqualTo("orders");
+      assertThat(new EmailApiRequest().withLinkedRecord("orders", null).getLinkedRecordId()).isNull();
+      assertThat(EmailApiMapper.request(new EmailApiRequest()).getCategory()).isNull();
+      assertThat(EmailApiMapper.request(new EmailApiRequest()).getLocale()).isEqualTo("en");
+      for(EmailApiCategory category : EmailApiCategory.values())
+      {
+         assertThat(EmailApiMapper.request(new EmailApiRequest().withCategory(category)).getCategory()).isEqualTo(EmailCategory.valueOf(category.name()));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** All statuses and refusal reasons survive mapping without sharing lists.
+    *******************************************************************************/
+   @Test
+   void mapsEveryStatus()
+   {
+      for(EmailSendResult.Status status : EmailSendResult.Status.values())
+      {
+         List<String> reasons = new ArrayList<>(List.of("Reason"));
+         EmailApiResult result = EmailApiMapper.result(new EmailSendResult(42, status, reasons));
+         reasons.clear();
+         assertThat(result.messageId()).isEqualTo(42);
+         assertThat(result.status()).isEqualTo(EmailApiStatus.valueOf(status.name()));
+         assertThat(result.reasons()).containsExactly("Reason");
+         assertThatThrownBy(() -> result.reasons().clear()).isInstanceOf(UnsupportedOperationException.class);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Sends keep validation, idempotency, suppression and permission behavior.
+    *******************************************************************************/
+   @Test
+   void sendsThroughExistingService() throws Exception
+   {
+      EmailApiResult queued = EmailApi.send(request("same"));
+      assertThat(queued.status()).isEqualTo(EmailApiStatus.QUEUED);
+      assertThat(EmailApi.send(request("same")).status()).isEqualTo(EmailApiStatus.DUPLICATE);
+      assertThat(EmailApi.send(request("same")).messageId()).isEqualTo(queued.messageId());
+      assertThat(EmailApi.send(null).status()).isEqualTo(EmailApiStatus.REFUSED);
+      assertThat(EmailApi.send(request("bad").withSubject("bad\nsubject")).status()).isEqualTo(EmailApiStatus.REFUSED);
+      SystemWrites.insertOne(EmailTables.SUPPRESSION, new QRecord().withValue("address", "pat@example.com")
+         .withValue("reason", "HARD_BOUNCE").withValue("scope", "ALL"));
+      assertThat(EmailApi.send(request("suppressed")).status()).isEqualTo(EmailApiStatus.SUPPRESSED);
+      instance.setEnforcePermissionsInCoreActions(true);
+      QContext.getQSession().setPermissions(Set.of());
+      assertThat(EmailApi.send(request("denied")).reasons()).contains("Sending email needs email.send");
+      assertThat(EmailApi.getMessage(queued.messageId())).isPresent();
+      assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix("same", 1)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Durable handoffs and internal replays agree on the committed row and header.
+    *******************************************************************************/
+   @Test
+   void sendsOnceAndReconcilesCurrentMetadata() throws Exception
+   {
+      String key = "job:2:row:email_%";
+      EmailApiRequest request = request(null).withLinkedRecord("orders", 42);
+      EmailApiEffectResult first = EmailApi.sendOnce(request, key);
+      assertThat(request.getIdempotencyKey()).isEqualTo(key);
+      assertThat(first.outcome()).isEqualTo(EmailApiEffectOutcome.QUEUED_OR_DUPLICATE);
+      assertThat(first.message().status()).isEqualTo("QUEUED");
+      assertThat(first.message().idempotencyKey()).isEqualTo(key);
+      assertThat(first.message().rfcMessageId()).matches("<[0-9a-f]{32}@example\\.com>");
+      assertThat(EmailApi.sendOnce(request(key), key)).isEqualTo(first);
+      var internal = EmailEffects.send(EmailApiMapper.request(request(null)), key);
+      assertThat(internal.outcome().name()).isEqualTo(first.outcome().name());
+      assertThat(internal.message().id()).isEqualTo(first.message().id());
+      assertThat(internal.message().rfcMessageId()).isEqualTo(first.message().rfcMessageId());
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+      assertThat(EmailApi.getMessage(first.message().id())).contains(first.message());
+      assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix("job:", 5)).containsExactly(first.message());
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", first.message().id())
+         .withValue("status", "SENT").withValue("providerMessageId", "provider-reference"));
+      instance.setEnforcePermissionsInCoreActions(true);
+      QContext.getQSession().setPermissions(Set.of());
+      EmailApiMessage found = EmailApi.getMessageByIdempotencyKey(key).orElseThrow();
+      assertThat(found.id()).isEqualTo(first.message().id());
+      assertThat(found.status()).isEqualTo("SENT");
+      assertThat(found.providerMessageId()).isEqualTo("provider-reference");
+      assertThat(found.rfcMessageId()).isEqualTo(first.message().rfcMessageId());
+      assertThat(found.linkedRecordId()).isEqualTo("42");
+      assertThat(EmailApi.getMessageByRfcMessageId(found.rfcMessageId())).contains(found);
+      for(String missing : new String[] {null, "", " ", "missing", "job:", "job:2:row:email_", "provider-reference"})
+      {
+         assertThat(EmailApi.getMessageByIdempotencyKey(missing)).isEmpty();
+         assertThat(EmailApi.getMessageByRfcMessageId(missing)).isEmpty();
+      }
+      assertThat(EmailApi.getMessageByRfcMessageId(found.rfcMessageId().replace("<", "").replace(">", ""))).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Invalid handoffs preserve caller keys and never create an outbox row.
+    *******************************************************************************/
+   @Test
+   void sendOnceRefusesInvalidRequestsAndKeys() throws Exception
+   {
+      for(String key : new String[] {null, "", " ", "a".repeat(501)})
+      {
+         EmailApiRequest request = request(null);
+         assertThat(EmailApi.sendOnce(request, key)).isEqualTo(new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+         assertThat(request.getIdempotencyKey()).isNull();
+         assertThat(EmailEffects.send(EmailApiMapper.request(request), key).outcome().name()).isEqualTo("REFUSED");
+      }
+      EmailApiRequest conflict = request("other");
+      assertThat(EmailApi.sendOnce(conflict, "effect").outcome()).isEqualTo(EmailApiEffectOutcome.REFUSED);
+      assertThat(conflict.getIdempotencyKey()).isEqualTo("other");
+      assertThat(EmailApi.sendOnce(null, "effect")).isEqualTo(new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+      assertThat(EmailApi.sendOnce(request(null).withSubject(null), "invalid"))
+         .isEqualTo(new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+      assertThat(EmailApi.sendOnce(request(null), "a".repeat(500)).outcome()).isEqualTo(EmailApiEffectOutcome.QUEUED_OR_DUPLICATE);
+      instance.setEnforcePermissionsInCoreActions(true);
+      QContext.getQSession().setPermissions(Set.of());
+      assertThat(EmailApi.sendOnce(request(null), "denied")).isEqualTo(new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Replaying a suppressed audit row remains refused with the same identity.
+    *******************************************************************************/
+   @Test
+   void sendOnceRefusesSuppressedDuplicates() throws Exception
+   {
+      SystemWrites.insertOne(EmailTables.SUPPRESSION, new QRecord().withValue("address", "pat@example.com")
+         .withValue("reason", "HARD_BOUNCE").withValue("scope", "ALL"));
+      EmailApiEffectResult first = EmailApi.sendOnce(request(null), "suppressed");
+      assertThat(first.outcome()).isEqualTo(EmailApiEffectOutcome.REFUSED);
+      assertThat(first.message().status()).isEqualTo("SUPPRESSED");
+      assertThat(first.message().rfcMessageId()).isNotBlank();
+      assertThat(EmailApi.sendOnce(request(null), "suppressed")).isEqualTo(first);
+      var internal = EmailEffects.send(EmailApiMapper.request(request(null)), "suppressed");
+      assertThat(internal.outcome().name()).isEqualTo(first.outcome().name());
+      assertThat(internal.message().id()).isEqualTo(first.message().id());
+      assertThat(internal.message().rfcMessageId()).isEqualTo(first.message().rfcMessageId());
+      assertThat(EmailApi.getMessageByIdempotencyKey("suppressed")).contains(first.message());
+      assertThat(EmailApi.getMessageByRfcMessageId(first.message().rfcMessageId())).contains(first.message());
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Neither a shared-backend transaction nor a deferred callback is a handoff.
+    *******************************************************************************/
+   @Test
+   void sendOnceRefusesCallerTransactions() throws Exception
+   {
+      try(QBackendTransaction transaction = QBackendTransaction.openFor(new InsertInput(EmailTables.MESSAGE)))
+      {
+         for(String backend : new String[] {null, "rdbms", "other"})
+         {
+            EmailApiRequest request = request(null).withTransaction(transaction, backend);
+            assertThat(EmailApi.sendOnce(request, "transaction")).isEqualTo(new EmailApiEffectResult(EmailApiEffectOutcome.REFUSED, null));
+            assertThat(request.getIdempotencyKey()).isNull();
+         }
+         transaction.commit();
+      }
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Ordinary sends retain caller transaction behavior through the public mapper.
+    *******************************************************************************/
+   @Test
+   void ordinarySendCanRollBackWithCaller() throws Exception
+   {
+      try(QBackendTransaction transaction = QBackendTransaction.openFor(new InsertInput(EmailTables.MESSAGE)))
+      {
+         assertThat(EmailApi.send(request("rollback").withTransaction(transaction, "rdbms")).status()).isEqualTo(EmailApiStatus.QUEUED);
+         transaction.rollback();
+      }
+      assertThat(EmailApi.getMessageByIdempotencyKey("rollback")).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Uncertain storage failures propagate so dispatchers can retry or reconcile.
+    *******************************************************************************/
+   @Test
+   void sendOncePropagatesStorageFailure() throws Exception
+   {
+      EmailApiEffectResult first = EmailApi.sendOnce(request(null), "collision");
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", first.message().id()).withValue("idempotencyKey", "changed"));
+      assertThatThrownBy(() -> EmailApi.sendOnce(request(null), "collision")).isInstanceOf(QException.class);
+      assertThat(EmailApi.getMessageByIdempotencyKey("collision")).isEmpty();
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Plain sends need only QQQ core messaging types at the consumer boundary.
+    *******************************************************************************/
+   @Test
+   void coreOnlyBridgeQueuesPlainMail() throws Exception
+   {
+      new SendMessageAction().execute(new SendMessageInput().withMessagingProviderName("qbitEmail")
+         .withTo(new Party().withAddress("pat@example.com")).withSubject("Ready")
+         .withContent(new Content().withContentRole(EmailContentRole.TEXT).withBody("Export ready")));
+      assertThat(H2Support.all(EmailTables.MESSAGE)).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueString("subject")).isEqualTo("Ready");
+         assertThat(row.getValueString("status")).isEqualTo("QUEUED");
+      });
+   }
+
+
+
+   /*******************************************************************************
+    ** Template fields are preserved, creation is idempotent, and edits survive.
+    *******************************************************************************/
+   @Test
+   void ensuresAndMapsTemplate() throws Exception
+   {
+      EmailApiTemplate template = template().withLabel("Notice").withKind("HTML").withTextBody("Text {{number}}")
+         .withVariables(List.of(new EmailApiVariable("number", "Number", "STRING", true, "PUBLIC")))
+         .withSampleVariables("{\"number\":\"42\"}").withOwnerQbit("orders").withEditPermissionName("orders.edit");
+      var mapped = EmailApiMapper.template(template);
+      for(Method getter : EmailApiTemplate.class.getDeclaredMethods())
+      {
+         if(getter.getName().startsWith("get") && !Set.of("getVariables", "getVariablesSchema", "getCategory").contains(getter.getName()))
+         {
+            assertThat(mapped.getClass().getMethod(getter.getName()).invoke(mapped)).as(getter.getName()).isEqualTo(getter.invoke(template));
+         }
+      }
+      assertThat(mapped.getCategory()).isEqualTo(EmailCategory.TRANSACTIONAL);
+      assertThat(mapped.getVariablesSchema()).contains("number", "Number", "STRING", "true", "PUBLIC");
+      assertThat(EmailApi.ensureTemplate(template)).isTrue();
+      assertThat(EmailApi.ensureTemplate(template.withBody("Replacement"))).isFalse();
+      assertThat(H2Support.all(EmailTables.TEMPLATE)).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueString("ownerQbit")).isEqualTo("orders");
+         assertThat(row.getValueString("editPermissionName")).isEqualTo("orders.edit");
+         assertThat(row.getValueBoolean("phiApproved")).isFalse();
+      });
+      assertThat(H2Support.all(EmailTables.VERSION)).singleElement().satisfies(row ->
+      {
+         assertThat(row.getValueString("bodySource")).isEqualTo("Hello {{number}}");
+         assertThat(row.getValueString("textSource")).isEqualTo("Text {{number}}");
+         assertThat(row.getValueString("status")).isEqualTo("ACTIVE");
+      });
+      assertThat(EmailApi.send(new EmailApiRequest().withTo("pat@example.com").withTemplateName("notice").withVariable("number", "42")).status())
+         .isEqualTo(EmailApiStatus.QUEUED);
+      assertThat(EmailApi.ensureTemplate(template().withName("raw_schema").withVariablesSchema("[]").withBody("Body"))).isTrue();
+      assertThatThrownBy(() -> EmailApi.ensureTemplate(new EmailApiTemplate())).isInstanceOf(IllegalArgumentException.class);
+      assertThatThrownBy(() -> EmailApi.ensureTemplate(template().withVariable("number", "Number", "STRING", true, "PUBLIC")
+         .withVariablesSchema("[]"))).hasMessageContaining("not both");
+      assertThatThrownBy(() -> EmailApi.ensureTemplate(template().withName("private_notice")
+         .withVariable(new EmailApiVariable("number", "Number", "STRING", true, "PHI"))
+         .withSampleVariables("{\"number\":\"synthetic\"}"))).hasMessageContaining("PHI");
+   }
+
+
+
+   /*******************************************************************************
+    ** Lookups expose only metadata and handle literal wildcard characters.
+    *******************************************************************************/
+   @Test
+   void looksUpMetadataAndLiteralPrefixes() throws Exception
+   {
+      EmailApi.ensureTemplate(template().withBody("Body"));
+      int templateId = H2Support.all(EmailTables.TEMPLATE).getFirst().getValueInteger("id");
+      Instant sent = Instant.parse("2030-01-02T03:04:05Z");
+      Instant delivered = sent.plusSeconds(5);
+      EmailApiResult result = EmailApi.send(request("orders:1").withLinkedRecord("orders", 9));
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", result.messageId()).withValue("templateId", templateId)
+         .withValue("status", "DELIVERED").withValue("providerMessageId", "provider-42").withValue("sentDate", sent).withValue("deliveredDate", delivered));
+      EmailApiMessage message = EmailApi.getMessage(result.messageId()).orElseThrow();
+      assertThat(message.id()).isEqualTo(result.messageId());
+      assertThat(message.status()).isEqualTo("DELIVERED");
+      assertThat(message.category()).isEqualTo(EmailApiCategory.TRANSACTIONAL);
+      assertThat(message.idempotencyKey()).isEqualTo("orders:1");
+      assertThat(message.rfcMessageId()).matches("<[0-9a-f]{32}@example\\.com>");
+      assertThat(message.providerMessageId()).isEqualTo("provider-42");
+      assertThat(message.templateName()).isEqualTo("notice");
+      assertThat(message.linkedTableName()).isEqualTo("orders");
+      assertThat(message.linkedRecordId()).isEqualTo("9");
+      assertThat(message.createDate()).isNotNull();
+      assertThat(message.sentDate()).isEqualTo(sent);
+      assertThat(message.deliveredDate()).isEqualTo(delivered);
+      assertThat(EmailApi.getMessage(null)).isEmpty();
+      assertThat(EmailApi.getMessage(Integer.MAX_VALUE)).isEmpty();
+      assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix("missing", 3)).isEmpty();
+      EmailApi.send(request("orders:2"));
+      assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix("orders:", 1)).containsExactly(message);
+      assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix("orders:", 10)).hasSize(2);
+      for(String prefix : List.of("literal_", "literal%", "literal\\"))
+      {
+         EmailApi.send(request("literalX" + prefix));
+         EmailApiResult matching = EmailApi.send(request(prefix + "match"));
+         assertThat(EmailApi.findMessagesByIdempotencyKeyPrefix(prefix, 1)).extracting(EmailApiMessage::id).containsExactly(matching.messageId());
+      }
+      for(String prefix : new String[] {null, "", " "})
+      {
+         assertThatThrownBy(() -> EmailApi.findMessagesByIdempotencyKeyPrefix(prefix, 1)).isInstanceOf(IllegalArgumentException.class);
+      }
+      assertThatThrownBy(() -> EmailApi.findMessagesByIdempotencyKeyPrefix("orders", 0)).isInstanceOf(IllegalArgumentException.class);
+      assertThatThrownBy(() -> EmailApi.findMessagesByIdempotencyKeyPrefix("orders", -1)).isInstanceOf(IllegalArgumentException.class);
+      assertThat(List.of(EmailApiMessage.class.getRecordComponents())).extracting(component -> component.getName())
+         .containsExactly("id", "status", "category", "idempotencyKey", "rfcMessageId", "providerMessageId", "templateName", "linkedTableName", "linkedRecordId", "createDate", "sentDate", "deliveredDate");
+   }
+
+
+
+   /*******************************************************************************
+    ** Default selection follows configuration and excludes disabled senders.
+    *******************************************************************************/
+   @Test
+   void selectsProviderAndEnabledSenders() throws Exception
+   {
+      assertThat(EmailApi.isInstalled(instance)).isTrue();
+      assertThat(EmailApi.isInstalled(null)).isFalse();
+      assertThat(EmailApi.isInstalled(new QInstance())).isFalse();
+      assertThat(EmailApi.defaultProviderName(instance)).contains("relay");
+      assertThat(EmailApi.defaultProviderName(null)).isEmpty();
+      assertThat(EmailApi.listSenderAddresses("unknown")).isEmpty();
+      assertThat(EmailApi.listSenderAddresses("relay")).isEmpty();
+      QRecord first = SystemWrites.insertOne(EmailTables.SENDER, new QRecord().withValue("providerId", provider.getValue("id"))
+         .withValue("address", "first@example.com").withValue("displayName", "First").withValue("enabled", true).withValue("isDefault", true));
+      SystemWrites.insertOne(EmailTables.SENDER, new QRecord().withValue("providerId", provider.getValue("id"))
+         .withValue("address", "disabled@example.com").withValue("enabled", false));
+      assertThat(EmailApi.listSenderAddresses(null)).containsExactly(new EmailApiSenderAddress(first.getValueInteger("id"), "first@example.com", "First", true));
+      EmailApiSenderAddress address = EmailApi.listSenderAddresses("relay").getFirst();
+      assertThat(address.id()).isEqualTo(first.getValueInteger("id"));
+      assertThat(address.address()).isEqualTo("first@example.com");
+      assertThat(address.displayName()).isEqualTo("First");
+      assertThat(address.isDefault()).isTrue();
+      EmailQBitConfig config = EmailRuntime.config();
+      config.withDefaultProviderName("unknown");
+      assertThat(EmailApi.defaultProviderName(instance)).isEmpty();
+      config.withDefaultProviderName("relay");
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("isDefault", false));
+      assertThat(EmailApi.defaultProviderName(instance)).contains("relay");
+      config.withDefaultProviderName(null);
+      assertThat(EmailApi.defaultProviderName(instance)).isEmpty();
+      config.withDefaultProviderName("relay");
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("enabled", false));
+      assertThat(EmailApi.defaultProviderName(instance)).isEmpty();
+      assertThat(EmailApi.listSenderAddresses("relay")).isEmpty();
+      QInstance other = H2Support.instance(H2Support.backend(), c -> c.withSecretResolver(name -> null));
+      assertThatThrownBy(() -> EmailApi.defaultProviderName(other)).hasMessageContaining("current QContext");
+   }
+
+
+
+   /*******************************************************************************
+    ** Readiness reports each missing prerequisite without leaking resolver data.
+    *******************************************************************************/
+   @Test
+   void reportsReadinessAndPhiPrerequisites() throws Exception
+   {
+      assertThat(EmailApi.readiness(null, null).reasons()).containsExactly("Email qbit is not installed");
+      assertThat(EmailApi.readiness(instance, null).ready()).isTrue();
+      EmailApiReadiness marketing = EmailApi.readiness(instance, EmailApiCategory.MARKETING);
+      assertThat(marketing.ready()).isFalse();
+      assertThat(marketing.reasons()).hasSize(3);
+      assertThat(String.join(";", marketing.reasons())).contains("publicBaseUrl", "QQQ_EMAIL_UNSUBSCRIBE_KEY", "postal address");
+      EmailQBitConfig config = EmailRuntime.config();
+      config.withPublicBaseUrl("https://app.example.com").withSecretResolver(name -> "synthetic-test-key");
+      QRecord brand = SystemWrites.insertOne(EmailTables.BRAND, new QRecord().withValue("name", "default").withValue("label", "Default").withValue("isDefault", true));
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.MARKETING).reasons()).singleElement().asString().contains("postal address");
+      SystemWrites.updateOne(EmailTables.BRAND, new QRecord().withValue("id", brand.getValue("id")).withValue("postalAddress", "1 Main St"));
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.MARKETING).ready()).isTrue();
+      assertThat(EmailApi.send(request("multi").withTo("other@example.com").withCategory(EmailApiCategory.MARKETING)).status()).isEqualTo(EmailApiStatus.REFUSED);
+      assertThat(EmailApi.send(request("marketing").withCategory(EmailApiCategory.MARKETING)).status()).isEqualTo(EmailApiStatus.QUEUED);
+      config.withPublicBaseUrl("javascript:bad");
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.MARKETING).reasons()).containsExactly("MARKETING email needs a valid publicBaseUrl");
+      config.withPublicBaseUrl("https://app.example.com").withSecretResolver(name ->
+      {
+         throw (new IllegalStateException("DO_NOT_EXPOSE_RESOLVER_DETAIL"));
+      });
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.MARKETING).reasons()).containsExactly("MARKETING email needs QQQ_EMAIL_UNSUBSCRIBE_KEY");
+      EmailApiReadiness phi = EmailApi.readiness(instance, EmailApiCategory.TRANSACTIONAL, true);
+      assertThat(phi.reasons()).hasSize(2);
+      assertThat(String.join(";", phi.reasons())).contains("QQQ_EMAIL_BODY_KEY", "PHI-enabled").doesNotContain("DO_NOT_EXPOSE");
+      config.withSecretResolver(name -> " ");
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.SECURITY, true).ready()).isFalse();
+      config.withSecretResolver(name -> "synthetic-test-key");
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id"))
+         .withValue("phiAllowed", true).withValue("trackOpens", false).withValue("trackClicks", false));
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.SECURITY, true).ready()).isTrue();
+      SystemWrites.updateOne(EmailTables.PROVIDER, new QRecord().withValue("id", provider.getValue("id")).withValue("enabled", false));
+      assertThat(EmailApi.readiness(instance, EmailApiCategory.TRANSACTIONAL).reasons()).containsExactly("An enabled named or default provider is required");
+   }
+
+
+
+   /*******************************************************************************
+    ** Small valid ad-hoc request.
+    *******************************************************************************/
+   private EmailApiRequest request(String key)
+   {
+      return (new EmailApiRequest().withTo("pat@example.com").withSubject("Ready").withTextBody("Body").withIdempotencyKey(key));
+   }
+
+
+
+   /*******************************************************************************
+    ** Template body may be replaced by tests using a different schema.
+    *******************************************************************************/
+   private EmailApiTemplate template()
+   {
+      return (new EmailApiTemplate().withName("notice").withLabel("Notice").withCategory(EmailApiCategory.TRANSACTIONAL)
+         .withSubject("Ready").withBody("Hello {{number}}"));
+   }
+}

src/test/java/com/kingsrook/qbits/email/outbox/EmailEffectsTest.java

@@ -0,0 +1,212 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.email.outbox;
+
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.HexFormat;
+import java.util.List;
+import com.kingsrook.qbits.email.H2Support;
+import com.kingsrook.qbits.email.metadata.EmailTables;
+import com.kingsrook.qbits.email.support.SystemWrites;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Durable handoff semantics and stored identity using the real H2 outbox.
+ *******************************************************************************/
+class EmailEffectsTest
+{
+   /*******************************************************************************
+    ** Prepare a provider without opening a transport connection.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      H2Support.setup(config ->
+      {
+      });
+      H2Support.provider(25);
+   }
+
+
+
+   /*******************************************************************************
+    ** Release the per-thread context.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+
+
+
+   /*******************************************************************************
+    ** A plain message uses the configured sender.
+    *******************************************************************************/
+   private EmailRequest request()
+   {
+      return (new EmailRequest().withTo("person@example.com").withSubject("Effect").withTextBody("Body"));
+   }
+
+
+
+   /*******************************************************************************
+    ** Replays return the same persisted reference without another row.
+    *******************************************************************************/
+   @Test
+   void handoffDeduplicatesAndLookupsReturnCurrentReferences() throws Exception
+   {
+      String key = "job:2:row:email";
+      EmailRequest request = request();
+      var first = EmailEffects.send(request, key);
+      var second = EmailEffects.send(request(), key);
+      assertThat(request.getIdempotencyKey()).isEqualTo(key);
+      assertThat(first.outcome()).isEqualTo(EmailEffectOutcome.QUEUED_OR_DUPLICATE);
+      assertThat(second).isEqualTo(first);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+      String token = HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(key.getBytes(StandardCharsets.UTF_8))).substring(0, 32);
+      assertThat(first.message().rfcMessageId()).isEqualTo("<" + token + "@example.com>");
+      assertThat(first.message().status()).isEqualTo("QUEUED");
+      assertThat(first.message().providerMessageId()).isNull();
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", first.message().id())
+         .withValue("status", "SENT").withValue("providerMessageId", "provider-reference"));
+      QContext.init(QContext.getQInstance(), new QSession().withPermissions(List.of()));
+      var found = EmailEffects.findByIdempotencyKey(key).orElseThrow();
+      assertThat(found.id()).isEqualTo(first.message().id());
+      assertThat(found.status()).isEqualTo("SENT");
+      assertThat(found.providerMessageId()).isEqualTo("provider-reference");
+      assertThat(EmailEffects.findByRfcMessageId(first.message().rfcMessageId())).contains(found);
+      assertThat(EmailEffects.findByIdempotencyKey("missing")).isEmpty();
+      assertThat(EmailEffects.findByRfcMessageId("<missing@example.com>")).isEmpty();
+      assertThat(EmailEffects.findByIdempotencyKey(null)).isEmpty();
+      assertThat(EmailEffects.findByRfcMessageId(null)).isEmpty();
+      assertThat(EmailEffects.findByIdempotencyKey(" ")).isEmpty();
+      assertThat(EmailEffects.findByRfcMessageId(" ")).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Conflicting or unusable requests never enqueue or change an existing key.
+    *******************************************************************************/
+   @Test
+   void refusesConflictingKeysAndInvalidRequests() throws Exception
+   {
+      EmailRequest conflict = request().withIdempotencyKey("other");
+      assertThat(EmailEffects.send(conflict, "effect").outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(conflict.getIdempotencyKey()).isEqualTo("other");
+      assertThat(EmailEffects.send(request(), null).outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(EmailEffects.send(request(), " ").outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(EmailEffects.send(request(), "a".repeat(501)).outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(EmailEffects.send(null, "effect").outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      var invalid = EmailEffects.send(request().withSubject(null), "invalid");
+      assertThat(invalid.outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(invalid.message()).isNull();
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+      assertThat(EmailEffects.send(request().withIdempotencyKey("same"), "same").outcome()).isEqualTo(EmailEffectOutcome.QUEUED_OR_DUPLICATE);
+   }
+
+
+
+   /*******************************************************************************
+    ** A suppressed audit row stays refused on replay, including duplicate results.
+    *******************************************************************************/
+   @Test
+   void suppressedRecipientsAreRefusedOnEveryHandoff() throws Exception
+   {
+      H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "person@example.com")
+         .withValue("scope", "ALL").withValue("reason", "HARD_BOUNCE"));
+      var first = EmailEffects.send(request(), "suppressed");
+      var second = EmailEffects.send(request(), "suppressed");
+      assertThat(first.outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+      assertThat(first.message().status()).isEqualTo("SUPPRESSED");
+      assertThat(second).isEqualTo(first);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Never acknowledge an uncommitted or deferred handoff to the jobs dispatcher.
+    *******************************************************************************/
+   @Test
+   void callerTransactionsAreRefused() throws Exception
+   {
+      try(QBackendTransaction transaction = QBackendTransaction.openFor(new InsertInput(EmailTables.MESSAGE)))
+      {
+         assertThat(EmailEffects.send(request().withTransaction(transaction, "rdbms"), "same-backend").outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+         assertThat(EmailEffects.send(request().withTransaction(transaction, "other"), "other-backend").outcome()).isEqualTo(EmailEffectOutcome.REFUSED);
+         transaction.commit();
+      }
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Database uniqueness failures must escape, rather than become terminal refusals.
+    *******************************************************************************/
+   @Test
+   void storageFailurePropagatesAndRfcIdIsUnique() throws Exception
+   {
+      var first = EmailEffects.send(request(), "collision");
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", first.message().id()).withValue("idempotencyKey", "changed"));
+      assertThatThrownBy(() -> EmailEffects.send(request(), "collision")).isInstanceOf(QException.class);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+      assertThat(EmailEffects.findByIdempotencyKey("collision")).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Identity hashing is deterministic, ignores display names and has a safe fallback.
+    *******************************************************************************/
+   @Test
+   void generationAndUnkeyedMessages() throws Exception
+   {
+      assertThat(EmailService.messageId("key", "Display <sender@EXAMPLE.com>"))
+         .isEqualTo(EmailService.messageId("key", "sender@example.com"));
+      for(String from : new String[] {null, "local", "broken <", "sender@[]"})
+      {
+         assertThat(EmailService.messageId("key", from)).matches("<[0-9a-f]{32}@qbit-email\\.local>");
+      }
+      var first = EmailService.send(request());
+      var second = EmailService.send(request());
+      var rows = H2Support.all(EmailTables.MESSAGE);
+      assertThat(rows).hasSize(2);
+      assertThat(rows.get(0).getValueString("rfcMessageId")).matches("<[0-9a-f]{32}@example\\.com>")
+         .isNotEqualTo(rows.get(1).getValueString("rfcMessageId"));
+      assertThat(first.messageId()).isNotEqualTo(second.messageId());
+   }
+}

src/test/java/com/kingsrook/qbits/email/tracking/EmailEventParserTest.java

@@ -26,8 +26,12 @@ import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
-import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
-import static org.assertj.core.api.Assertions.*;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.NOW;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.message;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.provider;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.resource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /*******************************************************************************
  ** Provider-specific normalization contracts use realistic fixture envelopes.

src/test/java/com/kingsrook/qbits/email/tracking/EmailInboxProcessorTest.java

@@ -47,8 +47,12 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
-import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
-import static org.assertj.core.api.Assertions.*;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.NOW;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.SECRET;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.message;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.provider;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.resource;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /*******************************************************************************
  ** Durable processing, rollback, suppression and race acceptance on real H2.
@@ -422,6 +426,7 @@ class EmailInboxProcessorTest
       /*******************************************************************************
        ** Test adapter implementation.
        *******************************************************************************/
+
       @Override
       public void raise(com.kingsrook.qqq.backend.core.alerts.AlertSignal signal)
       {

src/test/java/com/kingsrook/qbits/email/tracking/EmailWebhookRoutesTest.java

@@ -32,7 +32,6 @@ import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import com.kingsrook.qbits.email.EmailRuntime;
@@ -52,8 +51,18 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
-import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
-import static org.assertj.core.api.Assertions.*;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.NOW;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.SECRET;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.TOPIC;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.certificateTime;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.message;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.provider;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.resource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.response;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.signMailgun;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.signSns;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /*******************************************************************************
  ** Real random-port HTTP exercises the exact route provider and database boundary.

src/test/java/com/kingsrook/qbits/email/tracking/TrackingTestSupport.java

@@ -20,6 +20,9 @@
 
 package com.kingsrook.qbits.email.tracking;
 
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.net.ssl.SSLSession;
 import java.net.URI;
 import java.net.http.HttpHeaders;
 import java.net.http.HttpRequest;
@@ -36,9 +39,6 @@ import java.util.Base64;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import javax.crypto.Mac;
-import javax.crypto.spec.SecretKeySpec;
-import javax.net.ssl.SSLSession;
 import com.kingsrook.qbits.email.H2Support;
 import com.kingsrook.qbits.email.metadata.EmailTables;
 import com.kingsrook.qbits.email.support.SystemWrites;

src/test/java/com/kingsrook/qbits/email/tracking/WebhookVerifiersTest.java

@@ -31,8 +31,16 @@ import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
-import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.*;
-import static org.assertj.core.api.Assertions.*;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.NOW;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.SECRET;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.TOPIC;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.certificateTime;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.resource;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.response;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.signMailgun;
+import static com.kingsrook.qbits.email.tracking.TrackingTestSupport.signSns;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /*******************************************************************************
  ** Crypto boundaries use genuine signatures and a public test-only RSA fixture.

src/test/java/com/kingsrook/qbits/email/transport/EmailHttpTransportsTest.java

@@ -21,6 +21,7 @@
 package com.kingsrook.qbits.email.transport;
 
 
+import javax.net.ssl.SSLSession;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -44,7 +45,6 @@ import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
-import javax.net.ssl.SSLSession;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.kingsrook.qbits.email.EmailQBitConfig;
 import com.kingsrook.qbits.email.EmailRuntime;
@@ -147,9 +147,9 @@ class EmailHttpTransportsTest
     *******************************************************************************/
    private static EmailEnvelope envelope()
    {
-      return (new EmailEnvelope().withMessageId("42").withFrom("Sender <sender@example.com>")
+      return (new EmailEnvelope().withMessageId("42").withRfcMessageId("<0123456789abcdef0123456789abcdef@example.com>").withFrom("Sender <sender@example.com>")
          .withTo(List.of("Person <to@example.com>")).withCc(List.of("cc@example.com")).withBcc(List.of("bcc@example.com"))
-         .withReplyTo("Replies <reply@example.com>").withSubject("Hello \u00e9").withTextBody("Plain \u00e9").withHtmlBody("<b>Hello \u00e9</b>")
+         .withReplyTo("Replies <reply@example.com>").withSubject("Hello é").withTextBody("Plain é").withHtmlBody("<b>Hello é</b>")
          .withHeaders(Map.of("List-Unsubscribe", "<https://example.com/unsubscribe>",
             "List-Unsubscribe-Post", "List-Unsubscribe=One-Click", "X-Custom", "kept")));
    }
@@ -283,6 +283,11 @@ class EmailHttpTransportsTest
       assertThat(json.path("MessageStream").asText()).isEqualTo(tracking ? "broadcasts" : "outbound");
       assertThat(json.path("TrackOpens").asBoolean()).isEqualTo(tracking);
       assertThat(json.path("TrackLinks").asText()).isEqualTo(tracking ? "HtmlAndText" : "None");
+      assertThat(json.path("Headers")).anySatisfy(header ->
+      {
+         assertThat(header.path("Name").asText()).isEqualTo("Message-ID");
+         assertThat(header.path("Value").asText()).isEqualTo(envelope().getRfcMessageId());
+      });
       assertThat(json.path("Headers").toString()).contains("List-Unsubscribe", "List-Unsubscribe-Post", "X-QQQ-Email-Id", "42", "Auto-Submitted");
       assertThat(json.toString()).doesNotContain(SECRET);
    }
@@ -303,6 +308,7 @@ class EmailHttpTransportsTest
          return (response(202, "", Map.of("x-message-id", List.of("sg-id"))));
       });
       Map<String, String> headers = new java.util.HashMap<>(envelope().getHeaders());
+      headers.put("message-id", "<override@example.com>");
       for(String reserved : List.of("X-SG-ID", "x-sg-eid", "Received", "DKIM-Signature", "Content-Type",
          "Content-Transfer-Encoding", "To", "From", "Subject", "Reply-To", "CC", "BcC"))
       {
@@ -329,6 +335,8 @@ class EmailHttpTransportsTest
       assertThat(json.at("/content/1/type").asText()).isEqualTo("text/html");
       assertThat(json.at("/content/1/value").asText()).isEqualTo(envelope().getHtmlBody());
       assertThat(json.at("/custom_args/qqq_email_id").asText()).isEqualTo("42");
+      assertThat(json.at("/headers/Message-ID").asText()).isEqualTo(envelope().getRfcMessageId());
+      assertThat(json.path("headers").has("message-id")).isFalse();
       assertThat(json.path("headers").toString()).doesNotContain("must-be-removed").contains("List-Unsubscribe", "List-Unsubscribe-Post");
       assertThat(json.at("/tracking_settings/open_tracking/enable").asBoolean()).isEqualTo(tracking);
       assertThat(json.at("/tracking_settings/click_tracking/enable").asBoolean()).isEqualTo(tracking);
@@ -371,6 +379,8 @@ class EmailHttpTransportsTest
       assertThat(mime).doesNotContain("bcc@example.com", "Bcc:", SECRET);
       MimeMessage message = new MimeMessage(Session.getInstance(new Properties()), new ByteArrayInputStream(mime.getBytes(StandardCharsets.UTF_8)));
       assertThat(message.getSubject()).isEqualTo(envelope().getSubject());
+      assertThat(message.getMessageID()).isEqualTo(envelope().getRfcMessageId());
+      assertThat(multipart).contains("name=\"h:Message-Id\"\r\n\r\n" + envelope().getRfcMessageId() + "\r\n");
       assertThat(message.getHeader("X-QQQ-Email-Id", null)).isEqualTo("42");
       assertThat(message.getHeader("List-Unsubscribe-Post", null)).isEqualTo("List-Unsubscribe=One-Click");
       assertThat(message.getRecipients(jakarta.mail.Message.RecipientType.CC)[0].toString()).isEqualTo("cc@example.com");

src/test/java/com/kingsrook/qbits/email/transport/EmailTransportsTest.java

@@ -47,8 +47,15 @@ import jakarta.mail.internet.InternetAddress;
 import jakarta.mail.internet.MimeMessage;
 import org.junit.jupiter.api.Test;
 import software.amazon.awssdk.services.sesv2.SesV2Client;
-import software.amazon.awssdk.services.sesv2.model.*;
-import static org.assertj.core.api.Assertions.*;
+import software.amazon.awssdk.services.sesv2.model.Message;
+import software.amazon.awssdk.services.sesv2.model.MessageRejectedException;
+import software.amazon.awssdk.services.sesv2.model.SendEmailRequest;
+import software.amazon.awssdk.services.sesv2.model.SendEmailResponse;
+import software.amazon.awssdk.services.sesv2.model.SesV2Exception;
+import software.amazon.awssdk.services.sesv2.model.TooManyRequestsException;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.fail;
 
 /*******************************************************************************
  ** Email Transports Test.
@@ -61,7 +68,7 @@ class EmailTransportsTest
    EmailEnvelope envelope()
    {
       return (new EmailEnvelope()
-            .withMessageId("123")
+            .withMessageId("123").withRfcMessageId("<0123456789abcdef0123456789abcdef@example.com>")
             .withFrom("sender@example.com")
             .withTo(List.of("a@example.com"))
             .withBcc(List.of("hidden@example.com"))
@@ -135,10 +142,26 @@ class EmailTransportsTest
                   Session.getInstance(new Properties()),
                   new ByteArrayInputStream(captured.get().content().raw().data().asByteArray()));
       assertThat(mime.getHeader("Bcc")).isNull();
+      assertThat(mime.getMessageID()).isEqualTo(envelope().getRfcMessageId());
+      assertThat(mime.getHeader("X-QQQ-Email-Id")).containsExactly("123");
       assertThat(mime.getHeader("X-Test")).containsExactly("yes");
       assertThat(mime.getHeader("Auto-Submitted")).containsExactly("auto-generated");
    }
 
+   /*******************************************************************************
+    ** Repeated JavaMail saves cannot replace the identity or accept a custom override.
+    *******************************************************************************/
+   @Test
+   void mimeIdentitySurvivesSaveChanges() throws Exception
+   {
+      var envelope = envelope().withHeaders(Map.of("message-id", "<override@example.com>"));
+      var message = EmailMime.build(Session.getInstance(new Properties()), envelope);
+      message.saveChanges();
+      message.saveChanges();
+      assertThat(message.getHeader("Message-ID")).containsExactly(envelope.getRfcMessageId());
+      assertThat(EmailMime.build(Session.getInstance(new Properties()), envelope().withRfcMessageId(null)).getMessageID()).isNotBlank();
+   }
+
    /*******************************************************************************
     ** Ses Classifies Throttling Rejection And Server Failure.
     *******************************************************************************/
@@ -150,7 +173,7 @@ class EmailTransportsTest
                   TooManyRequestsException.builder().statusCode(429).message("secret").build(),
                   MessageRejectedException.builder().statusCode(400).message("secret").build(),
                   SesV2Exception.builder().statusCode(503).message("secret").build()))
-                  {
+      {
          var transport =
                new SesEmailTransport(
                      builder ->
@@ -313,7 +336,7 @@ class EmailTransportsTest
       for(String newline : List.of("\r", "\n"))
       {
          String bad = "a@example.com" + newline + "Bcc: hidden@example.com";
-         for(EmailEnvelope envelope : List.of(envelope().withSubject(bad), envelope().withFrom(bad), envelope().withReplyTo(bad),
+         for(EmailEnvelope envelope : List.of(envelope().withRfcMessageId(bad), envelope().withSubject(bad), envelope().withFrom(bad), envelope().withReplyTo(bad),
             envelope().withTo(List.of(bad)), envelope().withCc(List.of(bad)), envelope().withBcc(List.of(bad)),
             envelope().withHeaders(Map.of("X-Test", bad)), envelope().withFrom("broken <")))
          {

src/test/resources/email-h2.sql

@@ -38,6 +38,19 @@ CREATE TABLE email_provider (
    UNIQUE (default_slot)
 );
 
+CREATE TABLE email_provider_sender_address (
+   id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+   create_date TIMESTAMP,
+   modify_date TIMESTAMP,
+   provider_id INTEGER NOT NULL,
+   address VARCHAR(500) NOT NULL,
+   display_name VARCHAR(500),
+   is_default BOOLEAN,
+   enabled BOOLEAN,
+   note TEXT,
+   UNIQUE (provider_id, address)
+);
+
 CREATE TABLE email_message (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    create_date TIMESTAMP,
@@ -47,6 +60,7 @@ CREATE TABLE email_message (
    cc_addresses TEXT,
    bcc_addresses TEXT,
    from_address VARCHAR(500),
+   from_address_id INTEGER,
    reply_to VARCHAR(500),
    subject TEXT,
    template_id INTEGER,
@@ -61,6 +75,7 @@ CREATE TABLE email_message (
    phi_approved BOOLEAN,
    unsubscribe_url TEXT,
    idempotency_key VARCHAR(500),
+   rfc_message_id VARCHAR(500),
    provider_id INTEGER,
    provider_message_id VARCHAR(500),
    attempt_count INTEGER,
@@ -79,7 +94,8 @@ CREATE TABLE email_message (
    sender_process VARCHAR(500),
    suppressed_addresses TEXT,
    send_after TIMESTAMP,
-   UNIQUE (idempotency_key)
+   UNIQUE (idempotency_key),
+   UNIQUE (rfc_message_id)
 );
 
 CREATE TABLE email_message_attempt (