fix(testutil): derive a per-workspace database for non-.test binaries

AGENTS.md

@@ -408,6 +408,19 @@ manually on every API change even though the template won't remind you.
   bundled drive-by cleanup. CI must be green.
 - **Coverage floors** only move up (see Testing strategy).
 - **Postgres**: local dev runs on port **5433** (not 5432) via docker compose.
+- **Row locks in multi-statement transactions**: an `INSERT` holds
+  `FOR KEY SHARE` on every row it references by foreign key until commit, and
+  `FOR UPDATE` conflicts with that. So a `SELECT … FOR UPDATE` on a parent row
+  taken *after* inserting a child in the same transaction deadlocks against a
+  concurrent insert for the same parent (v1.9.0: the accept transaction
+  inserted the message, then the gate locked the agent `FOR UPDATE`; two
+  parallel sends → SQLSTATE 40P01). Lock the parent `FOR NO KEY UPDATE`
+  (excludes updates, deletes and other lockers, coexists with key shares), or
+  lock it before the insert. The accept transaction's full lock order is in
+  `docs/design/async-message-pipeline.md`; any new lock on that path must be
+  checked against it, and any parallel-write path needs a concurrency test
+  (see `TestPrepareDoesNotDeadlockAgainstConcurrentInsert` for the
+  deterministic two-transaction shape).
 - The Mailpit service in `docker-compose.yaml` is local-dev only — production
   deployments must drop it and point `E2A_OUTBOUND_SMTP_*` at a real relay.
 

api/openapi.yaml

@@ -1984,7 +1984,7 @@ components:
       additionalProperties: true
       properties:
         code:
