CHANGELOG.md

@@ -5,13 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and
 
 ## [Unreleased]
 
+### Changed
+- Built on QQQ 4.1.0-RC.7 (default QQQ profile).
+
 ### Added
+- `EmailJobEffectTarget` automatically registers the durable-jobs `email` target, with idempotent outbox handoff, refusal reasons, reconciliation and job-owner session enforcement; H2 integration coverage and payload/registration documentation.
 - `EmailApiReadiness.failedChecks()` and `failed(check)` with `EmailApiReadinessCheck` (NOT_INSTALLED, PROVIDER_MISSING, PROVIDER_INVALID, PHI_PROVIDER, PUBLIC_BASE_URL, UNSUBSCRIBE_KEY, BRAND_POSTAL_ADDRESS, BODY_KEY), so callers react to a specific prerequisite without parsing reason text.
 - `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.
+- Core-independent `EmailEffects` handoff, reference lookups by idempotency key or RFC Message-ID and refusal mapping; 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`.

README.md

@@ -13,7 +13,7 @@ qbit-object-storage is wired in.
 
 ## Install
 
-Requires Java 21, QQQ **4.1.0-RC.5**, and a transactional RDBMS with the supplied unique
+Requires Java 21, QQQ **4.1.0-RC.7**, and a transactional RDBMS with the supplied unique
 constraints. Apply [the PostgreSQL DDL](docs/ddl/email-postgres.sql) before producing the qbit. SQL timestamps store UTC values without a timezone, matching QQQ RC.5.
 The physical table names equal their metadata names; `backendName` selects the host's
 QQQ database backend, as it does in qbit-webhooks.
@@ -588,23 +588,29 @@ 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.
+Built on QQQ **4.1.0-RC.7**. Enable `DurableJobsMetaData` and define the jobs tables
+**before** producing the email qbit: `EmailQBitProducer` automatically registers
+`EmailJobEffectTarget` under target name `email`. If jobs are enabled later, call
+`EmailJobEffectTarget.register(instance)` afterward. Registration is a no-op when
+jobs are disabled.
 
-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.
+Declare `new JobEffectDeclaration("welcomeEmail", "<step>", "email",
+JobEffectType.IDEMPOTENT_EXTERNAL)` in the process's `DurableProcessMetaData`, then
+call `JobEffects.emit(input, "welcomeEmail", rowKey, payload)` in the backend step.
+The payload accepts recipients, template or ad-hoc bodies, variables, routing and
+record linkage; see the [complete payload contract and examples](docs/DURABLE-JOBS.md).
 
-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.
+The target deduplicates handoffs by the effect key. SENT means the email outbox has
+committed, while its dispatcher owns provider delivery and retries. Job effects
+send under the job owner's current session and captured security scope; an owner
+whose session cannot be established produces FAILED. Other exceptions produce
+UNKNOWN and can safely be retried using the same key. Reconcile finds the existing
+outbox row, including terminal audit rows.
+
+Each message retains its stored RFC Message-ID across provider attempts. SMTP,
+Postmark, SendGrid and Mailgun carry it; SES replaces it downstream. This supports
+correlation and receiver deduplication where available, without guaranteeing
+exactly-once inbox delivery. See the [schema upgrade and provider notes](docs/DURABLE-JOBS.md).
 
 ## License
 

docs/DURABLE-JOBS.md

@@ -1,8 +1,8 @@
 # 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.
+QQQ 4.1.0-RC.7 supplies the durable jobs side-effects outbox. QBit Email registers
+`com.kingsrook.qbits.email.outbox.EmailJobEffectTarget` as target `email` to hand
+those effects to its own committed outbox. All provider sending remains asynchronous.
 
 ## Stored identity and schema upgrade
 
@@ -86,44 +86,101 @@ 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)
+## Register, declare and emit
 
-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.
+Enable durable jobs **before** producing the email qbit, so the producer can register
+its target automatically:
 
 ```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));
-   }
-}
+instance.withSupplementalMetaData(new DurableJobsMetaData());
+new DurableJobsMetaDataProvider().defineAll(instance, "rdbms", null);
+new EmailQBitProducer()
+   .withQBitConfig(new EmailQBitConfig().withBackendName("rdbms"))
+   .produce(instance);
 ```
 
