feat(outbound): enforce sending policy at fire time (B6)

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

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"
@@ -343,29 +342,28 @@ 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)
 	registrars = append(registrars, sendramp.NewMaintenanceJobs(rampStore))
 	// Queue depth/age gauges: a 30s maintenance periodic sampling river_job

cmd/e2a/outbound_wiring.go

@@ -0,0 +1,51 @@
+package main
+
+import (
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/tokencanopy/e2a/internal/agent"
+	"github.com/tokencanopy/e2a/internal/outbound"
+	"github.com/tokencanopy/e2a/internal/outboundsend"
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+)
+
+// outboundSendingDeps is everything the outbound composition root needs. It
+// is a struct rather than positional arguments so the wiring test can build
+// the production composition from synthetic inputs and inspect the result.
+type outboundSendingDeps struct {
+	pool         *pgxpool.Pool
+	store        outboundsend.Store
+	relay        *outbound.SMTPRelay
+	secrets      sendingpolicy.Secrets
+	source       sendingpolicy.PolicySource
+	policy       sendingpolicy.RuntimePolicy
+	sesConfigSet string
+	metrics      outboundsend.Metrics
+	rate         outboundsend.RateGate
+}
+
+// outboundSending is the composed outbound send path.
+type outboundSending struct {
+	gate      sendingpolicy.Gate
+	submitter *outbound.ProviderSubmitter
+	jobs      *outboundsend.Jobs
+}
+
+// newOutboundSending is the ONE composition root for provider-bound customer
+// mail. The gate is the deployment's policy authority; the submitter is the
+// only object that opens a socket to the provider and it refuses to do so
+// without a token from that gate; the jobs bundle prepares an operation at
+// enqueue and authorizes every worker execution through the same gate. No
+// raw sender and no direct ramp store reach the worker from here.
+func newOutboundSending(d outboundSendingDeps) outboundSending {
+	gate := sendingpolicy.NewGate(d.pool, d.secrets, d.source, d.policy)
+	submitter := outbound.NewProviderSubmitter(d.relay, gate)
+	// Delivery feedback: tag outbound with the SES configuration set so SES
+	// publishes delivery/bounce/complaint events. Empty = off.
+	submitter.SetSESConfigurationSet(d.sesConfigSet)
+	jobs := outboundsend.NewJobs(d.store, agent.NewOutboundDeliverer(submitter), d.pool).
+		WithGate(gate).
+		WithMetrics(d.metrics).
+		WithRateGate(d.rate)
+	return outboundSending{gate: gate, submitter: submitter, jobs: jobs}
+}

cmd/e2a/sending_policy_wiring_test.go

@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/riverqueue/river"
+
+	"github.com/tokencanopy/e2a/internal/config"
+	"github.com/tokencanopy/e2a/internal/outbound"
+	"github.com/tokencanopy/e2a/internal/sendingpolicy"
+	"github.com/tokencanopy/e2a/internal/testutil/testdb"
+)
+
+// TestSendingPolicyWiring builds the production outbound composition from
+// synthetic inputs and proves the registered send path holds the concrete
+// Gate and the authorized submitter. It exists so that a refactor that
+// reintroduced a raw sender or a direct ramp gate in the worker's path could
+// not pass CI: the only deliverer the composition root may produce is the one
+// over outbound.ProviderSubmitter, and the only admission authority is the
+// sendingpolicy module.
+func TestSendingPolicyWiring(t *testing.T) {
+	pool := testdb.TestDB(t)
+	relay := outbound.NewSMTPRelay(&config.OutboundSMTPConfig{Host: "relay.invalid", Port: 587, FromDomain: "test.e2a.dev"})
+
+	composed := newOutboundSending(outboundSendingDeps{
+		pool:         pool,
+		store:        nil, // the store is not exercised by construction
+		relay:        relay,
+		secrets:      sendingpolicy.Secrets{},
+		source:       sendingpolicy.PolicySourceConfig,
+		policy:       sendingpolicy.DisabledPolicy(),
+		sesConfigSet: "e2a-delivery-test",
+	})
+
+	if _, ok := composed.gate.(*sendingpolicy.Module); !ok {
+		t.Fatalf("gate is %T, want the concrete *sendingpolicy.Module", composed.gate)
+	}
+	if composed.submitter == nil {
+		t.Fatal("no authorized submitter composed")
+	}
+	if got := composed.submitter.SESConfigurationSet(); got != "e2a-delivery-test" {
+		t.Fatalf("submitter configuration set = %q, want the deployment's — delivery feedback must stay on", got)
+	}
+	if composed.jobs.Gate() != composed.gate {
+		t.Fatal("the jobs bundle does not hold the composed gate")
+	}
+	// The worker RegisterJobs registers is what runs in production; it, not
+	// the bundle, must carry the gate and the legacy resolver. Without the
+	// resolver every job in flight at cutover would fail closed.
+	// Register exactly as main does and inspect what River received — the
+	// constructor alone would not catch a RegisterJobs that bypassed it.
+	composed.jobs.RegisterJobs(river.NewWorkers())
+	worker := composed.jobs.RegisteredSendWorker()
+	if worker == nil {
+		t.Fatal("RegisterJobs registered no send worker")
+	}
+	if worker.Gate() != composed.gate {
+		t.Fatal("the registered send worker does not hold the composed gate")
+	}
+	if !worker.HasOperationResolver() {
+		t.Fatal("the registered send worker has no legacy operation resolver")
+	}
+	if composed.jobs.TerminalReconcileWorker() == nil {
+		t.Fatal("no terminal reconciler composed")
+	}
+	if got := fmt.Sprintf("%T", composed.jobs.Deliverer()); !strings.HasSuffix(got, "agent.outboundDeliverer") {
+		t.Fatalf("worker deliverer is %s, want the ProviderSubmitter-backed agent.outboundDeliverer", got)
+	}
+
+	// The composed gate is live: a config-source module answers policy reads
+	// against the real database, which is what the worker will do.
+	if _, err := composed.gate.LookupOperation(context.Background(), "op_wiring_probe"); err == nil {
+		t.Fatal("a never-prepared operation resolved")
+	}
+}