-          description: "Machine-branchable error code — the stable discriminator clients switch on. Open set: treat it as a string and tolerate unknown values, since new codes may be added over time (branch on the ones you handle, fall back to the HTTP status otherwise). Exact current vocabulary (machine-checked): unauthorized, forbidden, blocked_by_policy, invalid_request, invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope, reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty, recipient_suppressed, not_found, attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found, gone, conflict, precondition_failed, agent_taken, domain_taken, alias_taken, address_in_trash, message_held, message_not_pending, message_not_yet_delivered, not_in_trash, purge_in_progress, send_in_progress, webhook_disabled, webhook_cooldown, domain_not_registered, domain_has_agents, domain_not_verified, inbound_mx_missing, limit_exceeded, rate_limited, contact_limit_reached, template_limit_reached, webhook_limit_reached, idempotency_in_flight, idempotency_key_reuse, payload_too_large, attachment_too_large, not_implemented, events_log_disabled, limits_unavailable, inbound_mx_check_failed, auth_unavailable, internal_error, method_not_allowed, unsupported_media_type, error. Grouped semantics: auth: unauthorized (401), forbidden (403), blocked_by_policy (403, outbound policy gate; experimental). Validation: invalid_request is the single canonical code for input-validation failures whether they arrive as 400 (malformed) or 422 (semantically invalid); field/resource-specific invalid_* refinements (invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope), reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty (all 400); recipient_suppressed (422). Not found: not_found (404) plus the *_not_found family (attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found); gone (410, past retention). Conflict/state: conflict (409, generic), precondition_failed (412, optimistic-concurrency validator is stale), the *_taken family — the requested identifier is already claimed — (agent_taken, domain_taken, alias_taken, all 409), address_in_trash (409), message_held (409), message_not_pending (409), message_not_yet_delivered (409, retry after the source outbound message is sent), not_in_trash (409), purge_in_progress (409, permanent delete already claimed), send_in_progress (409), webhook_disabled (409), webhook_cooldown (409), domain_not_registered (400), domain_has_agents (400), domain_not_verified (400 on create-agent, 403 on send), inbound_mx_missing (400). Capacity: limit_exceeded (402, plan quota — see LimitExceededDetails), rate_limited (429, request rate — see RateLimitedDetails), contact_limit_reached, template_limit_reached and webhook_limit_reached (400, fixed per-account caps). Idempotency: idempotency_in_flight (409, wait then retry the byte-identical request), idempotency_key_reuse (422, caller bug — do not retry as-is). Size: payload_too_large (413, request body), attachment_too_large (413, inline fetch over the cap — use download_url). Availability: not_implemented (501, feature not available on this deployment), events_log_disabled (501), limits_unavailable (503), inbound_mx_check_failed (503), auth_unavailable (503, an auth backend — e.g. a delegated-token verifier or the identity store — could not judge the credential; retry). Server/fallback: internal_error (5xx), method_not_allowed (405), unsupported_media_type (415), and the generic code error for any otherwise-unmapped status."
+          description: "Machine-branchable error code — the stable discriminator clients switch on. Open set: treat it as a string and tolerate unknown values, since new codes may be added over time (branch on the ones you handle, fall back to the HTTP status otherwise). Exact current vocabulary (machine-checked): unauthorized, forbidden, blocked_by_policy, sending_paused, invalid_request, invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope, reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty, recipient_suppressed, not_found, attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found, gone, conflict, precondition_failed, agent_taken, domain_taken, alias_taken, address_in_trash, message_held, message_not_pending, message_not_yet_delivered, not_in_trash, purge_in_progress, send_in_progress, webhook_disabled, webhook_cooldown, domain_not_registered, domain_has_agents, domain_not_verified, inbound_mx_missing, limit_exceeded, rate_limited, contact_limit_reached, template_limit_reached, webhook_limit_reached, idempotency_in_flight, idempotency_key_reuse, payload_too_large, attachment_too_large, not_implemented, events_log_disabled, limits_unavailable, inbound_mx_check_failed, auth_unavailable, internal_error, method_not_allowed, unsupported_media_type, error. Grouped semantics: auth: unauthorized (401), forbidden (403), blocked_by_policy (403, outbound policy gate; experimental), sending_paused (403, outbound sending is paused for the account by the platform abuse controls; queued mail is held, new sends are refused until an operator resumes; experimental). Validation: invalid_request is the single canonical code for input-validation failures whether they arrive as 400 (malformed) or 422 (semantically invalid); field/resource-specific invalid_* refinements (invalid_cursor, invalid_filter, invalid_domain, invalid_slug, invalid_recipient, invalid_attachment, invalid_template, invalid_event_type, invalid_webhook_url, invalid_expires_at, invalid_scope), reserved_domain, too_many_recipients, template_render_failed, template_rendered_empty (all 400); recipient_suppressed (422). Not found: not_found (404) plus the *_not_found family (attachment_not_found, contact_not_found, engagement_not_found, import_batch_not_found, template_not_found, starter_template_not_found); gone (410, past retention). Conflict/state: conflict (409, generic), precondition_failed (412, optimistic-concurrency validator is stale), the *_taken family — the requested identifier is already claimed — (agent_taken, domain_taken, alias_taken, all 409), address_in_trash (409), message_held (409), message_not_pending (409), message_not_yet_delivered (409, retry after the source outbound message is sent), not_in_trash (409), purge_in_progress (409, permanent delete already claimed), send_in_progress (409), webhook_disabled (409), webhook_cooldown (409), domain_not_registered (400), domain_has_agents (400), domain_not_verified (400 on create-agent, 403 on send), inbound_mx_missing (400). Capacity: limit_exceeded (402, plan quota — see LimitExceededDetails), rate_limited (429, request rate — see RateLimitedDetails), contact_limit_reached, template_limit_reached and webhook_limit_reached (400, fixed per-account caps). Idempotency: idempotency_in_flight (409, wait then retry the byte-identical request), idempotency_key_reuse (422, caller bug — do not retry as-is). Size: payload_too_large (413, request body), attachment_too_large (413, inline fetch over the cap — use download_url). Availability: not_implemented (501, feature not available on this deployment), events_log_disabled (501), limits_unavailable (503), inbound_mx_check_failed (503), auth_unavailable (503, an auth backend — e.g. a delegated-token verifier or the identity store — could not judge the credential; retry). Server/fallback: internal_error (5xx), method_not_allowed (405), unsupported_media_type (415), and the generic code error for any otherwise-unmapped status."
           type: string
           x-e2a-error-contracts:
             address_in_trash:
@@ -2259,6 +2259,11 @@ components:
               retryable: false
               statuses:
                 - 409
+            sending_paused:
+              family: auth
+              retryable: false
+              statuses:
+                - 403
             starter_template_not_found:
               family: not_found
               retryable: false
@@ -2317,6 +2322,7 @@ components:
                 - 400
           x-experimental-values:
             - blocked_by_policy