+Apply core's jobs database schema as usual. `EmailJobEffectTarget.register(instance)`
+is a no-op without `DurableJobsMetaData`. If jobs are enabled after producing email,
+call this helper explicitly after enabling them.
+
+Declare the effect on the process's `DurableProcessMetaData` (retain any existing
+durable settings). Replace `<step>` with the process's real backend step name:
+
+```java
+process.withSupplementalMetaData(new DurableProcessMetaData()
+   .withEffect(new JobEffectDeclaration("welcomeEmail", "<step>", "email",
+      JobEffectType.IDEMPOTENT_EXTERNAL)));
+```
+
+Then emit from that step using its `RunBackendStepInput input` and a stable row key:
+
+```java
+Map<String, Serializable> payload = new LinkedHashMap<>();
+payload.put("to", "person@example.com");
+payload.put("subject", "Welcome");
+payload.put("textBody", "Your account is ready.");
+JobEffects.emit(input, "welcomeEmail", rowKey, payload);
+```
+
+The jobs classes above come from `com.kingsrook.qqq.backend.core.jobs` and its
+`effects` package; `Map`, `LinkedHashMap` and `Serializable` are JDK types.
+Core stores the effect with the chunk's progress and dispatches after commit.
+Outside a durable job, `JobEffects.emit` calls the target immediately with a fresh
+key, still only enqueueing email; a result other than SENT throws.
+
+Use **IDEMPOTENT_EXTERNAL**: qbit-email deduplicates by the effect's exact key,
+while its own delivery outbox reuses the stored Message-ID on provider attempts.
+This acknowledges one durable handoff, not exactly-once delivery to a mailbox.
+
+## Payload contract
+
+Unknown keys are ignored. Recognized values must have the following types:
+
+| Keys | Values |
+|---|---|
+| `to`, `cc`, `bcc` | A String mailbox or a list of String mailboxes; omitted means empty. Comma-separated strings are not split. |
+| `subject`, `templateName` | Strings; ordinary template/ad-hoc validation applies. |
+| `markdownBody`, `htmlBody`, `textBody` | String body alternatives. |
+| `variables` | Map with String keys and serializable values; requires a template schema and retains the normal PHI/classification checks. |
+| `category` | String enum name: `TRANSACTIONAL`, `SECURITY`, `NOTIFICATION`, or `MARKETING`. |
+| `providerName`, `brandName`, `locale` | Strings; omitted locale retains `en`. |
+| `from`, `replyTo` | String mailboxes, subject to ordinary sender validation. |
+| `fromAddressId` | Integer identifying an enabled address for the selected provider. |
+| `sendAfter` | ISO-8601 Instant String, for example `2026-10-14T12:00:00Z`. |
+| `linkedTableName`, `linkedRecordId` | String table name; record id converted to String. |
+
+The target always sets `senderQbit = "jobs"`, `senderProcess = effectName`, and
+uses `EffectRequest.idempotencyKey()`. Payload keys cannot override these or attach
+a transaction. `recordVariables`, `securityValues` and `payloadRef` are not decoded;
+provide the supported values directly in `payload`.
+
+## Identity and outcomes
+
+The dispatcher invokes the target as the system user. When `jobId` is present, the
+target loads the job under core's system scope and builds its owner session through
+`JobSessions.forJob`. Sending runs inside `QContext.withTemporaryContext` using that
+owner session, with current permissions and the intersection of current and captured
+security keys. The original dispatcher context is restored even after an exception.
+Without a job id, sending retains the caller's session.
+
+- `QUEUED_OR_DUPLICATE` becomes `EffectResult.sent(outboxId)`; `providerRef` is the
+  String email row id, not the transport provider's reference.
+- `REFUSED` becomes `EffectResult.failed(reasons)`, including suppressed replays.
+- A missing job or an owner session rejected by authentication/permission validation
+  becomes FAILED with `The job's owner can no longer send it: ...`.
+- Other exceptions, including payload decoding and database failures, become UNKNOWN.
+  Error text omits raw exception details to avoid exposing payloads or database values.
+  Core may resend IDEMPOTENT_EXTERNAL effects using the same key.
+- Reconcile uses `EmailEffects.findByIdempotencyKey`: any existing row becomes FOUND
+  with its id, otherwise NOT_FOUND. Lookup failures propagate rather than falsely
+  reporting absence. Reconcile never creates mail or changes the owner session.
+
+The binding uses `EmailEffects` internally; other consuming qbits continue to use
+`EmailApi`. The H2 target tests install core jobs metadata with `defineAll`, create
+all columns of the `jobs_job` lookup table, and observe the owner through an insert
+customizer because email messages have no built-in created-by/user field. No
+production job-lookup test seam is needed.
+
 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.