docs/api.md

@@ -85,7 +85,8 @@ stable field are beta, `x-experimental-values` on that field):
 the screening + review-hold event types (`email.flagged`, `email.blocked`,
 `email.review_requested`, `email.review_approved`, `email.review_rejected` —
 marked via `x-experimental-values` on the stable `type` field). The stable
-`error.code` vocabulary likewise marks only `blocked_by_policy` experimental.
+`error.code` vocabulary likewise marks only `blocked_by_policy` and
+`sending_paused` experimental.
 See [events.md](events.md).
 
 The exact operation-level list is repeated with methods and paths in
@@ -313,6 +314,7 @@ retryable ones (the per-row retry notes in the table below are authoritative).
 | `unauthorized` | 401 | Missing or invalid credentials (REST and the WebSocket handshake). |
 | `forbidden` | 403 | Authenticated but not allowed (key scope, cross-tenant access). |
 | `blocked_by_policy` | 403 | **Experimental.** The outbound message was blocked by the agent's outbound policy gate. |
+| `sending_paused` | 403 | **Experimental.** Outbound sending is paused for the account by the platform abuse controls. Nothing was queued; queued mail is held until an operator resumes. |
 | **Validation** | | |
 | `invalid_request` | 400 / 422 | The canonical input-validation code — malformed (400) or semantically invalid (422). `error.details` carries the per-field list. |
 | `invalid_cursor` | 400 | Bad pagination cursor — drop it and re-fetch from the start. |
@@ -465,7 +467,8 @@ every `/v1` operation not listed here is covered by the GA freeze.
   `x-experimental-values` listing exactly those values — the field itself
   stays stable, the listed values (and their payloads) may still change, and
   every unlisted value is stable. The stable `ErrorBody.code` discriminator
-  similarly marks only `blocked_by_policy` experimental. Anything not marked
+  similarly marks only `blocked_by_policy` and `sending_paused` experimental.
+  Anything not marked
   beta or experimental is stable surface. One deliberate schema-level use of
   the beta marker under a **stable** operation: the account export's interior
   record schemas (`GET /v1/account/export`) are beta-marked because they are
@@ -859,6 +862,8 @@ retryability; clients must not reinterpret those fields independently:
 | `submission.provider_rejected` | `submission` | `failed` | false |
 | `submission.local_retries_exhausted` | `submission` | `failed` | true |
 | `submission.cancelled` | `submission` | `failed` | false |
+| `submission.policy_budget_expired` | `submission` | `failed` | true |
+| `submission.sending_setup_expired` | `submission` | `failed` | true |
 | `delivery.recipient_server_accepted` | `delivery` | `delivered` | false |
 | `delivery.temporary_delay` | `delivery` | `deferred` | true |
 | `delivery.permanent_bounce` | `delivery` | `bounced` | false |