+            - sending_paused
         details:
           additionalProperties: true
           description: Optional structured context, polymorphic by code. Treat it as an open object keyed off code; unknown codes and fields must be preserved.
@@ -2891,6 +2897,8 @@ components:
             - submission.provider_rejected
             - submission.local_retries_exhausted
             - submission.cancelled
+            - submission.policy_budget_expired
+            - submission.sending_setup_expired
             - delivery.recipient_server_accepted
             - delivery.temporary_delay
             - delivery.permanent_bounce

cli/package.json

@@ -34,7 +34,7 @@
     "@e2a/sdk": "^5.7.0"
   },
   "devDependencies": {
-    "@types/node": "^26.3.0",
+    "@types/node": "^26.4.1",
     "@vitest/coverage-v8": "^4.1.11",
     "typescript": "^7.0.2",
     "vitest": "^4.1.10"

cmd/e2a/main.go

@@ -38,7 +38,6 @@ import (
 	"github.com/tokencanopy/e2a/internal/limits"
 	"github.com/tokencanopy/e2a/internal/oauth"
 	"github.com/tokencanopy/e2a/internal/outbound"
-	"github.com/tokencanopy/e2a/internal/outboundsend"
 	"github.com/tokencanopy/e2a/internal/relay"
 	"github.com/tokencanopy/e2a/internal/senderidentity"
 	"github.com/tokencanopy/e2a/internal/sendingpolicy"
@@ -122,6 +121,7 @@ func main() {
 	flag.IntVar(&spFlags.activeBillingContract, "active-billing-contract", -1, "verified active billing contract level")
 	flag.StringVar(&spFlags.rollbackBillingDigest, "rollback-billing-digest", "", "verified rollback billing image digest")
 	flag.IntVar(&spFlags.rollbackBillingContract, "rollback-billing-contract", -1, "verified rollback billing contract level")
+	flag.BoolVar(&spFlags.reconcile, "reconcile-legacy-sending-jobs", false, "stamp a sending operation reference onto every pending provider-submitting job enqueued without one (cancelling orphans whose source row is gone), print counts, then exit; nonzero unless every job was decided")
 	flag.BoolVar(&spFlags.capabilities, "print-capabilities", false, "print the machine-readable capability marker (contract level, policy source, operator commitments), then exit")
 	flag.StringVar(&spFlags.reason, "reason", "", "nonblank reason recorded in the audit row of a sending-protection mutation")
 	flag.Parse()
@@ -343,30 +343,32 @@ func main() {
 
 	// Outbound delivery is queue-first and at-least-once for GA. The accept-tx
 	// enqueues an outbound_send job in the same transaction as the message row;
-	// there is no submit-inline fallback.
+	// there is no submit-inline fallback. Every provider call passes through
+	// the sending-protection gate and the authorized submitter — see
+	// newOutboundSending, whose wiring test pins that composition.
 	rampStore := sendramp.NewStore(pool)
-	outboundRamp := agent.NewOutboundRampGate(
-		rampStore,
-		sendramp.NewSchedule(cfg.SendingRamp.StartDaily, cfg.SendingRamp.TargetDaily, cfg.SendingRamp.RampDays),
-		cfg.SendingRamp.Enabled,
-	)
-	if cfg.SendingRamp.Enabled {
-		log.Printf("Outbound sending ramp enabled: %d→%d recipients over %d qualified days", cfg.SendingRamp.StartDaily, cfg.SendingRamp.TargetDaily, cfg.SendingRamp.RampDays)
-	}
 	outboundSendStore := agent.NewOutboundSendStore(store, webhookOutbox, usageTracker)
 	store.SetScheduledSendFinalizer(outboundSendStore)
-	outboundJobs := outboundsend.NewJobs(
-		outboundSendStore,
-		agent.NewOutboundDeliverer(sender),
-		pool,
-		outboundRamp,
-	).WithMetrics(metrics).
+	outboundSending := newOutboundSending(outboundSendingDeps{
+		pool:         pool,
+		store:        outboundSendStore,
+		relay:        smtpRelay,
+		secrets:      spSecrets,
+		source:       spSource,
+		policy:       spPolicy,
+		sesConfigSet: cfg.DeliveryFeedback.SESConfigurationSet,
+		metrics:      metrics,
 		// Fire-time per-agent rate limit (60 submissions/min/agent sliding
 		// window, durable in Postgres): the cross-replica counterpart of the
 		// acceptance-time in-memory limiter, enforced immediately before
 		// provider submission so scheduled-send bursts can't exceed it.
-		WithRateGate(sendrate.NewStore(pool, time.Minute, 60))
+		rate: sendrate.NewStore(pool, time.Minute, 60),
+	})
+	outboundJobs := outboundSending.jobs
 	registrars = append(registrars, outboundJobs)
+	// Platform mail the API sends itself (public feedback) crosses the same
+	// seam with tokens from the same gate.
+	sendingGate, providerSubmitter := outboundSending.gate, outboundSending.submitter
 	registrars = append(registrars, sendramp.NewMaintenanceJobs(rampStore))
 	// Queue depth/age gauges: a 30s maintenance periodic sampling river_job
 	// per queue+state (docs/observability.md).
@@ -391,10 +393,17 @@ func main() {
 	// later via SetDeliverer — mirrors inbound's late-bound Processor. Gated on the
 	// same relay+public-URL config as the notifier itself; when unconfigured, no jobs
 	// register and the hold takes the plain path (no notification).
-	var notifyJobs *hitlnotify.Jobs
 	notifierEnabled := cfg.OutboundSMTP.FromDomain != "" && cfg.HTTP.PublicURL != ""
-	if notifierEnabled {
-		notifyJobs = hitlnotify.NewJobs(store)
+	notification := newNotificationJobs(notificationDeps{
+		store:          store,
+		pool:           pool,
+		gate:           sendingGate,
+		metrics:        metrics,
+		hitlEnabled:    notifierEnabled,
+		webhookEnabled: cfg.OutboundSMTP.FromDomain != "",
+	})
+	notifyJobs := notification.hitl
+	if notifyJobs != nil {
 		registrars = append(registrars, notifyJobs)
 	}
 
@@ -407,9 +416,8 @@ func main() {
 	// (generic dashboard copy instead of a link). When unconfigured, no jobs
 	// register and the sweep transitions state without notifications
 	// (pre-feature behavior).
-	var webhookNotifyJobs *webhooknotify.Jobs
-	if cfg.OutboundSMTP.FromDomain != "" {
-		webhookNotifyJobs = webhooknotify.NewJobs(store).WithMetrics(metrics)
+	webhookNotifyJobs := notification.webhook
+	if webhookNotifyJobs != nil {
 		registrars = append(registrars, webhookNotifyJobs)
 	}
 
@@ -699,7 +707,7 @@ func main() {
 		// unreachable in practice — kept as a defensive guard against future drift.
 		log.Printf("[hitl] notifier disabled: notification job pipeline not registered")
 	} else {
-		notifier := hitlnotify.New(store, smtpRelay, approvalSigner, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
+		notifier := hitlnotify.New(store, providerSubmitter, approvalSigner, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
 		// Late-bind the concrete Deliverer onto the registered NotifyWorker (which
 		// has been running since jobsClient.Start; jobs enqueued before this bind
 		// simply retry) and give the hold path its accept-tx enqueuer. The HTTP
@@ -720,7 +728,7 @@ func main() {
 	// a BYODKIM custom from-address domain is signed here or not at all.
 	// Fail-open — no stored key (self-host default) sends unsigned.
 	if webhookNotifyJobs != nil {
-		whNotifier := webhooknotify.New(store, smtpRelay, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
+		whNotifier := webhooknotify.New(store, providerSubmitter, cfg.OutboundSMTP.FromDomain, cfg.Notifications.FromAddress, cfg.Notifications.ReplyTo, cfg.HTTP.PublicURL).WithDKIM(store)
 		webhookNotifyJobs.SetDeliverer(whNotifier)
 		log.Printf("[webhook-notify] enabled (from=%s)", whNotifier.FromAddress())
 	} else {
@@ -835,6 +843,7 @@ func main() {
 	// The outbound accept-tx enqueuer is mandatory: DeliverOutbound always
 	// persists+enqueues and returns accepted before provider submission.
 	api.SetOutboundEnqueuer(outboundJobs)
+	outboundSending.armAPI(api)
 	// Slices 6 + 7: customer-facing events API needs the raw pool to
 	// query webhook_events and write webhook_subscriber_deliveries on
 	// replay. Kept as a separate setter so a future refactor can route