pom.xml

@@ -241,7 +241,7 @@
             <activeByDefault>true</activeByDefault>
          </activation>
          <properties>
-            <qqq.snapshot.version>4.1.0-RC.5</qqq.snapshot.version>
+            <qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
          </properties>
          <dependencyManagement>
             <dependencies>

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

@@ -197,6 +197,7 @@ public class EmailQBitProducer implements QBitMetaDataProducer<EmailQBitConfig>
       output.add(new EmailOutboxMessagingProviderMetaData().withName(qBitConfig.getBridgeProviderName()));
       output.addSelfToInstance(qInstance);
       registerWebhooks(qInstance);
+      com.kingsrook.qbits.email.outbox.EmailJobEffectTarget.register(qInstance);
       return (new MetaDataProducerMultiOutput());
    }
 

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

@@ -42,8 +42,15 @@ public final class EmailEffects
    /*******************************************************************************
     ** A refusal can have a reference when suppression retained an audit row.
     *******************************************************************************/
-   public record Result(EmailEffectOutcome outcome, EmailMessageRef message)
+   public record Result(EmailEffectOutcome outcome, EmailMessageRef message, List<String> reasons)
    {
+      /*******************************************************************************
+       ** Preserve the original constructor for callers that only need the reference.
+       *******************************************************************************/
+      public Result(EmailEffectOutcome outcome, EmailMessageRef message)
+      {
+         this(outcome, message, List.of());
+      }
    }
 
 
@@ -68,14 +75,15 @@ public final class EmailEffects
          || request.getTransaction() != null
          || request.getIdempotencyKey() != null && !request.getIdempotencyKey().equals(effectIdempotencyKey))
       {
-         return (new Result(EmailEffectOutcome.REFUSED, null));
+         return (new Result(EmailEffectOutcome.REFUSED, null, List.of("An email request without a caller transaction and a valid, matching effect key are required")));
       }
       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));
+      List<String> reasons = message != null && "SUPPRESSED".equals(message.status()) ? List.of("All recipients are suppressed") : result.reasons();
+      return (new Result(refused ? EmailEffectOutcome.REFUSED : EmailEffectOutcome.QUEUED_OR_DUPLICATE, message, reasons));
    }
 
 

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

@@ -0,0 +1,200 @@
+/*
+ * 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.io.Serializable;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import com.kingsrook.qqq.backend.core.context.CapturedContext;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
+import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData;
+import com.kingsrook.qqq.backend.core.jobs.JobSessions;
+import com.kingsrook.qqq.backend.core.jobs.effects.EffectRequest;
+import com.kingsrook.qqq.backend.core.jobs.effects.EffectResult;
+import com.kingsrook.qqq.backend.core.jobs.effects.EffectTarget;
+import com.kingsrook.qqq.backend.core.jobs.effects.ReconcileResult;
+import com.kingsrook.qqq.backend.core.jobs.store.DurableJobStore;
+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.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+
+
+/*******************************************************************************
+ ** Durable jobs target for IDEMPOTENT_EXTERNAL email effects. SENT acknowledges
+ ** the committed email outbox, whose key deduplication makes retries safe. The
+ ** email dispatcher owns provider delivery and preserves the stored Message-ID.
+ ** Payload keys and types are documented in docs/DURABLE-JOBS.md.
+ *******************************************************************************/
+public class EmailJobEffectTarget implements EffectTarget
+{
+   public static final String TARGET_NAME = "email";
+
+
+
+   /*******************************************************************************
+    ** Register after enabling durable jobs; leave instances without jobs alone.
+    *******************************************************************************/
+   public static void register(QInstance instance)
+   {
+      DurableJobsMetaData jobs = DurableJobsMetaData.of(instance);
+      if(jobs != null)
+      {
+         jobs.withEffectTarget(TARGET_NAME, new QCodeReference(EmailJobEffectTarget.class));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Enqueue under the job owner's current permissions and captured scope.
+    ** Other exceptions are uncertain; core may safely resend using the same key.
+    *******************************************************************************/
+   @Override
+   public EffectResult deliver(EffectRequest r)
+   {
+      try
+      {
+         EmailRequest request = decode(r.payload()).withSender("jobs", r.effectName());
+         EmailEffects.Result result;
+         if(r.jobId() == null)
+         {
+            result = EmailEffects.send(request, r.idempotencyKey());
+         }
+         else
+         {
+            QInstance instance = QContext.getQInstance();
+            QSession owner;
+            try
+            {
+               QRecord job = new DurableJobStore(DurableJobsMetaData.of(instance)).getJob(r.jobId());
+               if(job == null)
+               {
+                  return (EffectResult.failed("The job's owner can no longer send it: job not found"));
+               }
+               owner = JobSessions.forJob(instance, job);
+            }
+            catch(QPermissionDeniedException | QAuthenticationException e)
+            {
+               return (EffectResult.failed("The job's owner can no longer send it: owner session unavailable"));
+            }
+            AtomicReference<EmailEffects.Result> handoff = new AtomicReference<>();
+            QContext.withTemporaryContext(new CapturedContext(instance, owner),
+               () -> handoff.set(EmailEffects.send(request, r.idempotencyKey())));
+            result = handoff.get();
+         }
+         if(result.outcome() == EmailEffectOutcome.REFUSED)
+         {
+            return (EffectResult.failed(String.join("; ", result.reasons())));
+         }
+         return (EffectResult.sent(String.valueOf(result.message().id())));
+      }
+      catch(Exception e)
+      {
+         // Exception details can contain payloads or database values.
+         return (EffectResult.unknown("Email handoff could not be confirmed (" + e.getClass().getSimpleName() + ")"));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Reconcile handoff by existence, including suppressed and terminal audit rows.
+    *******************************************************************************/
+   @Override
+   public ReconcileResult reconcile(EffectRequest r) throws QException
+   {
+      return (EmailEffects.findByIdempotencyKey(r.idempotencyKey())
+         .map(message -> ReconcileResult.found(String.valueOf(message.id())))
+         .orElseGet(ReconcileResult::notFound));
+   }
+
+
+
+   /*******************************************************************************
+    ** Decode only supported values; identity, sender and transaction cannot be
+    ** overridden by payload keys. Variables retain normal template/PHI validation.
+    *******************************************************************************/
+   static EmailRequest decode(Map<String, Serializable> payload)
+   {
+      Map<String, Serializable> values = payload == null ? Map.of() : payload;
+      EmailRequest request = new EmailRequest()
+         .withTo(addresses(values.get("to")))
+         .withCc(addresses(values.get("cc")))
+         .withBcc(addresses(values.get("bcc")))
+         .withFrom((String) values.get("from"))
+         .withFromAddressId((Integer) values.get("fromAddressId"))
+         .withReplyTo((String) values.get("replyTo"))
+         .withSubject((String) values.get("subject"))
+         .withTemplateName((String) values.get("templateName"))
+         .withMarkdownBody((String) values.get("markdownBody"))
+         .withHtmlBody((String) values.get("htmlBody"))
+         .withTextBody((String) values.get("textBody"))
+         .withProviderName((String) values.get("providerName"))
+         .withBrandName((String) values.get("brandName"))
+         .withLinkedRecord((String) values.get("linkedTableName"), values.get("linkedRecordId"));
+      if(values.get("locale") != null)
+      {
+         request.withLocale((String) values.get("locale"));
+      }
+      if(values.get("category") != null)
+      {
+         request.withCategory(EmailCategory.valueOf((String) values.get("category")));
+      }
+      if(values.get("sendAfter") != null)
+      {
+         request.withSendAfter(Instant.parse((String) values.get("sendAfter")));
+      }
+      if(values.get("variables") instanceof Map<?, ?> variables)
+      {
+         variables.forEach((key, value) -> request.withVariable((String) key, (Serializable) value));
+      }
+      else if(values.get("variables") != null)
+      {
+         throw (new IllegalArgumentException("variables must be a map"));
+      }
+      return (request);
+   }
+
+
+
+   /*******************************************************************************
+    ** Accept one mailbox or a list of mailboxes, without comma splitting.
+    *******************************************************************************/
+   private static List<String> addresses(Object value)
+   {
+      if(value == null)
+      {
+         return (List.of());
+      }
+      if(value instanceof String address)
+      {
+         return (List.of(address));
+      }
+      return (((List<?>) value).stream().map(String.class::cast).toList());
+   }
+}

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

@@ -0,0 +1,494 @@
+/*
+ * 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.io.Serializable;
+import java.sql.Connection;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import com.kingsrook.qbits.email.EmailQBitConfig;
+import com.kingsrook.qbits.email.EmailQBitProducer;
+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.customizers.TableCustomizerInterface;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
+import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaData;
+import com.kingsrook.qqq.backend.core.jobs.DurableJobsMetaDataProvider;
+import com.kingsrook.qqq.backend.core.jobs.DurableJobsTables;
+import com.kingsrook.qqq.backend.core.jobs.effects.EffectRequest;
+import com.kingsrook.qqq.backend.core.jobs.effects.EffectResult;
+import com.kingsrook.qqq.backend.core.jobs.effects.JobEffectType;
+import com.kingsrook.qqq.backend.core.jobs.effects.ReconcileResult;
+import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
+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.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
+import com.kingsrook.qqq.backend.core.model.session.QUser;
+import com.kingsrook.qqq.backend.core.modules.authentication.implementations.FullyAnonymousAuthenticationModule;
+import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager;
+import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData;
+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;
+
+
+/*******************************************************************************
+ ** Real H2 outbox and jobs lookup; no transport or mocked owner lookup.
+ *******************************************************************************/
+class EmailJobEffectTargetTest
+{
+   private final EmailJobEffectTarget target = new EmailJobEffectTarget();
+   private static QSession observedSession;
+
+
+
+   /*******************************************************************************
+    ** Install a local outbox and provider without dispatching any messages.
+    *******************************************************************************/
+   @BeforeEach
+   void setup() throws Exception
+   {
+      H2Support.setup(config ->
+      {
+      });
+      H2Support.provider(25);
+      observedSession = null;
+   }
+
+
+
+   /*******************************************************************************
+    ** Restore per-test context and hook state.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+      observedSession = null;
+   }
+
+
+
+   /*******************************************************************************
+    ** A normal effect contains only serializable payload values.
+    *******************************************************************************/
+   private EffectRequest request(String key, Integer jobId, Map<String, Serializable> payload)
+   {
+      return (new EffectRequest(key, "welcomeEmail", JobEffectType.IDEMPOTENT_EXTERNAL, jobId, 1, "row", payload, null, 1));
+   }
+
+
+
+   /*******************************************************************************
+    ** Valid ad-hoc email payload.
+    *******************************************************************************/
+   private Map<String, Serializable> payload()
+   {
+      return (new LinkedHashMap<>(Map.of("to", "person@example.com", "subject", "Welcome", "textBody", "Hello")));
+   }
+
+
+
+   /*******************************************************************************
+    ** Replays acknowledge the same committed row and ignore payload identity keys.
+    *******************************************************************************/
+   @Test
+   void deliversOnceAndReconciles() throws Exception
+   {
+      var payload = payload();
+      payload.put("senderQbit", "spoofed");
+      payload.put("senderProcess", "spoofed");
+      payload.put("idempotencyKey", "spoofed");
+      payload.put("transaction", "ignored");
+      var request = request("same", null, payload);
+      assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.notFound());
+      var first = target.deliver(request);
+      assertThat(first.outcome()).isEqualTo(EffectResult.Outcome.SENT);
+      assertThat(target.deliver(request)).isEqualTo(first);
+      var rows = H2Support.all(EmailTables.MESSAGE);
+      assertThat(rows).hasSize(1);
+      var row = rows.getFirst();
+      assertThat(first.providerRef()).isEqualTo(row.getValueString("id"));
+      assertThat(row.getValueString("senderQbit")).isEqualTo("jobs");
+      assertThat(row.getValueString("senderProcess")).isEqualTo("welcomeEmail");
+      assertThat(row.getValueString("idempotencyKey")).isEqualTo("same");
+      assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.found(first.providerRef()));
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", row.getValueInteger("id")).withValue("status", "DEAD_LETTER"));
+      assertThat(target.reconcile(request)).isEqualTo(ReconcileResult.found(first.providerRef()));
+   }
+
+
+
+   /*******************************************************************************
+    ** Suppression is a permanent refusal on every attempt, but reconciles by row.
+    *******************************************************************************/
+   @Test
+   void suppressedRecipientFails() throws Exception
+   {
+      H2Support.insert(EmailTables.SUPPRESSION, new QRecord().withValue("address", "person@example.com")
+         .withValue("scope", "ALL").withValue("reason", "HARD_BOUNCE"));
+      var request = request("suppressed", null, payload());
+      assertThat(target.deliver(request)).isEqualTo(EffectResult.failed("All recipients are suppressed"));
+      assertThat(target.deliver(request).outcome()).isEqualTo(EffectResult.Outcome.FAILED);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+      assertThat(target.reconcile(request).outcome()).isEqualTo(ReconcileResult.Outcome.FOUND);
+   }
+
+
+
+   /*******************************************************************************
+    ** Service validation reasons reach the effect without bypassing template rules.
+    *******************************************************************************/
+   @Test
+   void validationRefusalRetainsReasons() throws Exception
+   {
+      var payload = payload();
+      payload.put("variables", new LinkedHashMap<>(Map.of("name", "Person")));
+      assertThat(target.deliver(request("variables", null, payload)))
+         .isEqualTo(EffectResult.failed("Variables require a template with a schema"));
+      assertThat(target.deliver(request("empty", null, null)).outcome()).isEqualTo(EffectResult.Outcome.FAILED);
+      assertThat(target.deliver(request(" ", null, payload())).message()).contains("valid, matching effect key");
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** A database failure remains uncertain and does not disclose raw diagnostics.
+    *******************************************************************************/
+   @Test
+   void storageFailureIsUnknown() throws Exception
+   {
+      var request = request("collision", null, payload());
+      var sent = target.deliver(request);
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", Integer.valueOf(sent.providerRef())).withValue("idempotencyKey", "changed"));
+      var result = target.deliver(request);
+      assertThat(result.outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN);
+      assertThat(result.message()).doesNotContain("person@example.com", "INSERT", "Hello");
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Malformed supported values follow the specified safe same-key retry policy.
+    *******************************************************************************/
+   @Test
+   void decodingExceptionsAreUnknown() throws Exception
+   {
+      for(var invalid : List.of(Map.entry("to", 4), Map.entry("category", "invalid"), Map.entry("variables", "invalid")))
+      {
+         var payload = payload();
+         payload.put(invalid.getKey(), (Serializable) invalid.getValue());
+         assertThat(target.deliver(request("bad", null, payload)).outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN);
+      }
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Every documented optional value reaches the request without losing types.
+    *******************************************************************************/
+   @Test
+   void decodesSupportedPayload()
+   {
+      var payload = payload();
+      payload.put("to", new ArrayList<>(List.of("one@example.com", "two@example.com")));
+      payload.put("cc", "cc@example.com");
+      payload.put("bcc", new ArrayList<>(List.of("bcc@example.com")));
+      for(String key : List.of("from", "replyTo", "templateName", "markdownBody", "htmlBody", "providerName", "brandName", "linkedTableName"))
+      {
+         payload.put(key, key);
+      }
+      payload.put("fromAddressId", 7);
+      payload.put("locale", "fr");
+      payload.put("category", "SECURITY");
+      payload.put("sendAfter", "2026-10-14T12:00:00Z");
+      payload.put("linkedRecordId", 42);
+      payload.put("variables", new LinkedHashMap<>(Map.of("count", 3, "active", true)));
+      var decoded = EmailJobEffectTarget.decode(payload);
+      assertThat(decoded.getTo()).containsExactly("one@example.com", "two@example.com");
+      assertThat(decoded.getCc()).containsExactly("cc@example.com");
+      assertThat(decoded.getBcc()).containsExactly("bcc@example.com");
+      assertThat(decoded.getFrom()).isEqualTo("from");
+      assertThat(decoded.getFromAddressId()).isEqualTo(7);
+      assertThat(decoded.getReplyTo()).isEqualTo("replyTo");
+      assertThat(decoded.getSubject()).isEqualTo("Welcome");
+      assertThat(decoded.getTemplateName()).isEqualTo("templateName");
+      assertThat(decoded.getMarkdownBody()).isEqualTo("markdownBody");
+      assertThat(decoded.getHtmlBody()).isEqualTo("htmlBody");
+      assertThat(decoded.getTextBody()).isEqualTo("Hello");
+      assertThat(decoded.getProviderName()).isEqualTo("providerName");
+      assertThat(decoded.getBrandName()).isEqualTo("brandName");
+      assertThat(decoded.getLocale()).isEqualTo("fr");
+      assertThat(decoded.getCategory()).isEqualTo(EmailCategory.SECURITY);
+      assertThat(decoded.getSendAfter()).isEqualTo(Instant.parse("2026-10-14T12:00:00Z"));
+      assertThat(decoded.getLinkedTableName()).isEqualTo("linkedTableName");
+      assertThat(decoded.getLinkedRecordId()).isEqualTo("42");
+      assertThat(decoded.getVariables()).containsEntry("count", 3).containsEntry("active", true);
+      assertThat(EmailJobEffectTarget.decode(Map.of()).getLocale()).isEqualTo("en");
+   }
+
+
+
+   /*******************************************************************************
+    ** Use core's jobs metadata and a real H2 jobs_job table with all its columns.
+    ** Only the lookup table needs physical storage; no jobs worker runs here.
+    *******************************************************************************/
+   private void enableJobs() throws Exception
+   {
+      QInstance instance = QContext.getQInstance();
+      instance.withSupplementalMetaData(new DurableJobsMetaData());
+      new DurableJobsMetaDataProvider().defineAll(instance, "rdbms", null);
+      instance.getAuthentication().withModuleCodeReference(new QCodeReference(OwnerAuthentication.class));
+      QTableMetaData table = instance.getTable(DurableJobsTables.TABLE_JOB);
+      List<String> columns = new ArrayList<>();
+      for(var field : table.getFields().values())
+      {
+         String sqlType = switch(field.getType())
+         {
+            case INTEGER -> "INTEGER";
+            case BOOLEAN -> "BOOLEAN";
+            case DATE_TIME -> "TIMESTAMP";
+            default -> "VARCHAR(10000)";
+         };
+         columns.add("\"" + field.getName() + "\" " + sqlType);
+      }
+      try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) instance.getBackend("rdbms")))
+      {
+         connection.createStatement().execute("CREATE TABLE jobs_job (" + String.join(", ", columns) + ")");
+      }
+      instance.getTable(EmailTables.MESSAGE).withCustomizer(TableCustomizers.PRE_INSERT_RECORD, new QCodeReference(ObserveOwner.class));
+      QContext.init(instance, new QSystemUserSession());
+   }
+
+
+
+   /*******************************************************************************
+    ** Seed just the job identity and captured scope through JDBC, like a stored job.
+    *******************************************************************************/
+   private void job(String userId) throws Exception
+   {
+      try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) QContext.getQInstance().getBackend("rdbms"));
+         var statement = connection.prepareStatement("INSERT INTO jobs_job (\"id\", \"userId\", \"credentialScopeJson\", \"clientChannel\") VALUES (1, ?, ?, ?)"))
+      {
+         statement.setString(1, userId);
+         statement.setString(2, "{\"tenant\":[\"allowed\"]}");
+         statement.setString(3, "assistant");
+         statement.executeUpdate();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The actual enqueue observes the owner and narrowed scope, then restores system.
+    *******************************************************************************/
+   @Test
+   void sendsAsJobOwner() throws Exception
+   {
+      enableJobs();
+      job("owner");
+      QSession dispatcher = QContext.getQSession();
+      assertThat(target.deliver(request("owner", 1, payload())).outcome()).isEqualTo(EffectResult.Outcome.SENT);
+      assertThat(observedSession.getUser().getIdReference()).isEqualTo("owner");
+      assertThat(observedSession.getClientChannel()).isEqualTo("assistant");
+      assertThat(observedSession.getSecurityKeyValues().get("tenant")).containsExactly("allowed");
+      assertThat(QContext.getQSession()).isSameAs(dispatcher);
+      assertThat(H2Support.all(EmailTables.MESSAGE)).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** A missing owner fails permanently instead of sending as the system user.
+    *******************************************************************************/
+   @Test
+   void missingOwnerFails() throws Exception
+   {
+      enableJobs();
+      job(null);
+      assertOwnerFailure();
+   }
+
+
+
+   /*******************************************************************************
+    ** Authentication refusal fails permanently.
+    *******************************************************************************/
+   @Test
+   void authenticationFailureFails() throws Exception
+   {
+      enableJobs();
+      job("denied");
+      assertOwnerFailure();
+   }
+
+
+
+   /*******************************************************************************
+    ** Core rejects a session for someone other than the persisted job owner.
+    *******************************************************************************/
+   @Test
+   void wrongOwnerFails() throws Exception
+   {
+      enableJobs();
+      job("wrong");
+      assertOwnerFailure();
+   }
+
+
+
+   /*******************************************************************************
+    ** A deleted job cannot fall back to the dispatcher identity.
+    *******************************************************************************/
+   @Test
+   void missingJobFails() throws Exception
+   {
+      enableJobs();
+      assertOwnerFailure();
+   }
+
+
+
+   /*******************************************************************************
+    ** Refusal happens before the outbox sees the request and preserves the context.
+    *******************************************************************************/
+   private void assertOwnerFailure() throws Exception
+   {
+      QSession dispatcher = QContext.getQSession();
+      var result = target.deliver(request("denied", 1, payload()));
+      assertThat(result.outcome()).isEqualTo(EffectResult.Outcome.FAILED);
+      assertThat(result.message()).startsWith("The job's owner can no longer send it:");
+      assertThat(QContext.getQSession()).isSameAs(dispatcher);
+      assertThat(observedSession).isNull();
+      assertThat(H2Support.all(EmailTables.MESSAGE)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** A failed send also restores the system session after using the owner context.
+    *******************************************************************************/
+   @Test
+   void ownerContextRestoredAfterException() throws Exception
+   {
+      enableJobs();
+      job("owner");
+      var request = request("collision", 1, payload());
+      var sent = target.deliver(request);
+      SystemWrites.updateOne(EmailTables.MESSAGE, new QRecord().withValue("id", Integer.valueOf(sent.providerRef())).withValue("idempotencyKey", "changed"));
+      QSession dispatcher = QContext.getQSession();
+      assertThat(target.deliver(request).outcome()).isEqualTo(EffectResult.Outcome.UNKNOWN);
+      assertThat(QContext.getQSession()).isSameAs(dispatcher);
+      assertThat(observedSession.getUser().getIdReference()).isEqualTo("owner");
+   }
+
+
+
+   /*******************************************************************************
+    ** A failed lookup must never be interpreted as proof of no previous handoff.
+    *******************************************************************************/
+   @Test
+   void reconcileStorageFailurePropagates() throws Exception
+   {
+      try(Connection connection = new ConnectionManager().getConnection((RDBMSBackendMetaData) QContext.getQInstance().getBackend("rdbms")))
+      {
+         connection.createStatement().execute("ALTER TABLE email_message RENAME TO unavailable_message");
+      }
+      assertThatThrownBy(() -> target.reconcile(request("missing", null, payload()))).isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class);
+   }
+
+
+
+   /*******************************************************************************
+    ** The helper is optional, and producing email after jobs registers automatically.
+    *******************************************************************************/
+   @Test
+   void registersOnlyWhenEnabled() throws Exception
+   {
+      QInstance disabled = QContext.getQInstance();
+      EmailJobEffectTarget.register(disabled);
+      assertThat(DurableJobsMetaData.of(disabled)).isNull();
+      QInstance enabled = new QInstance().withSupplementalMetaData(new DurableJobsMetaData());
+      enabled.addBackend(H2Support.backend());
+      new DurableJobsMetaDataProvider().defineAll(enabled, "rdbms", null);
+      new EmailQBitProducer().withQBitConfig(new EmailQBitConfig().withBackendName("rdbms")).produce(enabled);
+      assertThat(DurableJobsMetaData.of(enabled).getEffectTargets()).containsOnlyKeys("email");
+      assertThat(DurableJobsMetaData.of(enabled).getEffectTargets().get("email").getName()).isEqualTo(EmailJobEffectTarget.class.getName());
+      EmailJobEffectTarget.register(enabled);
+      assertThat(DurableJobsMetaData.of(enabled).getEffectTargets()).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test authentication supplies current permissions and a wider live scope.
+    *******************************************************************************/
+   public static class OwnerAuthentication extends FullyAnonymousAuthenticationModule
+   {
+      /*******************************************************************************
+       ** Simulate an active owner, revoked authentication, or mismatched identity.
+       *******************************************************************************/
+      @Override
+      public QSession createAutomatedSessionForUser(QInstance instance, Serializable userId) throws QAuthenticationException
+      {
+         if("denied".equals(userId))
+         {
+            throw (new QAuthenticationException("Access revoked"));
+         }
+         return (H2Support.admin().withUser(new QUser().withIdReference("wrong".equals(userId) ? "someone-else" : String.valueOf(userId)))
+            .withSecurityKeyValues(Map.of("tenant", List.of("allowed", "new"))));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Observe the session inside the real send without adding production test seams.
+    *******************************************************************************/
+   public static class ObserveOwner implements TableCustomizerInterface
+   {
+      /*******************************************************************************
+       ** Email messages have no built-in user field, so observe at the insert hook.
+       *******************************************************************************/
+      @Override
+      public List<QRecord> preInsertOrUpdate(AbstractActionInput input, List<QRecord> records, boolean preview, Optional<List<QRecord>> oldRecords)
+      {
+         observedSession = QContext.getQSession();
+         return (records);
+      }
+   }
+}